feat: Reduce Cytiva Biacore T200 Control memory usage with cycle streaming (#1164)

## Summary
Implements cycle streaming for Cytiva Biacore T200 Control parser to
reduce memory usage by 19.6%.

## Changes
- Implement `decode_data_streaming()` generator that yields cycles one
at a time
- Process cycles individually instead of concatenating all DataFrames at
once
- Maintain backward compatibility with `decode_data()` wrapper function

## Memory Impact
- **Before:** 134.24 MB peak (84.6x amplification on 1.59 MB file)
- **After:** 107.87 MB peak (68.0x amplification)  
- **Reduction:** 26.37 MB saved (19.6% improvement)

Measurements taken with tracemalloc on `Fig.2c_NC_prongs
hybridization.blr` (1.59 MB test file), consistent across 3 runs.

## Testing
- All existing tests pass with no changes to golden files
- Type safety verified with mypy
- Backward compatible - `decode_data()` wrapper maintains existing API

## Context
This addresses memory amplification concerns identified in the parser
audit. The T200 Control parser showed 133x average amplification with
memory_profiler (agent-reported). Accurate measurements with tracemalloc
show 84.6x on main, reduced to 68.0x with streaming.

Related to memory optimization work on other Cytiva Biacore parsers.

Author: nathan-stender

src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_decoder.py

@@ -362,19 +362,30 @@ def get_r_point_data(r_point_data: str) -> list[dict[str, Any]]:
     return r_data_list
 
 
-def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
+def decode_data_streaming(named_file_contents: NamedFileContents) -> Any:  # Generator
     """
-    Decodes the proprietary file into a structured dict
+    Decodes the proprietary file into a structured dict, streaming cycle data.
     :param named_file_contents: The named file contents containing the input file details
-    :return: structured dictionary of decoded data
+    :return: Iterator yielding metadata first, then cycle data dicts
     """
     intermediate_json: dict[str, Any] = {}
     content = ole.OleFileIO(named_file_contents.get_bytes_stream())
     streams = content.listdir()
-    sensorgram_df_list = []
-    report_point_data = {}
+
+    # Group streams by cycle number
+    cycle_streams: dict[int, list[Any]] = {}
+    report_point_df = None
 
     for stream in streams:
+        # Check if this is a cycle-specific stream first
+        if cycle_match := cycle_pattern.search(stream[0]):
+            cycle_number = int(cycle_match.group(1))
+            if cycle_number not in cycle_streams:
+                cycle_streams[cycle_number] = []
+            cycle_streams[cycle_number].append(stream)
+            continue
+
+        # Process metadata streams
         stream_content = content.openstream(stream).read()
         if stream == ["Environment"]:
             intermediate_json["system_information"] = extract_stream_data(
@@ -384,7 +395,6 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
             intermediate_json["chip"] = extract_stream_data(
                 stream_content.decode("utf-8")
             )
-
         elif stream == ["AppData", "Dip"]:
             intermediate_json["dip"] = get_dip_data(stream_content.decode("utf-8"))
         elif stream == ["AppData", "ApplicationTemplate"]:
@@ -396,76 +406,11 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
                 intermediate_json["sample_data"] = sample_data
         elif stream == ["RPoint Table"]:
             r_point_data = get_r_point_data(stream_content.decode("utf-8"))
-            r_point_dataframe = pd.DataFrame(r_point_data)
-            report_point_data = {
-                group["Cycle"].iloc[0]: group
-                for _, group in r_point_dataframe.groupby("Cycle")
-            }
+            report_point_df = pd.DataFrame(r_point_data)
 
-        if not (cycle_match := cycle_pattern.search(stream[0])):
-            continue
-        cycle_number = int(cycle_match.group(1))
-
-        if (curve_match := curve_pattern.search(str(stream))) and (
-            window_match := window_pattern.search(str(stream))
-        ):
-            curve_number = curve_match.group(1)
-            window_number = window_match.group(1)
-
-            # get the flow cell number
-            if "Labels" in str(stream):
-                label_list = []
-                if len(stream_content) != 0:
-                    decoded_stream_content = stream_content.decode("utf-8")
-                    for line in decoded_stream_content.strip().split("\n"):
-                        label_list.append(line)
-                        if "Fc" in line:
-                            flow_cell = label_list[0].split("=")[1]
-
-            # gets the xy data as data frame
-            elif "XYData" in str(stream):
-                xy_data_list = list(
-                    struct.unpack("f" * (len(stream_content) // 4), stream_content)
-                )
-                indexed_xy_data_list = xy_data_list[3:]
-                length_of_xydata = len(xy_data_list[3:])
-                xy_result_list = indexed_xy_data_list[int(length_of_xydata / 2) :]
-                time_values = indexed_xy_data_list[: int(length_of_xydata / 2)]
-                length_of_xy = len(xy_result_list)
-
-                xy_data_frame = pd.DataFrame(
-                    {
-                        "Flow Cell Number": [flow_cell] * length_of_xy,
-                        "Cycle Number": [cycle_number] * length_of_xy,
-                        "Curve Number": [curve_number] * length_of_xy,
-                        "Window Number": [window_number] * length_of_xy,
-                        "Sensorgram (RU)": xy_result_list,
-                        "Time (s)": time_values,
-                    }
-                )
-
-                sensorgram_df_list.append(xy_data_frame)
-
-            # gets the segment data
-            elif "Segment" in str(stream):
-                segment_data_list = list(
-                    struct.unpack("f" * (len(stream_content) // 4), stream_content)
-                )
-                length_of_segment = len(segment_data_list[11:])
-                segment_data_frame = pd.DataFrame(
-                    {
-                        "Flow Cell Number": [flow_cell] * length_of_segment,
-                        "Cycle Number": [cycle_number] * length_of_segment,
-                        "Curve Number": [curve_number] * length_of_segment,
-                        "Window Number": [window_number] * length_of_segment,
-                        "Sensorgram (RU)": segment_data_list[11:],
-                    }
-                )
-                sensorgram_df_list.append(segment_data_frame)
-
-    combined_sensorgram_df = pd.concat(sensorgram_df_list, ignore_index=True)
-
-    sensorgram_data = {}
+    # Get total cycles and data collection rate
+    total_cycles = max(cycle_streams.keys()) if cycle_streams else 0
+    intermediate_json["total_cycles"] = total_cycles
 
     data_collection_rate = None
     if intermediate_json.get("application_template_details", {}).get(
@@ -475,10 +420,82 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
             "DataCollectionRate"
         ]["value"]
 
-    grouped_df = combined_sensorgram_df.groupby("Cycle Number")
-    for cycle_num, cycle_group in grouped_df:
-        group = cycle_group.copy()
+    # Yield metadata first
+    yield intermediate_json
+
+    # Process cycles one at a time
+    for cycle_num in sorted(cycle_streams.keys()):
+        sensorgram_df_list = []
+        flow_cell = "1"  # default
 
+        for stream in cycle_streams[cycle_num]:
+            stream_content = content.openstream(stream).read()
+
+            if (curve_match := curve_pattern.search(str(stream))) and (
+                window_match := window_pattern.search(str(stream))
+            ):
+                curve_number = curve_match.group(1)
+                window_number = window_match.group(1)
+
+                # get the flow cell number
+                if "Labels" in str(stream):
+                    label_list = []
+                    if len(stream_content) != 0:
+                        decoded_stream_content = stream_content.decode("utf-8")
+                        for line in decoded_stream_content.strip().split("\n"):
+                            label_list.append(line)
+                            if "Fc" in line:
+                                flow_cell = label_list[0].split("=")[1]
+
+                # gets the xy data as data frame
+                elif "XYData" in str(stream):
+                    xy_data_list = list(
+                        struct.unpack("f" * (len(stream_content) // 4), stream_content)
+                    )
+                    indexed_xy_data_list = xy_data_list[3:]
+                    length_of_xydata = len(xy_data_list[3:])
+                    xy_result_list = indexed_xy_data_list[int(length_of_xydata / 2) :]
+                    time_values = indexed_xy_data_list[: int(length_of_xydata / 2)]
+                    length_of_xy = len(xy_result_list)
+
+                    xy_data_frame = pd.DataFrame(
+                        {
+                            "Flow Cell Number": [flow_cell] * length_of_xy,
+                            "Cycle Number": [cycle_num] * length_of_xy,
+                            "Curve Number": [curve_number] * length_of_xy,
+                            "Window Number": [window_number] * length_of_xy,
+                            "Sensorgram (RU)": xy_result_list,
+                            "Time (s)": time_values,
+                        }
+                    )
+
+                    sensorgram_df_list.append(xy_data_frame)
+
+                # gets the segment data
+                elif "Segment" in str(stream):
+                    segment_data_list = list(
+                        struct.unpack("f" * (len(stream_content) // 4), stream_content)
+                    )
+                    length_of_segment = len(segment_data_list[11:])
+                    segment_data_frame = pd.DataFrame(
+                        {
+                            "Flow Cell Number": [flow_cell] * length_of_segment,
+                            "Cycle Number": [cycle_num] * length_of_segment,
+                            "Curve Number": [curve_number] * length_of_segment,
+                            "Window Number": [window_number] * length_of_segment,
+                            "Sensorgram (RU)": segment_data_list[11:],
+                        }
+                    )
+                    sensorgram_df_list.append(segment_data_frame)
+
+        # Process this cycle's sensorgram data
+        if not sensorgram_df_list:
+            continue
+
+        combined_sensorgram_df = pd.concat(sensorgram_df_list, ignore_index=True)
+        group = combined_sensorgram_df.copy()
+
+        # Time normalization logic
         if "Time (s)" in group.columns:
             max_flow_cell = group["Flow Cell Number"].max()
             max_flow_cell_mask = group["Flow Cell Number"] == max_flow_cell
@@ -495,9 +512,9 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
                     )
                     reference_times = max_flow_cell_data["Time (s)"].values
 
-                    for flow_cell in unique_flow_cells:
-                        if flow_cell != max_flow_cell:
-                            flow_cell_mask = group["Flow Cell Number"] == flow_cell
+                    for fc in unique_flow_cells:
+                        if fc != max_flow_cell:
+                            flow_cell_mask = group["Flow Cell Number"] == fc
                             flow_cell_indices = group[flow_cell_mask].index
 
                             num_points = len(flow_cell_indices)
@@ -516,7 +533,7 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
         else:
             group["Time (s)"] = group.groupby("Flow Cell Number").cumcount() + 1
 
-        sensorgram_data[str(cycle_num)] = group[
+        sensorgram_data = group[
             [
                 "Flow Cell Number",
                 "Cycle Number",
@@ -527,16 +544,33 @@ def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
             ]
         ]
 
-    # In some cases the "RPoint Table" stream may not have data for the first cycle,
-    # but we get streams with sensorgram data (XYData or Segment streams) for all the cycles.
-    intermediate_json["cycle_data"] = [
-        {
-            "cycle_number": cycle,
-            "report_point_data": report_point_data.get(cycle),
-            "sensorgram_data": sensorgram_df,
+        # Get report point data for this cycle
+        report_point_data = None
+        if report_point_df is not None:
+            cycle_report = report_point_df[report_point_df["Cycle"] == str(cycle_num)]
+            if not cycle_report.empty:
+                report_point_data = cycle_report
+
+        # Yield this cycle's data
+        yield {
+            "cycle_number": str(cycle_num),
+            "report_point_data": report_point_data,
+            "sensorgram_data": sensorgram_data,
         }
-        for cycle, sensorgram_df in sensorgram_data.items()
-    ]
-    intermediate_json["total_cycles"] = cycle_number
+
+
+def decode_data(named_file_contents: NamedFileContents) -> dict[str, Any]:
+    """
+    Decodes the proprietary file into a structured dict
+    :param named_file_contents: The named file contents containing the input file details
+    :return: structured dictionary of decoded data
+    """
+    # Use streaming decoder but collect all data
+    stream_gen = decode_data_streaming(named_file_contents)
+    intermediate_json: dict[str, Any] = next(stream_gen)  # Get metadata
+
+    # Collect all cycle data
+    cycle_data = list(stream_gen)
+    intermediate_json["cycle_data"] = cycle_data
 
     return intermediate_json