feat: Perkin elmer envision get unread (#1073)

Author: oscar-castro-bl

src/allotropy/allotrope/schema_mappers/adm/plate_reader/rec/_2024/_06/plate_reader.py

@@ -236,27 +236,31 @@ def _get_technique_document(
     ) -> PlateReaderDocumentItem:
         plate_reader_doc = PlateReaderDocumentItem(
             analyst=measurement_group.analyst,
-            measurement_aggregate_document=MeasurementAggregateDocument(
-                analytical_method_identifier=measurement_group.analytical_method_identifier,
-                experimental_data_identifier=measurement_group.experimental_data_identifier,
-                experiment_type=measurement_group.experiment_type,
-                container_type=ContainerType.well_plate.value,
-                plate_well_count=TQuantityValueNumber(
-                    value=measurement_group.plate_well_count
+            measurement_aggregate_document=add_custom_information_document(
+                MeasurementAggregateDocument(
+                    analytical_method_identifier=measurement_group.analytical_method_identifier,
+                    experimental_data_identifier=measurement_group.experimental_data_identifier,
+                    experiment_type=measurement_group.experiment_type,
+                    container_type=ContainerType.well_plate.value,
+                    plate_well_count=TQuantityValueNumber(
+                        value=measurement_group.plate_well_count
+                    ),
+                    measurement_time=self.get_date_time(
+                        measurement_group.measurement_time
+                    ),
+                    measurement_document=[
+                        self._get_measurement_document(measurement)
+                        for measurement in measurement_group.measurements
+                    ],
                 ),
-                measurement_time=self.get_date_time(measurement_group.measurement_time),
-                measurement_document=[
-                    self._get_measurement_document(measurement)
-                    for measurement in measurement_group.measurements
-                ],
+                measurement_group.custom_info,
             ),
         )
         custom_info_doc = {
             "maximum wavelength signal": quantity_or_none(
                 TQuantityValueNanometer, measurement_group.maximum_wavelength_signal
             )
         }
-        custom_info_doc.update(measurement_group.custom_info or {})
         return add_custom_information_document(plate_reader_doc, custom_info_doc)
 
     def _get_measurement_document(
@@ -572,23 +576,26 @@ def _get_calculated_data_aggregate_document(
 
         return CalculatedDataAggregateDocument(
             calculated_data_document=[
-                CalculatedDataDocumentItem(
-                    calculated_data_identifier=calculated_data_item.uuid,
-                    calculated_data_name=calculated_data_item.name,
-                    calculation_description=calculated_data_item.description,
-                    calculated_result=TQuantityValueModel(
-                        value=calculated_data_item.value,
-                        unit=calculated_data_item.unit or Unitless.unit,
-                    ),
-                    data_source_aggregate_document=DataSourceAggregateDocument(
-                        data_source_document=[
-                            DataSourceDocumentItem(
-                                data_source_identifier=item.reference.uuid,
-                                data_source_feature=item.feature,
-                            )
-                            for item in calculated_data_item.data_sources
-                        ]
+                add_custom_information_document(
+                    CalculatedDataDocumentItem(
+                        calculated_data_identifier=calculated_data_item.uuid,
+                        calculated_data_name=calculated_data_item.name,
+                        calculation_description=calculated_data_item.description,
+                        calculated_result=TQuantityValueModel(
+                            value=calculated_data_item.value,
+                            unit=calculated_data_item.unit or Unitless.unit,
+                        ),
+                        data_source_aggregate_document=DataSourceAggregateDocument(
+                            data_source_document=[
+                                DataSourceDocumentItem(
+                                    data_source_identifier=item.reference.uuid,
+                                    data_source_feature=item.feature,
+                                )
+                                for item in calculated_data_item.data_sources
+                            ]
+                        ),
                     ),
+                    calculated_data_item.custom_info,
                 )
                 for calculated_data_item in calculated_data_items
             ]

src/allotropy/parsers/perkin_elmer_envision/perkin_elmer_envision_structure.py

@@ -22,11 +22,16 @@
 from enum import Enum
 from pathlib import Path
 from re import search
+from typing import Any
 
 import numpy as np
 import pandas as pd
 
 from allotropy.allotrope.models.shared.components.plate_reader import SampleRoleType
+from allotropy.allotrope.models.shared.definitions.custom import (
+    TQuantityValueDegreeCelsius,
+    TQuantityValuePercent,
+)
 from allotropy.allotrope.schema_mappers.adm.plate_reader.rec._2024._06.plate_reader import (
     Measurement,
     MeasurementGroup,
@@ -47,7 +52,9 @@
 from allotropy.parsers.utils.values import (
     assert_not_none,
     num_to_chars,
+    quantity_or_none,
 )
+from allotropy.parsers.utils.warnings_tools import suppress_unused_keys_warning
 
 
 class ReadType(Enum):
@@ -104,6 +111,7 @@ def get_plate_number_and_barcode(data: SeriesData) -> tuple[str, str]:
 class CalculatedPlateInfo(PlateInfo):
     formula: str
     name: str
+    custom_info: dict[str, Any] | None = None
 
     @staticmethod
     def create(data: SeriesData) -> CalculatedPlateInfo:
@@ -126,6 +134,7 @@ def create(data: SeriesData) -> CalculatedPlateInfo:
             ),
             formula=formula,
             name=name.strip(),
+            custom_info=data.get_unread(),
         )
 
 
@@ -134,6 +143,9 @@ class ResultPlateInfo(PlateInfo):
     label: str
     measinfo: str
     emission_filter_id: str
+    sample_custom_info: dict[str, Any]
+    device_control_custom_info: dict[str, Any]
+    custom_info: dict[str, Any]
 
     @staticmethod
     def create(data: SeriesData) -> ResultPlateInfo | None:
@@ -159,6 +171,42 @@ def create(data: SeriesData) -> ResultPlateInfo | None:
                 search("De=(...)", measinfo),
                 msg=f"Unable to find emission filter ID for Plate {barcode}.",
             ).group(1),
+            device_control_custom_info={
+                "Ambient temperature at start": quantity_or_none(
+                    TQuantityValueDegreeCelsius,
+                    data.get(float, "Ambient temperature at start"),
+                ),
+                "Ambient temperature at end": quantity_or_none(
+                    TQuantityValueDegreeCelsius,
+                    data.get(float, "Ambient temperature at end"),
+                ),
+                "Chamber temperature at end": quantity_or_none(
+                    TQuantityValueDegreeCelsius,
+                    data.get(float, "Chamber temperature at end"),
+                ),
+                "Humidity at start": quantity_or_none(
+                    TQuantityValuePercent, data.get(float, "Humidity at start")
+                ),
+                "Humidity at end": quantity_or_none(
+                    TQuantityValuePercent, data.get(float, "Humidity at end")
+                ),
+                "Kinetics": data.get(float, "Kinetics"),
+                "ScanX": data.get(float, "ScanX"),
+                "ScanY": data.get(float, "ScanY"),
+                "Inside temperature at start": quantity_or_none(
+                    TQuantityValueDegreeCelsius,
+                    data.get(float, "Inside temperature at start"),
+                ),
+                "Inside temperature at end": quantity_or_none(
+                    TQuantityValueDegreeCelsius,
+                    data.get(float, "Inside temperature at end"),
+                ),
+            },
+            sample_custom_info={
+                "group identifier": data.get(float, "Group"),
+                "Repeat": data.get(float, "Repeat"),
+            },
+            custom_info=data.get_unread(),
         )
 
     def match(self, background_info: BackgroundInfo) -> bool:
@@ -176,13 +224,15 @@ class BackgroundInfo:
     plate_num: str
     label: str
     measinfo: str
+    custom_info: dict[str, Any]
 
     @staticmethod
     def create(data: SeriesData) -> BackgroundInfo:
         return BackgroundInfo(
             plate_num=data[str, "Plate"],
             label=data[str, "Label"],
             measinfo=data[str, "MeasInfo"],
+            custom_info=data.get_unread(),
         )
 
 
@@ -288,6 +338,7 @@ class ResultPlate(Plate):
 class CalculatedPlate(Plate):
     plate_info: CalculatedPlateInfo
     results: list[CalculatedResult]
+    custom_info: dict[str, Any] | None = None
 
     def get_source_results(self, plate_list: PlateList) -> Iterator[list[Result]]:
         if not self.background_infos:
@@ -327,6 +378,7 @@ def create_plate(reader: CsvReader) -> ResultPlate | CalculatedPlate:
             plate_info=CalculatedPlateInfo.create(series),
             background_infos=create_background_infos(reader),
             results=create_calculated_results(reader),
+            custom_info=series.get_unread(),
         )
 
 
@@ -372,6 +424,7 @@ def get_measurement_time(self) -> str:
 class BasicAssayInfo:
     protocol_id: str | None
     assay_id: str | None
+    custom_info: dict[str, Any]
 
     @staticmethod
     def create(reader: CsvReader) -> BasicAssayInfo:
@@ -384,12 +437,17 @@ def create(reader: CsvReader) -> BasicAssayInfo:
             data_frame.columns.astype(str).str.replace(":", "").str.strip()
         )
         data = df_to_series_data(data_frame)
-        return BasicAssayInfo(data.get(str, "Protocol ID"), data.get(str, "Assay ID"))
+        return BasicAssayInfo(
+            data.get(str, "Protocol ID"),
+            data.get(str, "Assay ID"),
+            data.get_unread(skip={"Serial#"}),
+        )
 
 
 @dataclass(frozen=True)
 class PlateType:
     number_of_wells: float
+    custom_info: dict[str, Any]
 
     @staticmethod
     def create(reader: CsvReader) -> PlateType:
@@ -399,7 +457,8 @@ def create(reader: CsvReader) -> PlateType:
         )
         data = df_to_series_data(data_frame.T)
         return PlateType(
-            number_of_wells=data[float, "Number of the wells in the plate"]
+            number_of_wells=data[float, "Number of the wells in the plate"],
+            custom_info=data.get_unread(),
         )
 
 
@@ -510,6 +569,7 @@ class Filter:
     bandwidth: float | None = None
 
     @staticmethod
+    @suppress_unused_keys_warning
     def create(reader: CsvReader) -> Filter | None:
         if not reader.current_line_exists() or reader.match(
             "(^Mirror modules)|(^Instrument:)|(^Aperture:)"
@@ -569,6 +629,7 @@ class Labels:
     detector_gain_setting: str | None = None
 
     @staticmethod
+    @suppress_unused_keys_warning
     def create(reader: CsvReader) -> Labels:
         reader.drop_until_inclusive("^Labels")
         data_frame = assert_not_none(
@@ -676,7 +737,7 @@ class Data:
     software: Software
     plate_list: PlateList
     basic_assay_info: BasicAssayInfo
-    number_of_wells: float
+    plate_type_info: PlateType
     plate_maps: dict[str, PlateMap]
     labels: Labels
     instrument: Instrument
@@ -686,7 +747,7 @@ def create(reader: CsvReader) -> Data:
         return Data(
             plate_list=PlateList.create(reader),
             basic_assay_info=BasicAssayInfo.create(reader),
-            number_of_wells=PlateType.create(reader).number_of_wells,
+            plate_type_info=PlateType.create(reader),
             plate_maps=create_plate_maps(reader),
             labels=Labels.create(reader),
             instrument=Instrument.create(reader),
@@ -714,14 +775,23 @@ def create_metadata(
 def _create_measurement(
     plate_info: ResultPlateInfo,
     result: Result,
-    plate_maps: dict[str, PlateMap],
-    labels: Labels,
+    data: Data,
     read_type: ReadType,
 ) -> Measurement:
     plate_barcode = plate_info.barcode
     well_location = f"{result.col}{result.row}"
-    ex_filter = labels.excitation_filter
-    em_filter = labels.get_emission_filter(plate_info.emission_filter_id)
+    ex_filter = data.labels.excitation_filter
+    em_filter = data.labels.get_emission_filter(plate_info.emission_filter_id)
+    sample_custom_info = {
+        **plate_info.sample_custom_info,
+        "Name of the plate type": data.plate_type_info.custom_info.pop(
+            "Name of the plate type", None
+        ),
+    }
+    device_control_custom_info = {
+        **plate_info.device_control_custom_info,
+        **data.plate_type_info.custom_info,
+    }
     return Measurement(
         type_=read_type.measurement_type,
         device_type=read_type.device_type,
@@ -731,21 +801,24 @@ def _create_measurement(
         location_identifier=well_location,
         sample_role_type=(
             p_map.get_sample_role_type(result.col, result.row)
-            if (p_map := plate_maps.get(plate_info.number))
+            if (p_map := data.plate_maps.get(plate_info.number))
             else None
         ),
         compartment_temperature=plate_info.chamber_temperature_at_start,
         absorbance=result.value if read_type is ReadType.ABSORBANCE else None,
         fluorescence=result.value if read_type is ReadType.FLUORESCENCE else None,
         luminescence=result.value if read_type is ReadType.LUMINESCENCE else None,
         detector_distance_setting=plate_info.measured_height,
-        number_of_averages=labels.number_of_flashes,
-        detector_gain_setting=labels.detector_gain_setting,
-        scan_position_setting=labels.scan_position_setting,
+        number_of_averages=data.labels.number_of_flashes,
+        detector_gain_setting=data.labels.detector_gain_setting,
+        scan_position_setting=data.labels.scan_position_setting,
         detector_wavelength_setting=em_filter.wavelength if em_filter else None,
         detector_bandwidth_setting=em_filter.bandwidth if em_filter else None,
         excitation_wavelength_setting=ex_filter.wavelength if ex_filter else None,
         excitation_bandwidth_setting=ex_filter.bandwidth if ex_filter else None,
+        device_control_custom_info=device_control_custom_info,
+        sample_custom_info=sample_custom_info,
+        measurement_custom_info=plate_info.custom_info,
     )
 
 
@@ -757,8 +830,7 @@ def create_measurement_groups(data: Data) -> list[MeasurementGroup]:
             measurement = _create_measurement(
                 plate.plate_info,
                 result,
-                data.plate_maps,
-                data.labels,
+                data,
                 read_type,
             )
             well_loc_measurements[
@@ -769,10 +841,11 @@ def create_measurement_groups(data: Data) -> list[MeasurementGroup]:
     return [
         MeasurementGroup(
             measurement_time=measurement_time,
-            plate_well_count=data.number_of_wells,
+            plate_well_count=data.plate_type_info.number_of_wells,
             analytical_method_identifier=data.basic_assay_info.protocol_id,
             experimental_data_identifier=data.basic_assay_info.assay_id,
             measurements=well_loc_measurements[well_location],
+            custom_info=data.basic_assay_info.custom_info,
         )
         for well_location in sorted(
             well_loc_measurements.keys(),

src/allotropy/parsers/utils/calculated_data_documents/definition.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Iterator
 from dataclasses import dataclass
+from typing import Any
 
 
 @dataclass
@@ -24,6 +25,7 @@ class CalculatedDocument(Referenceable):
     unit: str | None = None
     description: str | None = None
     iterated: bool = False
+    custom_info: dict[str, Any] | None = None
 
     def iter_struct(self) -> Iterator[CalculatedDocument]:
         if self.iterated: