feat: Migrate ctl_immunospot to use SeriesData.get_unread (#1076)

Co-authored-by: Nathan Stender <nathan.stender@benchling.com>

Author: oscar-castro-bl

src/allotropy/allotrope/converter.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import asdict, field, fields, is_dataclass, make_dataclass, MISSING
 from enum import Enum
+import keyword
 from types import GenericAlias, UnionType
 from typing import (
     Any,
@@ -164,6 +165,8 @@ def add_custom_information_document(
 
 def _convert_model_key_to_dict_key(key: str) -> str:
     key = SPECIAL_KEYS.get(key, key)
+    if key.startswith("_KW"):
+        key = key[3:]
     if key.startswith("___") and key[3].isdigit():
         key = key[3:]
     for dict_val, model_val in DICT_KEY_TO_MODEL_KEY_REPLACEMENTS.items():
@@ -173,6 +176,8 @@ def _convert_model_key_to_dict_key(key: str) -> str:
 
 def _convert_dict_to_model_key(key: str) -> str:
     key = SPECIAL_KEYS_INVERSE.get(key, key)
+    if keyword.iskeyword(key):
+        key = f"_KW{key}"
     if key[0].isdigit():
         key = f"___{key}"
     for dict_val, model_val in DICT_KEY_TO_MODEL_KEY_REPLACEMENTS.items():

src/allotropy/parsers/ctl_immunospot/ctl_immunospot_reader.py

@@ -202,8 +202,6 @@ def _read_plate_data_7_0_38(self, reader: CsvReader) -> list[pd.DataFrame]:
         return plates
 
     def _read_header(self, reader: CsvReader) -> SeriesData:
-        lines = [line.strip() for line in reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE)]
-
         def fix_line(line: str) -> str:
             # Add missing key for file path line.
             if line.endswith(".txt") or line.endswith(".xls"):
@@ -213,13 +211,47 @@ def fix_line(line: str) -> str:
                 line = f"Review Date: {date_str}"
             return line.strip()
 
-        lines = [fix_line(line) for raw_line in lines for line in raw_line.split(";")]
+        def split_multi_key_line(line: str) -> list[str]:
+            """Split lines that contain multiple key-value pairs into separate lines."""
+            line = line.strip(' "')  # Clean up quotes and whitespace
+
+            # Case 1: Parenthetical format: (Auto Areas: Estimated, Manual Areas: Normalized)
+            if line.startswith("(") and line.endswith(")") and "," in line:
+                return [pair.strip() for pair in line[1:-1].split(",")]
+
+            # Case 2: Tab-separated format: Assay: 1\t\t\t\t\t\tEdge Compensation Level: 1.0
+            if "\t" in line and line.count(":") > 1:
+                return [
+                    part.strip()
+                    for part in line.split("\t")
+                    if part.strip() and ":" in part
+                ]
+
+            return [line]
+
+        def process_lines(raw_lines: list[str]) -> list[str]:
+            """Process a list of raw lines and return split/fixed lines."""
+            processed = []
+            for raw_line in raw_lines:
+                for line in raw_line.split(";"):
+                    fixed_line = fix_line(line)
+                    # Only process lines that have colons (key: value format)
+                    if ":" in fixed_line:
+                        processed.extend(split_multi_key_line(fixed_line))
+            return processed
+
+        # Process initial header lines
+        lines = [line.strip() for line in reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE)]
+        all_lines = process_lines(lines)
+
+        # Process additional header lines if they exist
         reader.drop_empty(EMPTY_STR_OR_CSV_LINE)
         if ":" in (reader.get() or ""):
-            lines.extend(list(reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE)))
+            additional_lines = list(reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE))
+            all_lines.extend(process_lines(additional_lines))
 
         df = read_csv(
-            StringIO("\n".join(lines)),
+            StringIO("\n".join(all_lines)),
             sep=r"^([^:]+):\s+",
             header=None,
             engine="python",

src/allotropy/parsers/ctl_immunospot/ctl_immunospot_structure.py

@@ -33,8 +33,18 @@ def _create_measurement(
     well_plate_identifier: str,
     plate_data: dict[str, pd.DataFrame],
     histograms: dict[str, tuple[list[float], list[float]]],
+    header_data: dict[str, float | str | None],
 ) -> Measurement:
     location_identifier = f"{well_row}{well_col}"
+    data_processing_document = {
+        "Min. SpotSize": header_data.pop("Min. SpotSize", None),
+        "Max. SpotSize": header_data.pop("Max. SpotSize", None),
+        "Spot Separation": header_data.pop("Spot Separation", None),
+    }
+    has_data = any(
+        value is not None and str(value).strip() != ""
+        for value in data_processing_document.values()
+    )
     return Measurement(
         type_=MeasurementType.OPTICAL_IMAGING,
         device_type=constants.DEVICE_TYPE,
@@ -45,6 +55,7 @@ def _create_measurement(
         detection_type=constants.DETECTION_TYPE,
         processed_data=ProcessedData(
             identifier=random_uuid_str(),
+            data_processing_document=data_processing_document if has_data else None,
             features=[
                 ImageFeature(
                     identifier=random_uuid_str(),
@@ -77,6 +88,7 @@ def _create_measurement(
         ]
         if histograms and location_identifier in histograms
         else None,
+        device_control_custom_info=header_data,
     )
 
 
@@ -94,16 +106,28 @@ def create_measurement_groups(
         round_to_nearest_well_count(first_plate.size),
         f"Unable to determine valid plate count from dataframe of size: {first_plate.size}",
     )
+    measurement_time = header[str, ("Counted", "Review Date")]
+    analyst = header[str, "Authenticated user"]
+    custom_info = {
+        "Assay": header.get(str, "Assay"),
+    }
+    header_data = header.get_unread()
     return [
         MeasurementGroup(
             plate_well_count=plate_well_count,
-            measurement_time=header[str, ("Counted", "Review Date")],
-            analyst=header[str, "Authenticated user"],
+            measurement_time=measurement_time,
+            analyst=analyst,
             measurements=[
                 _create_measurement(
-                    row, col, well_plate_identifier, plate_data, histograms
+                    row,
+                    col,
+                    well_plate_identifier,
+                    plate_data,
+                    histograms,
+                    header_data.copy(),
                 )
             ],
+            custom_info=custom_info,
         )
         for row in first_plate.index
         for col in first_plate.columns

src/allotropy/parsers/utils/pandas.py

@@ -279,6 +279,7 @@ def _key_matches(self, match_key: str, key: str) -> bool:
         # "++" can cause re.compile to fail. Since it is never a valid regex expression
         # it is safe to escape it to prevent the error.
         match_key = match_key.replace("++", r"\+\+")
+
         return bool(re.fullmatch(match_key, key))
 
     def _get_matching_keys(self, key_or_keys: str | set[str]) -> set[str]: