feat: Luminex xPONENT - get unread data

src/allotropy/allotrope/schema_mappers/adm/multi_analyte_profiling/benchling/_2024/_09/multi_analyte_profiling.py

@@ -1,5 +1,7 @@
 from dataclasses import dataclass
+from typing import Any
 
+from allotropy.allotrope.converter import add_custom_information_document
 from allotropy.allotrope.models.adm.multi_analyte_profiling.benchling._2024._09.multi_analyte_profiling import (
     AnalyteAggregateDocument,
     AnalyteDocumentItem,
@@ -101,6 +103,11 @@ class Measurement:
     minimum_assay_bead_count_setting: float | None = None
     detector_gain_setting: str | None = None
 
+    # custom
+    measurement_custom_info: dict[str, Any] | None = None
+    sample_custom_info: dict[str, Any] | None = None
+    device_control_custom_info: dict[str, Any] | None = None
+
 
 @dataclass(frozen=True)
 class MeasurementGroup:
@@ -112,6 +119,7 @@ class MeasurementGroup:
     method_version: str | None = None
     container_type: str | None = None
     experiment_type: str | None = None
+    custom_info: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)
@@ -136,6 +144,7 @@ class Metadata:
     product_manufacturer: str | None = None
 
     calibrations: list[Calibration] | None = None
+    custom_info: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)
@@ -151,14 +160,17 @@ class Mapper(SchemaMapper[Data, Model]):
     def map_model(self, data: Data) -> Model:
         return Model(
             multi_analyte_profiling_aggregate_document=MultiAnalyteProfilingAggregateDocument(
-                device_system_document=DeviceSystemDocument(
-                    model_number=data.metadata.model_number,
-                    product_manufacturer=data.metadata.product_manufacturer,
-                    firmware_version=data.metadata.firmware_version,
-                    equipment_serial_number=data.metadata.equipment_serial_number,
-                    calibration_aggregate_document=self._get_calibration_aggregate_document(
-                        data.metadata.calibrations
+                device_system_document=add_custom_information_document(
+                    DeviceSystemDocument(
+                        model_number=data.metadata.model_number,
+                        product_manufacturer=data.metadata.product_manufacturer,
+                        firmware_version=data.metadata.firmware_version,
+                        equipment_serial_number=data.metadata.equipment_serial_number,
+                        calibration_aggregate_document=self._get_calibration_aggregate_document(
+                            data.metadata.calibrations
+                        ),
                     ),
+                    custom_info_doc=data.metadata.custom_info,
                 ),
                 data_system_document=DataSystemDocument(
                     data_system_instance_identifier=data.metadata.data_system_instance_identifier,
@@ -186,44 +198,55 @@ def _get_technique_document(
     ) -> MultiAnalyteProfilingDocumentItem:
         return MultiAnalyteProfilingDocumentItem(
             analyst=measurement_group.analyst,
-            measurement_aggregate_document=MeasurementAggregateDocument(
-                experiment_type=measurement_group.experiment_type,
-                analytical_method_identifier=measurement_group.analytical_method_identifier,
-                method_version=measurement_group.method_version,
-                experimental_data_identifier=measurement_group.experimental_data_identifier,
-                container_type=measurement_group.container_type,
-                plate_well_count=quantity_or_none(
-                    TQuantityValueNumber, measurement_group.plate_well_count
+            measurement_aggregate_document=add_custom_information_document(
+                MeasurementAggregateDocument(
+                    experiment_type=measurement_group.experiment_type,
+                    analytical_method_identifier=measurement_group.analytical_method_identifier,
+                    method_version=measurement_group.method_version,
+                    experimental_data_identifier=measurement_group.experimental_data_identifier,
+                    container_type=measurement_group.container_type,
+                    plate_well_count=quantity_or_none(
+                        TQuantityValueNumber, measurement_group.plate_well_count
+                    ),
+                    measurement_document=[
+                        self._get_measurement_document(measurement, metadata)
+                        for measurement in measurement_group.measurements
+                    ],
                 ),
-                measurement_document=[
-                    self._get_measurement_document(measurement, metadata)
-                    for measurement in measurement_group.measurements
-                ],
+                custom_info_doc=measurement_group.custom_info,
             ),
         )
 
     def _get_measurement_document(
         self, measurement: Measurement, metadata: Metadata
     ) -> MeasurementDocumentItem:
-        return MeasurementDocumentItem(
+        measurement_document = MeasurementDocumentItem(
             measurement_identifier=measurement.identifier,
             measurement_time=self.get_date_time(measurement.measurement_time),
-            sample_document=self._get_sample_document(measurement),
+            sample_document=add_custom_information_document(
+                self._get_sample_document(measurement),
+                custom_info_doc=measurement.sample_custom_info,
+            ),
             device_control_aggregate_document=DeviceControlAggregateDocument(
                 device_control_document=[
-                    DeviceControlDocumentItem(
-                        device_type=metadata.device_type,
-                        sample_volume_setting=quantity_or_none(
-                            TQuantityValueMicroliter, measurement.sample_volume_setting
-                        ),
-                        dilution_factor_setting=quantity_or_none(
-                            TQuantityValueUnitless, measurement.dilution_factor_setting
-                        ),
-                        detector_gain_setting=measurement.detector_gain_setting,
-                        minimum_assay_bead_count_threshold_setting=quantity_or_none(
-                            TQuantityValueNumber,
-                            measurement.minimum_assay_bead_count_setting,
+                    add_custom_information_document(
+                        DeviceControlDocumentItem(
+                            device_type=metadata.device_type,
+                            sample_volume_setting=quantity_or_none(
+                                TQuantityValueMicroliter,
+                                measurement.sample_volume_setting,
+                            ),
+                            dilution_factor_setting=quantity_or_none(
+                                TQuantityValueUnitless,
+                                measurement.dilution_factor_setting,
+                            ),
+                            detector_gain_setting=measurement.detector_gain_setting,
+                            minimum_assay_bead_count_threshold_setting=quantity_or_none(
+                                TQuantityValueNumber,
+                                measurement.minimum_assay_bead_count_setting,
+                            ),
                         ),
+                        custom_info_doc=measurement.device_control_custom_info,
                     )
                 ]
             ),
@@ -275,6 +298,11 @@ def _get_measurement_document(
             ),
         )
 
+        return add_custom_information_document(
+            measurement_document,
+            custom_info_doc=measurement.measurement_custom_info,
+        )
+
     def _get_sample_document(self, measurement: Measurement) -> SampleDocument:
         return SampleDocument(
             sample_identifier=measurement.sample_identifier,

src/allotropy/parsers/luminex_xponent/luminex_xponent_structure.py

@@ -64,35 +64,49 @@ class Header:
     detector_gain_setting: str | None
     minimum_assay_bead_count_setting: float | None
     analyst: str | None
+    custom_info: dict[str, Any]
 
     @classmethod
     def create(
-        cls, header_data: pd.DataFrame, minimum_assay_bead_count_setting: float | None
+        cls,
+        header_data: pd.DataFrame,
+        header_row: SeriesData,
+        minimum_assay_bead_count_setting: float | None,
     ) -> Header:
-        info_row = SeriesData(header_data.iloc[0])
-        raw_datetime = info_row[str, "BatchStartTime"]
-        sample_volume = info_row.get(str, ["SampleVolume", "MaxSampleUptakeVolume"])
+        sample_volume = header_row.get(str, ["SampleVolume", "MaxSampleUptakeVolume"])
+
+        header_row.mark_read({"Program", "ProtocolPlate", "Date"})
 
         return Header(
             model_number=cls._get_model_number(header_data),
-            software_version=info_row.get(str, "Build"),
-            equipment_serial_number=info_row.get(str, "SN"),
-            analytical_method_identifier=info_row.get(str, "ProtocolName"),
-            method_version=info_row.get(str, "ProtocolVersion"),
-            experimental_data_identifier=info_row.get(str, "Batch"),
+            software_version=header_row.get(str, "Build"),
+            equipment_serial_number=header_row.get(str, "SN"),
+            analytical_method_identifier=header_row.get(str, "ProtocolName"),
+            method_version=header_row.get(str, "ProtocolVersion"),
+            experimental_data_identifier=header_row.get(str, "Batch"),
             sample_volume_setting=(
                 try_float(sample_volume.split()[0], "sample volume setting")
                 if sample_volume
                 else None
             ),
             plate_well_count=cls._get_plate_well_count(header_data),
-            measurement_time=raw_datetime,
-            detector_gain_setting=info_row.get(
+            measurement_time=header_row[str, "BatchStartTime"],
+            detector_gain_setting=header_row.get(
                 str, ["ProtocolReporterGain", "ProtocolOperatingMode"]
             ),
-            data_system_instance_identifier=info_row[str, "ComputerName"],
+            data_system_instance_identifier=header_row[str, "ComputerName"],
             minimum_assay_bead_count_setting=minimum_assay_bead_count_setting,
-            analyst=info_row.get(str, "Operator"),
+            analyst=header_row.get(str, "Operator"),
+            custom_info={
+                "Country Code": header_row.get(str, "Country Code"),
+                "ProtocolDevelopingCompany": header_row.get(
+                    str, "ProtocolDevelopingCompany"
+                ),
+                "Version": header_row.get(str, "Version"),
+                # Used for the measurement groups
+                "BatchStopTime": header_row.get(str, "BatchStopTime"),
+                "ProtocolDescription": header_row.get(str, "ProtocolDescription"),
+            },
         )
 
     @classmethod
@@ -138,7 +152,10 @@ def create_calibration(calibration_data: SeriesData) -> Calibration:
         msg = f"Expected at least two columns on the calibration line, got:\n{calibration_data.series}."
         raise AllotropeConversionError(msg)
 
+    # Read the calibration data using SeriesData methods
+    calibration_name = calibration_data.series.iloc[0]
     calibration_info = str(calibration_data.series.iloc[1]).strip()
+    calibration_data.get_unread()
 
     # Check if the calibration info starts with a known status value
     status_values = ["Passed", "Failed", "Calibrated", "Verified"]
@@ -164,7 +181,7 @@ def create_calibration(calibration_data: SeriesData) -> Calibration:
         raise AllotropeConversionError(msg)
 
     return Calibration(
-        name=calibration_data.series.iloc[0].replace("Last", "").strip(),
+        name=calibration_name.replace("Last", "").strip(),
         time=time,
         report=report,
     )
@@ -180,13 +197,17 @@ class Measurement:
     analytes: list[Analyte]
     calculated_data: list[CalculatedDocument]
     errors: list[Error] | None = None
+    measurement_custom_info: dict[str, Any] | None = None
+    sample_custom_info: dict[str, Any] | None = None
+    device_control_custom_info: dict[str, Any] | None = None
 
     @classmethod
     def create(
         cls,
         results_data: dict[str, pd.DataFrame],
         count_data: SeriesData,
         bead_ids_data: SeriesData,
+        header_row: SeriesData,
     ) -> Measurement:
         location = str(count_data.series.name)
         dilution_factor_data = results_data["Dilution Factor"]
@@ -201,9 +222,12 @@ def create(
         metadata_keys = ["Sample", "Total Events"]
 
         well_location, location_id = cls._get_location_details(location)
-        dilution_factor_setting = SeriesData(dilution_factor_data.loc[location]).get(
+        dilution_factor_series = SeriesData(dilution_factor_data.loc[location])
+        dilution_factor_setting = dilution_factor_series.get(
             float, "Dilution Factor", NEGATIVE_ZERO, validate=SeriesData.NOT_NAN
         )
+        # Capture unread dilution factor data to prevent warnings
+        dilution_factor_series.get_unread()
         errors: list[Error] = []
         data_errors = cls._get_errors(errors_data, well_location) or []
         for data_error in data_errors:
@@ -289,6 +313,20 @@ def get_statistic_dimensions(analyte: str) -> list[StatisticDimension]:
                     )
                 )
 
+        device_control_custom_info = {
+            "ProtocolHeater": header_row.get(str, "ProtocolHeater"),
+            "DDGate": header_row.get(str, "DDGate"),
+            "SampleTimeout": header_row.get(str, "SampleTimeout"),
+            "ProtocolAnalysis": header_row.get(str, "ProtocolAnalysis"),
+            "ProtocolMicrosphere": header_row.get(str, "ProtocolMicrosphere"),
+            "PlateReadDirection": header_row.get(str, "PlateReadDirection"),
+        }
+        sample_custom_info = {
+            "BatchDescription": header_row.get(str, "BatchDescription"),
+            "PanelName": header_row.get(str, "PanelName"),
+            "BeadType": header_row.get(str, "BeadType"),
+        }
+
         return Measurement(
             identifier=measurement_identifier,
             sample_identifier=count_data[str, "Sample"],
@@ -298,6 +336,9 @@ def get_statistic_dimensions(analyte: str) -> list[StatisticDimension]:
             analytes=analytes,
             errors=errors,
             calculated_data=calculated_data,
+            device_control_custom_info=device_control_custom_info,
+            sample_custom_info=sample_custom_info,
+            measurement_custom_info=count_data.get_unread() | header_row.get_unread(),
         )
 
     @classmethod
@@ -316,17 +357,24 @@ def _get_errors(
     ) -> list[str] | None:
         if errors_data is None or well_location not in errors_data.index:
             return None
-        return map_rows(
-            errors_data.loc[[well_location]], lambda data: data[str, "Message"]
-        )
+
+        def extract_message(data: SeriesData) -> str:
+            message = data[str, "Message"]
+            # Capture unread error data to prevent warnings
+            data.get_unread()
+            return message
+
+        return map_rows(errors_data.loc[[well_location]], extract_message)
 
 
 @dataclass(frozen=True)
 class MeasurementList:
     measurements: list[Measurement]
 
     @classmethod
-    def create(cls, results_data: dict[str, pd.DataFrame]) -> MeasurementList:
+    def create(
+        cls, results_data: dict[str, pd.DataFrame], header_row: SeriesData
+    ) -> MeasurementList:
         if missing_sections := [
             section for section in REQUIRED_SECTIONS if section not in results_data
         ]:
@@ -352,6 +400,7 @@ def create_measurement(count_data: SeriesData) -> Measurement:
                 results_data=results_data,
                 count_data=count_data,
                 bead_ids_data=bead_ids_data,
+                header_row=header_row,
             )
 
         return MeasurementList(map_rows(results_data["Count"], create_measurement))
@@ -365,12 +414,14 @@ class Data:
 
     @classmethod
     def create(cls, reader: LuminexXponentReader) -> Data:
+        header_row = SeriesData(reader.header_data.iloc[0])
+
         return Data(
             header=Header.create(
-                reader.header_data, reader.minimum_assay_bead_count_setting
+                reader.header_data, header_row, reader.minimum_assay_bead_count_setting
             ),
             calibrations=map_rows(reader.calibration_data, create_calibration),
-            measurement_list=MeasurementList.create(reader.results_data),
+            measurement_list=MeasurementList.create(reader.results_data, header_row),
         )
 
 
@@ -389,6 +440,7 @@ def create_metadata(
         software_name=DEFAULT_SOFTWARE_NAME,
         software_version=header.software_version,
         device_type=DEFAULT_DEVICE_TYPE,
+        custom_info=header.custom_info,
     )
 
 
@@ -416,11 +468,24 @@ def create_measurement_groups(
                     assay_bead_count=measurement.assay_bead_count,
                     analytes=measurement.analytes,
                     errors=measurement.errors if measurement.errors else None,
+                    measurement_custom_info=measurement.measurement_custom_info,
+                    sample_custom_info=measurement.sample_custom_info,
+                    device_control_custom_info=measurement.device_control_custom_info,
                 )
             ],
+            custom_info={
+                "BatchStopTime": header.custom_info.get("BatchStopTime", None),
+                "ProtocolDescription": header.custom_info.get(
+                    "ProtocolDescription", None
+                ),
+            },
         )
         for measurement in measurements
     ]
+    # Remove the header custom info that was used to create the measurement groups.
+    header.custom_info.pop("BatchStopTime", None)
+    header.custom_info.pop("ProtocolDescription", None)
+
     calculated_documents = list(
         itertools.chain(*[measurement.calculated_data for measurement in measurements])
     )