Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -48,13 +48,16 @@
from allotropy.parsers.utils.uuids import random_uuid_str
from allotropy.parsers.utils.values import quantity_or_none, try_float_or_none

# Precompiled patterns for faster flow cell id normalization
_FLOWCELL_ID_RE = re.compile(r"\d+")


def _get_sensorgram_datacube(
sensorgram_df: pd.DataFrame, *, cycle: int, flow_cell: str
) -> DataCube:
# Extract all sensorgram data points
time_vals = sensorgram_df["Time (s)"].astype(float).to_list()
resp_vals = sensorgram_df["Sensorgram (RU)"].astype(float).to_list()
time_vals = sensorgram_df["Time (s)"].to_numpy(copy=False)
resp_vals = sensorgram_df["Sensorgram (RU)"].to_numpy(copy=False)
return DataCube(
label=f"Cycle{cycle}_FlowCell{flow_cell}",
structure_dimensions=[
Expand All @@ -63,8 +66,8 @@ def _get_sensorgram_datacube(
structure_measures=[
DataCubeComponent(FieldComponentDatatype.double, "resonance", "RU")
],
dimensions=[time_vals],
measures=[resp_vals],
dimensions=[time_vals.tolist()],
Comment thread
joshua-benchling marked this conversation as resolved.
measures=[resp_vals.tolist()],
)


Expand Down Expand Up @@ -261,10 +264,22 @@ def _normalize_flow_cell_id(value: Any) -> str:
# Don't normalize reference-subtracted flow cell IDs (e.g., "2-1", "3-1", "4-1")
if "-" in s:
return s
# Only normalize pure numeric flow cell IDs
m = re.match(r"\d+", s)
# Fast path for pure digits
if s.isdigit():
return s
# Only normalize pure numeric prefix if present
m = _FLOWCELL_ID_RE.match(s)
return m.group(0) if m else s

# Precompute kinetic analysis mapping from identifiers to flow cell ids (e.g., EvaluationItem2 -> "2")
kinetic_map: dict[str, KineticResult] | None = None
if data.kinetic_analysis and data.kinetic_analysis.results_by_identifier:
kinetic_map = {}
for key, result in data.kinetic_analysis.results_by_identifier.items():
m = _FLOWCELL_ID_RE.search(str(key))
if m:
kinetic_map[m.group(0)] = result

# Process all flow cells (including reference-subtracted ones like "2-1", "3-1", "4-1")
for flow_cell, df_fc in sensorgram_df.groupby("Flow Cell Number"):
fc_id = _normalize_flow_cell_id(flow_cell)
Expand Down Expand Up @@ -357,33 +372,8 @@ def _normalize_flow_cell_id(value: Any) -> str:
}
break

# Extract kinetic analysis data for this specific flow cell
# Match EvaluationItem identifier to flow cell identifier
combined_kinetic_data = None
if data.kinetic_analysis and data.kinetic_analysis.results_by_identifier:
# Try to find the specific EvaluationItem for this flow cell
# Flow cell IDs are typically "1", "2", "3", "4"
# EvaluationItem IDs are typically "EvaluationItem1", "EvaluationItem2", etc.
matching_eval_item = None

# First, try direct mapping: flow cell "1" -> "EvaluationItem1"
eval_item_key = f"EvaluationItem{fc_id}"
if eval_item_key in data.kinetic_analysis.results_by_identifier:
matching_eval_item = eval_item_key
else:
# If direct mapping fails, look for any EvaluationItem that might correspond to this flow cell
# This could be enhanced with more sophisticated matching logic if needed
for eval_key in data.kinetic_analysis.results_by_identifier.keys():
if fc_id in eval_key or eval_key.endswith(fc_id):
matching_eval_item = eval_key
break

# Use only the matching EvaluationItem data for this flow cell
if matching_eval_item:
result = data.kinetic_analysis.results_by_identifier[matching_eval_item]
combined_kinetic_data = result

kinetic_data = combined_kinetic_data
# Extract kinetic analysis data using precomputed mapping (if present)
kinetic_data = kinetic_map.get(fc_id) if kinetic_map is not None else None
Comment thread
felipenarv marked this conversation as resolved.

measurements.append(
Measurement(
Expand Down Expand Up @@ -458,37 +448,37 @@ def _normalize_flow_cell_id(value: Any) -> str:


def create_measurement_groups(data: Data) -> list[MeasurementGroup]:
sys = data.system_information
system_info = data.system_information
# Prefer application template timestamp if present in run metadata
if data.run_metadata.timestamp and not sys.measurement_time:
sys = SystemInformation(
application_name=sys.application_name,
application_version=sys.application_version,
user_name=sys.user_name,
system_controller_identifier=sys.system_controller_identifier,
os_type=sys.os_type,
os_version=sys.os_version,
if data.run_metadata.timestamp and not system_info.measurement_time:
system_info = SystemInformation(
application_name=system_info.application_name,
application_version=system_info.application_version,
user_name=system_info.user_name,
system_controller_identifier=system_info.system_controller_identifier,
os_type=system_info.os_type,
os_version=system_info.os_version,
measurement_time=data.run_metadata.timestamp,
unread_application_properties=sys.unread_application_properties,
measurement_aggregate_fields=sys.measurement_aggregate_fields,
unread_application_properties=system_info.unread_application_properties,
measurement_aggregate_fields=system_info.measurement_aggregate_fields,
)
# As a final fallback, look directly in application_template_details.properties
if not sys.measurement_time and data.application_template_details:
if not system_info.measurement_time and data.application_template_details:
props = data.application_template_details.get("properties", {})
ts = props.get("Timestamp")
if ts:
sys = SystemInformation(
application_name=sys.application_name,
application_version=sys.application_version,
user_name=sys.user_name,
system_controller_identifier=sys.system_controller_identifier,
os_type=sys.os_type,
os_version=sys.os_version,
system_info = SystemInformation(
application_name=system_info.application_name,
application_version=system_info.application_version,
user_name=system_info.user_name,
system_controller_identifier=system_info.system_controller_identifier,
os_type=system_info.os_type,
os_version=system_info.os_version,
measurement_time=ts,
unread_application_properties=sys.unread_application_properties,
measurement_aggregate_fields=sys.measurement_aggregate_fields,
unread_application_properties=system_info.unread_application_properties,
measurement_aggregate_fields=system_info.measurement_aggregate_fields,
)
if not sys.measurement_time:
if not system_info.measurement_time:
msg = "Missing measurement time. Expected application_template_details.properties.Timestamp."
raise AllotropeParsingError(msg)
groups: list[MeasurementGroup] = []
Expand All @@ -499,7 +489,7 @@ def create_measurement_groups(data: Data) -> list[MeasurementGroup]:
"data collection rate": quantity_or_none(
TQuantityValueHertz, data.run_metadata.data_collection_rate
),
**sys.measurement_aggregate_fields,
**system_info.measurement_aggregate_fields,
}
# Add aggregate-level experimental data identifier for convenience (first measurement's FC)
if measurements:
Expand All @@ -519,7 +509,7 @@ def create_measurement_groups(data: Data) -> list[MeasurementGroup]:

groups.append(
MeasurementGroup(
measurement_time=sys.measurement_time,
measurement_time=system_info.measurement_time,
Comment thread
felipenarv marked this conversation as resolved.
measurements=measurements,
experiment_type=None,
analytical_method_identifier=None,
Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
from __future__ import annotations

import datetime as _dt
import io
import re
import struct
from typing import Any

import numpy as np
from numpy.typing import NDArray
import olefile as ole
import pandas as pd
import xmltodict
Expand All @@ -20,8 +22,6 @@

def _convert_datetime(days_str: str) -> str:
# Biacore epoch: 1899-12-30 UTC
import datetime as _dt

days = float(days_str)
start = _dt.datetime(year=1899, month=12, day=30, tzinfo=_dt.timezone.utc)
return (start + _dt.timedelta(days=days)).isoformat()
Expand Down Expand Up @@ -234,8 +234,6 @@ def _extract_kinetic_analysis(

# Use the full EvaluationItem identifier as the key
if "EvaluationItem" in path_str:
import re

match = re.search(r"(EvaluationItem\d+)", path_str)
if match:
flow_cell_id = match.group(1) # e.g., "EvaluationItem2"
Expand Down Expand Up @@ -377,13 +375,20 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
with ole.OleFileIO(named_file_contents.get_bytes_stream()) as content:
streams = content.listdir()

sensorgram_df_list: list[pd.DataFrame] = []
# Accumulate arrays for a single DataFrame build at the end
fc_list: list[NDArray[np.object_]] = []
cycle_list: list[NDArray[np.integer[Any]]] = []
curve_list: list[NDArray[np.object_]] = []
window_list: list[NDArray[np.object_]] = []
values_list: list[NDArray[np.floating[Any]]] = []
times_list: list[NDArray[np.floating[Any]]] = []
report_point_by_cycle: dict[str, pd.DataFrame] = {}
dip_data: dict[str, Any] = {}
kinetic_analysis: dict[str, Any] = {}
sample_data: Any = None

flow_cell = None
total_cycles_detected = 0

for stream in streams:
path_str = "/".join(stream)
Expand Down Expand Up @@ -427,10 +432,8 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
continue
if stream == ["RPoint Table"]:
data = content.openstream(stream).read()
lines = data.decode("utf-8").strip().split("\n")
header = lines[0].split("\t")
rows = [line.split("\t") for line in lines[1:] if line]
df = pd.DataFrame([dict(zip(header, r, strict=True)) for r in rows])

df = pd.read_csv(io.BytesIO(data), sep="\t")
report_point_by_cycle = {
grp["Cycle"].iloc[0]: grp for _, grp in df.groupby("Cycle")
}
Expand Down Expand Up @@ -501,49 +504,53 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
if "XYData" in path_str:
# Read all data from the file
raw = content.openstream(stream).read()
xy = list(struct.unpack("f" * (len(raw) // 4), raw))
indexed = xy[3:]
half = int(len(indexed) / 2)
arr = np.frombuffer(raw, dtype="<f4")
Comment thread
felipenarv marked this conversation as resolved.
indexed = arr[3:]
half = indexed.size // 2
values = indexed[half:]
times = indexed[:half]
length = len(values)
sensorgram_df_list.append(
pd.DataFrame(
{
"Flow Cell Number": [flow_cell or 1] * length,
"Cycle Number": [cycle_number] * length,
"Curve Number": [curve_number] * length,
"Window Number": [window_number] * length,
"Sensorgram (RU)": values,
"Time (s)": times,
}
)
)
length = values.size
fc_list.append(np.full(length, flow_cell or 1, dtype=object))
cycle_list.append(np.full(length, cycle_number, dtype=np.int64))
curve_list.append(np.full(length, curve_number, dtype=object))
window_list.append(np.full(length, window_number, dtype=object))
values_list.append(values)
times_list.append(times)
total_cycles_detected += 1
Comment thread
felipenarv marked this conversation as resolved.
continue

if "Segment" in path_str:
# Read all data from the file
raw = content.openstream(stream).read()
seg = list(struct.unpack("f" * (len(raw) // 4), raw))
seg_vals = seg[11:]
length = len(seg_vals)
sensorgram_df_list.append(
pd.DataFrame(
{
"Flow Cell Number": [flow_cell or 1] * length,
"Cycle Number": [cycle_number] * length,
"Curve Number": [curve_number] * length,
"Window Number": [window_number] * length,
"Sensorgram (RU)": seg_vals,
}
)
)

combined_df = (
pd.concat(sensorgram_df_list, ignore_index=True)
if sensorgram_df_list
else pd.DataFrame()
)
seg_arr = np.frombuffer(raw, dtype="<f4")
seg_vals = seg_arr[11:]
length = seg_vals.size
fc_list.append(np.full(length, flow_cell or 1, dtype=object))
cycle_list.append(np.full(length, cycle_number, dtype=np.int64))
curve_list.append(np.full(length, curve_number, dtype=object))
window_list.append(np.full(length, window_number, dtype=object))
values_list.append(seg_vals)
times_list.append(np.full(length, np.nan, dtype=np.float64))
# no time aggregation; removed

if values_list:
combined_df = pd.DataFrame(
{
"Flow Cell Number": np.concatenate(fc_list, axis=0),
"Cycle Number": np.concatenate(cycle_list, axis=0),
"Curve Number": np.concatenate(curve_list, axis=0),
"Window Number": np.concatenate(window_list, axis=0),
"Sensorgram (RU)": np.concatenate(values_list, axis=0),
"Time (s)": np.concatenate(times_list, axis=0),
}
)
# Use categorical dtype for string-like columns to save memory and speed up grouping
combined_df["Curve Number"] = combined_df["Curve Number"].astype("category")
Comment thread
felipenarv marked this conversation as resolved.
combined_df["Window Number"] = combined_df["Window Number"].astype(
"category"
)
else:
combined_df = pd.DataFrame()
Comment thread
felipenarv marked this conversation as resolved.

if not combined_df.empty:
# Normalize time per flow cell using reference as in control
Expand All @@ -564,23 +571,17 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
g = group.copy()
if "Time (s)" in g.columns:
max_fc = g["Flow Cell Number"].max()
mask = g["Flow Cell Number"] == max_fc
ref_times = g.loc[mask, "Time (s)"]
ref_mask = g["Flow Cell Number"] == max_fc
ref_times = g.loc[ref_mask, "Time (s)"]
if pd.isna(ref_times).any():
g["Time (s)"] = g.groupby("Flow Cell Number").cumcount() + 1
else:
unique_fc = g["Flow Cell Number"].unique()
if len(unique_fc) > 1:
ref = g[mask].reset_index(drop=True)["Time (s)"].values
for fc in unique_fc:
if fc == max_fc:
continue
fc_mask = g["Flow Cell Number"] == fc
idx = g[fc_mask].index
if len(ref) > 0:
g.loc[idx, "Time (s)"] = ref[
np.arange(len(idx)) % len(ref)
]
elif g["Flow Cell Number"].nunique() > 1:
ref = ref_times.reset_index(drop=True).to_numpy()
nonref_mask = ~ref_mask
if ref.size > 0 and nonref_mask.any():
pos = g.groupby("Flow Cell Number").cumcount()
pos_nonref = pos[nonref_mask].to_numpy()
g.loc[nonref_mask, "Time (s)"] = ref[pos_nonref % ref.size]
elif dcr is not None:
g["Time (s)"] = g.groupby("Flow Cell Number").cumcount() * (1 / dcr)
else:
Expand Down
Loading