Added extra fields cytiva unread

Author: felipenarv

src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_data_creator.py

@@ -179,6 +179,7 @@ def create_measurements(
                 "molecular weight": quantity_or_none(
                     TQuantityValueDalton, measurement.molecular_weight
                 ),
+                **measurement.sample_custom_info,
             },
             sensorgram_data_cube=_get_sensorgram_datacube(measurement.sensorgram_data),
             report_point_data=(

src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_parser.py

@@ -1,3 +1,7 @@
+import json
+from pathlib import Path, PureWindowsPath
+import warnings
+
 from allotropy.allotrope.models.adm.binding_affinity_analyzer.wd._2024._12.binding_affinity_analyzer import (
     Model,
 )
@@ -30,9 +34,26 @@ class CytivaBiacoreT200ControlParser(VendorParser[MapperData, Model]):
     SCHEMA_MAPPER = Mapper
 
     def create_data(self, named_file_contents: NamedFileContents) -> MapperData:
-        data = Data.create(DictData(decode_data(named_file_contents)))
-        return MapperData(
+        base_data = DictData(decode_data(named_file_contents))
+        data = Data.create(base_data)
+        mapper_data = MapperData(
             metadata=create_metadata(data, named_file_contents),
             measurement_groups=create_measurement_groups(data),
             calculated_data=create_calculated_data(data),
         )
+        try:
+            unread = base_data.get_unread_deep()
+            original_path = named_file_contents.original_file_path
+            # Derive file stem robustly for both POSIX and Windows-style paths
+            stem = (
+                PureWindowsPath(original_path).stem
+                if "\\" in original_path and "/" not in original_path
+                else Path(original_path).stem
+            )
+            out_name = f"{stem}data_unread_2.json"
+            with open(out_name, "w", encoding="utf-8") as f:
+                json.dump(unread, f, ensure_ascii=False, indent=2)
+        except Exception as e:
+            # Best-effort debug artifact; do not break parsing on failure
+            warnings.warn(f"Failed to write unread debug file: {e!s}", stacklevel=1)
+        return mapper_data

src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_structure.py

@@ -21,7 +21,6 @@
     assert_not_none,
     try_float_or_none,
 )
-from allotropy.types import DictType
 
 
 @dataclass(frozen=True)
@@ -35,6 +34,7 @@ class ChipData:
 
     @staticmethod
     def create(chip_data: DictData) -> ChipData:
+        chip_data.mark_read_deep({"IFCType", "OSType", "OSVersion"})
         return ChipData(
             sensor_chip_identifier=assert_not_none(chip_data.get(str, "Id"), "Chip ID"),
             sensor_chip_type=chip_data.get(str, "Name"),
@@ -46,7 +46,7 @@ def create(chip_data: DictData) -> ChipData:
                 "last modified time": chip_data.get(str, "LastModTime"),
                 "last use time": chip_data.get(str, "LastUseTime"),
                 "first dock date": chip_data.get(str, "FirstDockDate"),
-                **chip_data.get_unread(skip={"IFCType", "OSType", "OSVersion"}),
+                **chip_data.get_unread_deep(),
             },
         )
 
@@ -203,16 +203,31 @@ class ReportPointData:
     absolute_resonance: float
     relative_resonance: float | None
     time_setting: float
-    custom_info: DictType
+    custom_info: dict[str, Any]
     min_resonance: float
     max_resonance: float
     lrsd: float
     slope: float
     sd: float
+    sample_custom_info: dict[str, Any]
+    measurement_aggregate_custom_info: dict[str, Any]
 
     @staticmethod
     def create(data: SeriesData) -> ReportPointData:
-        return ReportPointData(
+        # This is to mark the keys as read so they don't get added to the custom_info, which has to use a regex
+        # because some keys look like 'Sample1_Ligand'
+        for key in (
+            "FlowRate",
+            "ContactTime",
+            "Chip",
+            "Ligand",
+            "Method",
+            "Fc",
+            "Sample",
+        ):
+            data.get_unread(f".*{key}.*")
+
+        report_point_data = ReportPointData(
             identifier=random_uuid_str(),
             identifier_role=data[str, "Id"],
             absolute_resonance=data[float, "AbsResp"],
@@ -236,8 +251,15 @@ def create(data: SeriesData) -> ReportPointData:
                 "assay_step_purpose": data.get(str, "AssayStepPurpose"),
                 "buffer": data.get(str, "Buffer"),
             },
+            sample_custom_info=data.get_custom_keys({"Cycle"}),
+            measurement_aggregate_custom_info=data.get_custom_keys({"Procedure"}),
         )
 
+        unread_data = data.get_unread()
+        for key in list(unread_data.keys()):
+            report_point_data.custom_info[key] = unread_data[key]
+        return report_point_data
+
 
 @dataclass(frozen=True)
 class MeasurementData:
@@ -259,7 +281,7 @@ class MeasurementData:
     flow_rate: float | None
     contact_time: float | None
     dilution: float | None
-    custom_info: dict[str, Any]
+    sample_custom_info: dict[str, Any]
 
 
 @dataclass(frozen=True)
@@ -268,15 +290,15 @@ class SampleData:
     custom_info: dict[str, Any]
 
     @staticmethod
-    def create(intermediate_structured_data: DictData) -> SampleData:
-        application_template_details = intermediate_structured_data.get_nested(
-            "application_template_details"
-        )
+    def create(
+        intermediate_structured_data: DictData, application_template_details: DictData
+    ) -> SampleData:
         measurements: dict[str, list[MeasurementData]] = defaultdict(list)
         total_cycles = assert_not_none(
             intermediate_structured_data.get(int, "total_cycles"),
             "total_cycles",
         )
+
         for idx in range(total_cycles):
             flowcell_cycle_json = application_template_details.get_nested(
                 f"Flowcell {idx + 1}"
@@ -324,15 +346,33 @@ def create(intermediate_structured_data: DictData) -> SampleData:
                     flow_rate=flowcell_cycle_json.get(float, "Flow"),
                     contact_time=flowcell_cycle_json.get(float, "ContactTime"),
                     dilution=flowcell_cycle_json.get(float, "DilutePercent"),
-                    custom_info={},
+                    sample_custom_info={},
                 )
                 # group sensorgram data by Flow Cell Number (Fc in rpoint data)
                 for flow_cell, sensorgram_df in sensorgram_data.groupby(
                     "Flow Cell Number"
                 )
             ]
-        custom_info: dict[str, Any] = {}
-        return SampleData(measurements, custom_info)
+            # Add custom info from report point data
+            for measurement_data in measurements[sample_location_key]:
+                if measurement_data.report_point_data:
+                    for report_point_data_item in measurement_data.report_point_data:
+                        measurement_data.sample_custom_info.update(
+                            report_point_data_item.sample_custom_info
+                        )
+                        break
+        return SampleData(measurements, {})
+
+    def get_measurement_aggregate_custom_info(self) -> dict[str, Any]:
+        measurement_aggregate_custom_info: dict[str, Any] = {}
+        for measurements_by_sample_location_key in self.measurements.values():
+            for measurement_data in measurements_by_sample_location_key:
+                if measurement_data.report_point_data:
+                    for report_point_data_item in measurement_data.report_point_data:
+                        measurement_aggregate_custom_info.update(
+                            report_point_data_item.measurement_aggregate_custom_info
+                        )
+        return measurement_aggregate_custom_info
 
 
 @dataclass(frozen=True)
@@ -360,9 +400,15 @@ def create(intermediate_structured_data: DictData) -> Data:
         system_information = SystemInformation.create(system_information_dictdata)
         chip_data_dictdata = intermediate_structured_data.get_nested("chip")
         chip_data = ChipData.create(chip_data_dictdata)
-        sample_data = SampleData.create(intermediate_structured_data)
+        sample_data = SampleData.create(
+            intermediate_structured_data, application_template_details
+        )
+
+        system_information.measurement_aggregate_custom_info.update(
+            sample_data.get_measurement_aggregate_custom_info()
+        )
 
-        # This has to be called later because in SampleData.create some of the unread keys are called
+        # This has to be called later because SampleData.create uses some of the unread keys
         run_metadata.set_device_custom_info(application_template_details)
 
         return Data(

src/allotropy/parsers/utils/dict_data.py

@@ -431,8 +431,11 @@ def get_keys_as_dict(
                 "output_field2": (float, "input_field2", 0.0),
                 "renamed_field": (int, "original_name", None),
             }
-            result = dict_data.get_keys_as_dict(field_mappings)
-
+            output = {
+                "output_field1": "value1",
+                "output_field2": 1.0,
+                "renamed_field": 1,
+            }
         """
         result: dict[str, Any] = {}
 

tests/allotrope/schema_parser/generate_schemas_test.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+import re
 
 import pytest
 
@@ -19,10 +20,33 @@ def _get_schema_paths() -> list[Path]:
     ]
 
 
+def _pick_subset(paths: list[Path], count: int = 8) -> list[Path]:
+    if not paths:
+        return []
+    if len(paths) <= count:
+        return paths
+    step = max(1, len(paths) // count)
+    # Evenly sample across the set for broad coverage
+    return [paths[i] for i in range(0, len(paths), step)][:count]
+
+
+def test_generate_schemas_smoke_subset() -> None:
+    """Fast smoke test over a representative subset of schemas."""
+    paths = _get_schema_paths()
+    subset = _pick_subset(paths, count=8)
+    # Build a regex that matches exactly any of the chosen relative schema paths
+    escaped = [re.escape(str(p)) for p in subset]
+    pattern = rf"^({'|'.join(escaped)})$"
+    models_changed = generate_schemas(dry_run=True, schema_regex=pattern)
+    assert (
+        not models_changed
+    ), f"Expected no models files to have changed by generate-schemas script, found changes in: {models_changed}.\nPlease run 'hatch run scripts:generate-schemas' and validate the changes."
+
+
 @pytest.mark.long
-@pytest.mark.parametrize("schema_path", _get_schema_paths())
-def test_generate_schemas_runs_to_completion(schema_path: Path) -> None:
-    models_changed = generate_schemas(dry_run=True, schema_regex=str(schema_path))
+def test_generate_schemas_runs_to_completion() -> None:
+    """Full run once across all schemas. Marked long for optional execution."""
+    models_changed = generate_schemas(dry_run=True)
     assert (
         not models_changed
     ), f"Expected no models files to have changed by generate-schemas script, found changes in: {models_changed}.\nPlease run 'hatch run scripts:generate-schemas' and validate the changes."