perf: Cytiva Biacore T200 Evaluation Module - Added optimizations on decoding logic (#1109)

Author: felipenarv

src/allotropy/parsers/cytiva_biacore_t200_evaluation/cytiva_biacore_t200_evaluation_data_creator.py

@@ -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=[
@@ -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()],
+        measures=[resp_vals.tolist()],
     )
 
 
@@ -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)
@@ -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
 
         measurements.append(
             Measurement(
@@ -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] = []
@@ -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:
@@ -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,
                 measurements=measurements,
                 experiment_type=None,
                 analytical_method_identifier=None,

src/allotropy/parsers/cytiva_biacore_t200_evaluation/cytiva_biacore_t200_evaluation_decoder.py

@@ -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
@@ -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()
@@ -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"
@@ -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)
@@ -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")
                 }
@@ -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")
+                    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
                     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")
+            combined_df["Window Number"] = combined_df["Window Number"].astype(
+                "category"
+            )
+        else:
+            combined_df = pd.DataFrame()
 
         if not combined_df.empty:
             # Normalize time per flow cell using reference as in control
@@ -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: