feat: Cytiva Biacore Insight - Support multiple processed data documents (enables Kinetics + Affinity model data)

src/allotropy/allotrope/schema_mappers/adm/binding_affinity_analyzer/benchling/_2024/_12/binding_affinity_analyzer.py

@@ -46,7 +46,7 @@
 from allotropy.parsers.utils.calculated_data_documents.definition import (
     CalculatedDocument,
 )
-from allotropy.parsers.utils.values import assert_not_none, has_value, quantity_or_none
+from allotropy.parsers.utils.values import assert_not_none, quantity_or_none
 from allotropy.types import DictType
 
 
@@ -108,6 +108,19 @@ class DeviceControlDocument:
     device_control_custom_info: DictType | None = None
 
 
+@dataclass(frozen=True)
+class ProcessedData:
+    """Represents a single processed data document (one analysis model)."""
+
+    model_name: str
+    binding_on_rate_measurement_datum__kon_: float | None = None
+    binding_off_rate_measurement_datum__koff_: float | None = None
+    equilibrium_dissociation_constant__kd_: float | None = None
+    maximum_binding_capacity__rmax_: float | None = None
+    processed_data_custom_info: DictType | None = None
+    data_processing_document: DictType | None = None
+
+
 @dataclass(frozen=True)
 class Measurement:
     identifier: str
@@ -126,19 +139,12 @@ class Measurement:
     # Sensorgram
     sensorgram_data_cube: DataCube | None = None
 
-    # Processed Data
-    binding_on_rate_measurement_datum__kon_: float | None = None
-    binding_off_rate_measurement_datum__koff_: float | None = None
-    equilibrium_dissociation_constant__kd_: float | None = None
-    maximum_binding_capacity__rmax_: float | None = None
-    processed_data_custom_info: DictType | None = None
+    # Processed Data - supports multiple analysis models
+    processed_data: list[ProcessedData] | None = None
 
     # Report point
     report_point_data: list[ReportPoint] | None = None
 
-    # Data processing
-    data_processing_document: DictType | None = None
-
 
 @dataclass(frozen=True)
 class MeasurementGroup:
@@ -249,60 +255,76 @@ def _get_measurement_document_item(
     def _get_surface_plasmon_resonance_measurement_document(
         self, measurement: Measurement, metadata: Metadata
     ) -> MeasurementDocumentItem:
-        processed_data_document = ProcessedDataDocumentItem(
-            data_processing_document=(
-                {
-                    key: value
-                    for key, value in measurement.data_processing_document.items()
-                    if value is not None
-                }
-                if measurement.data_processing_document
-                else None
-            ),
-            binding_on_rate_measurement_datum__kon_=quantity_or_none(
-                TQuantityValuePerMolarPerSecond,
-                measurement.binding_on_rate_measurement_datum__kon_,
-            ),
-            binding_off_rate_measurement_datum__koff_=quantity_or_none(
-                TQuantityValuePerSecond,
-                measurement.binding_off_rate_measurement_datum__koff_,
-            ),
-            equilibrium_dissociation_constant__kd_=quantity_or_none(
-                TQuantityValueMolar,
-                measurement.equilibrium_dissociation_constant__kd_,
-            ),
-            maximum_binding_capacity__rmax_=quantity_or_none(
-                TQuantityValueResponseUnit,
-                measurement.maximum_binding_capacity__rmax_,
-            ),
-            report_point_aggregate_document=(
-                ReportPointAggregateDocument(
-                    report_point_document=[
-                        add_custom_information_document(
-                            ReportPointDocumentItem(
-                                report_point_identifier=report_point.identifier,
-                                identifier_role=report_point.identifier_role,
-                                absolute_resonance=TQuantityValueResponseUnit(
-                                    value=report_point.absolute_resonance
-                                ),
-                                relative_resonance=quantity_or_none(
-                                    TQuantityValueResponseUnit,
-                                    report_point.relative_resonance,
-                                ),
-                                time_setting=TQuantityValueSecondTime(
-                                    value=report_point.time_setting
-                                ),
+        # Create report point aggregate document (only for the first processed data document)
+        report_point_aggregate = (
+            ReportPointAggregateDocument(
+                report_point_document=[
+                    add_custom_information_document(
+                        ReportPointDocumentItem(
+                            report_point_identifier=report_point.identifier,
+                            identifier_role=report_point.identifier_role,
+                            absolute_resonance=TQuantityValueResponseUnit(
+                                value=report_point.absolute_resonance
                             ),
-                            custom_info_doc=report_point.custom_info,
-                        )
-                        for report_point in measurement.report_point_data
-                    ]
-                )
-                if measurement.report_point_data
-                else None
-            ),
+                            relative_resonance=quantity_or_none(
+                                TQuantityValueResponseUnit,
+                                report_point.relative_resonance,
+                            ),
+                            time_setting=TQuantityValueSecondTime(
+                                value=report_point.time_setting
+                            ),
+                        ),
+                        custom_info_doc=report_point.custom_info,
+                    )
+                    for report_point in measurement.report_point_data
+                ]
+            )
+            if measurement.report_point_data
+            else None
         )
 
+        # Create processed data documents - one for each analysis model
+        processed_data_documents = []
+        if measurement.processed_data:
+            for idx, proc_data in enumerate(measurement.processed_data):
+                doc = ProcessedDataDocumentItem(
+                    data_processing_document=(
+                        {
+                            key: value
+                            for key, value in proc_data.data_processing_document.items()
+                            if value is not None
+                        }
+                        if proc_data.data_processing_document
+                        else None
+                    ),
+                    binding_on_rate_measurement_datum__kon_=quantity_or_none(
+                        TQuantityValuePerMolarPerSecond,
+                        proc_data.binding_on_rate_measurement_datum__kon_,
+                    ),
+                    binding_off_rate_measurement_datum__koff_=quantity_or_none(
+                        TQuantityValuePerSecond,
+                        proc_data.binding_off_rate_measurement_datum__koff_,
+                    ),
+                    equilibrium_dissociation_constant__kd_=quantity_or_none(
+                        TQuantityValueMolar,
+                        proc_data.equilibrium_dissociation_constant__kd_,
+                    ),
+                    maximum_binding_capacity__rmax_=quantity_or_none(
+                        TQuantityValueResponseUnit,
+                        proc_data.maximum_binding_capacity__rmax_,
+                    ),
+                    # Only include report points in the first processed data document
+                    report_point_aggregate_document=report_point_aggregate
+                    if idx == 0
+                    else None,
+                )
+                processed_data_documents.append(
+                    add_custom_information_document(
+                        doc,
+                        custom_info_doc=proc_data.processed_data_custom_info,
+                    )
+                )
+
         return MeasurementDocumentItem(
             measurement_identifier=measurement.identifier,
             sample_document=add_custom_information_document(
@@ -361,14 +383,9 @@ def _get_surface_plasmon_resonance_measurement_document(
             ),
             processed_data_aggregate_document=(
                 ProcessedDataAggregateDocument(
-                    processed_data_document=[
-                        add_custom_information_document(
-                            processed_data_document,
-                            custom_info_doc=measurement.processed_data_custom_info,
-                        )
-                    ]
+                    processed_data_document=processed_data_documents
                 )
-                if has_value(processed_data_document)
+                if processed_data_documents
                 else None
             ),
         )

src/allotropy/allotrope/schemas.py

@@ -85,7 +85,7 @@ def validate_asm_schema(asm_dict: dict[str, Any]) -> None:
         resolver = jsonschema.RefResolver(
             base_uri=schema.get("$id", ""),
             referrer=schema,
-            store=store,
+            store=store,  # type: ignore[arg-type]
         )
         validator = jsonschema.validators.Draft202012Validator(
             schema, resolver=resolver, format_checker=FORMAT_CHECKER

src/allotropy/parsers/cytiva_biacore_insight/cytiva_biacore_insight_data_creator.py

@@ -20,6 +20,7 @@
     MeasurementGroup,
     MeasurementType,
     Metadata,
+    ProcessedData,
     ReportPoint,
 )
 from allotropy.parsers.constants import NOT_APPLICABLE
@@ -113,77 +114,106 @@ def _get_measurements(
 ) -> list[Measurement]:
     data_processing_document = dict(metadata.data_processing_document or {})
 
-    return [
-        Measurement(
-            identifier=measurement.identifier,
-            sample_identifier=measurement.sample_identifier,
-            type_=MeasurementType.SURFACE_PLASMON_RESONANCE,
-            method_name=measurement.method_name,
-            ligand_identifier=measurement.ligand_identifier,
-            device_control_document=measurement.device_control_document,
-            sample_custom_info=measurement.sample_custom_info,
-            binding_on_rate_measurement_datum__kon_=(
-                measurement.kinetics.binding_on_rate_measurement_datum
-            ),
-            binding_off_rate_measurement_datum__koff_=(
-                measurement.kinetics.binding_off_rate_measurement_datum
-            ),
-            equilibrium_dissociation_constant__kd_=(
-                measurement.kinetics.equilibrium_dissociation_constant
-            ),
-            maximum_binding_capacity__rmax_=(
-                measurement.kinetics.maximum_binding_capacity
-            ),
-            processed_data_custom_info=_clean_custom_info(
-                {
-                    (
-                        "Affinity Chi squared"
-                        if measurement.kinetics.is_affinity_measurement
-                        else "Kinetics Chi squared"
-                    ): (
-                        TQuantityValue(value=v, unit="RU^2")
-                        if (v := measurement.kinetics.kinetics_chi_squared) is not None
-                        else None
+    measurements = []
+    for measurement in measurement_data:
+        # Create ProcessedData for each kinetics model
+        processed_data_list = []
+
+        if measurement.kinetics:
+            # If we have kinetics data, create one ProcessedData per model
+            for kinetics in measurement.kinetics:
+                # Determine chi-squared label based on model type
+                chi_squared_label = (
+                    "Affinity Chi squared"
+                    if "affinity" in kinetics.model_name.lower()
+                    else "Kinetics Chi squared"
+                )
+
+                processed_data = ProcessedData(
+                    model_name=kinetics.model_name,
+                    binding_on_rate_measurement_datum__kon_=(
+                        kinetics.binding_on_rate_measurement_datum
                     ),
-                    "tc": quantity_or_none(
-                        TQuantityValueUnitless, measurement.kinetics.tc
+                    binding_off_rate_measurement_datum__koff_=(
+                        kinetics.binding_off_rate_measurement_datum
                     ),
-                    "offset": quantity_or_none(
-                        TQuantityValueResponseUnit,
-                        measurement.kinetics.offset,
+                    equilibrium_dissociation_constant__kd_=(
+                        kinetics.equilibrium_dissociation_constant
                     ),
-                }
-            ),
-            report_point_data=[
-                ReportPoint(
-                    identifier=rp.identifier,
-                    identifier_role=rp.identifier_role,
-                    absolute_resonance=rp.absolute_resonance,
-                    time_setting=rp.time_setting,
-                    relative_resonance=rp.relative_resonance,
-                    custom_info=_clean_custom_info(
+                    maximum_binding_capacity__rmax_=(kinetics.maximum_binding_capacity),
+                    processed_data_custom_info=_clean_custom_info(
                         {
-                            "Step purpose": rp.step_purpose,
-                            "Window": quantity_or_none(
-                                TQuantityValueSecondTime, rp.window
+                            chi_squared_label: (
+                                TQuantityValue(value=v, unit="RU^2")
+                                if (v := kinetics.kinetics_chi_squared) is not None
+                                else None
+                            ),
+                            "tc": quantity_or_none(TQuantityValueUnitless, kinetics.tc),
+                            "U-value": quantity_or_none(
+                                TQuantityValueUnitless, kinetics.u_value
+                            ),
+                            "offset": quantity_or_none(
+                                TQuantityValueResponseUnit,
+                                kinetics.offset,
                             ),
-                            "Baseline": rp.baseline,
                         }
                     ),
+                    data_processing_document=_clean_custom_info(
+                        {
+                            **data_processing_document,
+                            "Acceptance State": kinetics.acceptance_state,
+                            "Curve Markers": kinetics.curve_markers,
+                            "Kinetics Model": kinetics.model_name,
+                        }
+                    ),
+                )
+                processed_data_list.append(processed_data)
+        else:
+            # If no kinetics data, create a single ProcessedData with just data processing info
+            processed_data_list.append(
+                ProcessedData(
+                    model_name="N/A",  # No model when there's no kinetics data
+                    data_processing_document=_clean_custom_info(
+                        data_processing_document
+                    )
+                    if data_processing_document
+                    else None,
                 )
-                for rp in measurement.report_point_data
-            ],
-            data_processing_document=_clean_custom_info(
-                {
-                    **data_processing_document,
-                    "Acceptance State": measurement.kinetics.acceptance_state,
-                    "Curve Markers": measurement.kinetics.curve_markers,
-                    "Kinetics Model": measurement.kinetics.kinetics_model,
-                }
-            ),
+            )
+
+        measurements.append(
+            Measurement(
+                identifier=measurement.identifier,
+                sample_identifier=measurement.sample_identifier,
+                type_=MeasurementType.SURFACE_PLASMON_RESONANCE,
+                method_name=measurement.method_name,
+                ligand_identifier=measurement.ligand_identifier,
+                device_control_document=measurement.device_control_document,
+                sample_custom_info=measurement.sample_custom_info,
+                processed_data=processed_data_list,
+                report_point_data=[
+                    ReportPoint(
+                        identifier=rp.identifier,
+                        identifier_role=rp.identifier_role,
+                        absolute_resonance=rp.absolute_resonance,
+                        time_setting=rp.time_setting,
+                        relative_resonance=rp.relative_resonance,
+                        custom_info=_clean_custom_info(
+                            {
+                                "Step purpose": rp.step_purpose,
+                                "Window": quantity_or_none(
+                                    TQuantityValueSecondTime, rp.window
+                                ),
+                                "Baseline": rp.baseline,
+                            }
+                        ),
+                    )
+                    for rp in measurement.report_point_data
+                ],
+            )
         )
-        for measurement in measurement_data
-    ]
+
+    return measurements
 
 
 def create_calculated_data(data: Data) -> list[CalculatedDocument]: