feat: Migrate cytiva_unicorn to use get_unread (#1051)

felipenarv · web-flow · commit cf67c758b2b7 · 2025-08-15T10:25:56.000-04:00
Code for [BNCH-115083](https://jira.benchling.team/browse/BNCH-115083)
diff --git a/src/allotropy/allotrope/schema_mappers/adm/liquid_chromatography/benchling/_2023/_09/liquid_chromatography.py b/src/allotropy/allotrope/schema_mappers/adm/liquid_chromatography/benchling/_2023/_09/liquid_chromatography.py
@@ -146,6 +146,7 @@ class Fraction:
     field_type: str | None = None
     retention_time: float | None = None
     retention_volume: float | None = None
+    custom_info: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)
@@ -155,6 +156,7 @@ class Log:
     log_entry: str | None = None
     retention_time: float | None = None
     retention_volume: float | None = None
+    custom_info: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)
@@ -236,6 +238,7 @@ class Measurement:
     injection_custom_info: dict[str, Any] | None = None
     column_custom_info: dict[str, Any] | None = None
     measurement_custom_info: dict[str, Any] | None = None
+    processed_data_custom_info: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)
@@ -361,11 +364,9 @@ def _get_measurement_document_item(
                         for device_control_doc in measurement.device_control_docs
                     ]
                 ),
-                chromatogram_data_cube=(
-                    get_data_cube(
-                        measurement.chromatogram_data_cube,
-                        ChromatogramDataCube,
-                    )
+                chromatogram_data_cube=get_data_cube(
+                    measurement.chromatogram_data_cube,
+                    ChromatogramDataCube,
                 ),
             ),
             measurement.measurement_custom_info,
@@ -547,17 +548,27 @@ def build_data_document_item(
                 if pdd and get_data_processing_documents(pdd)
                 else None
             )
-            item = ProcessedDataDocumentItem(
-                chromatogram_data_cube=get_data_cube(
-                    measurement.processed_data_chromatogram_data_cube, TDatacube
-                ),
-                derived_column_pressure_data_cube=get_data_cube(
-                    measurement.derived_column_pressure_data_cube,
-                    DerivedColumnPressureDataCube,
+            item = add_custom_information_document(
+                ProcessedDataDocumentItem(
+                    chromatogram_data_cube=add_custom_information_document(
+                        get_data_cube(
+                            measurement.processed_data_chromatogram_data_cube, TDatacube
+                        ),
+                        getattr(
+                            measurement.processed_data_chromatogram_data_cube,
+                            "custom_info",
+                            None,
+                        ),
+                    ),
+                    derived_column_pressure_data_cube=get_data_cube(
+                        measurement.derived_column_pressure_data_cube,
+                        DerivedColumnPressureDataCube,
+                    ),
+                    processed_data_identifier=measurement.processed_data_identifier,
+                    peak_list=build_peak_list(),
+                    data_processing_aggregate_document=data_processing,
                 ),
-                processed_data_identifier=measurement.processed_data_identifier,
-                peak_list=build_peak_list(),
-                data_processing_aggregate_document=data_processing,
+                measurement.processed_data_custom_info,
             )
             return (
                 add_custom_information_document(item, pdd.custom_info) if pdd else item
@@ -670,7 +681,7 @@ def _get_fraction_aggregate_document(
         )
 
     def _get_fraction_document(self, fraction_doc: Fraction) -> FractionDocumentItem:
-        return FractionDocumentItem(
+        fraction_doc_item = FractionDocumentItem(
             index=fraction_doc.index,
             fraction_role=fraction_doc.fraction_role,
             field_type=fraction_doc.field_type,
@@ -682,6 +693,10 @@ def _get_fraction_document(self, fraction_doc: Fraction) -> FractionDocumentItem
             ),
         )
 
+        return add_custom_information_document(
+            fraction_doc_item, fraction_doc.custom_info
+        )
+
     def _get_log_aggregate_document(
         self, logs: list[Log] | None
     ) -> LogAggregateDocument | None:
@@ -693,7 +708,7 @@ def _get_log_aggregate_document(
         )
 
     def _get_log_document(self, log_doc: Log) -> LogDocumentItem:
-        return LogDocumentItem(
+        log_doc_item = LogDocumentItem(
             index=log_doc.index,
             method_identifier=log_doc.method_identifier,
             log_entry=log_doc.log_entry,
@@ -704,3 +719,4 @@ def _get_log_document(self, log_doc: Log) -> LogDocumentItem:
                 TQuantityValueMilliliter, log_doc.retention_volume
             ),
         )
+        return add_custom_information_document(log_doc_item, log_doc.custom_info)
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurement_group.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurement_group.py
@@ -63,14 +63,34 @@ def create_measurement_groups(
         TemperatureMeasurement.create_or_none(handler, elements, static_docs),
     ]
 
-    return [
+    measurement_group = [
         MeasurementGroup(
             measurements=[measurement for measurement in measurements if measurement],
             fractions=create_fractions(event_curves) if event_curves else None,
             logs=create_logs(event_curves) if event_curves else None,
+            measurement_aggregate_custom_info={},
         )
     ]
 
+    chrom_1.mark_read({"element:TimeUnit", "element:VolumeUnit", "element:IsReadonly"})
+    custom_info = chrom_1.get_unread()
+
+    if measurement_group[0].measurement_aggregate_custom_info is not None:
+        custom_info.pop("UNICORNVersion")
+        custom_info.update(
+            {
+                "RunIndex": results.get_sub_text_or_none("RunIndex"),
+                "RunType": results.get_sub_text_or_none("RunType"),
+                "Name": results.get_sub_text_or_none("Name"),
+            }
+        )
+        custom_info_sorted = dict(sorted(custom_info.items()))
+        measurement_group[0].measurement_aggregate_custom_info.update(
+            custom_info_sorted
+        )
+
+    return measurement_group
+
 
 def create_fractions(event_curves: StrictXmlElement) -> list[Fraction]:
     event_curve_fraction = None
@@ -94,6 +114,7 @@ def create_fractions(event_curves: StrictXmlElement) -> list[Fraction]:
                     else t_min * 60
                 ),
                 retention_volume=event.get_sub_float_or_none("EventVolume"),
+                custom_info=event.get_unread(),
             )
             for idx, event in enumerate(events.findall("Event"), start=1)
             if event.get_attr_or_none("EventType") in ["Fraction", "Method"]
@@ -144,6 +165,7 @@ def create_logs(event_curves: StrictXmlElement) -> list[Log]:
                         else t_min * 60
                     ),
                     retention_volume=event.get_sub_float_or_none("EventVolume"),
+                    custom_info=event.get_unread(),
                 )
             )
 
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/absorbance.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/absorbance.py
@@ -57,7 +57,7 @@ def create_or_none(
         elements: list[StrictXmlElement],
         static_docs: StaticDocs,
     ) -> UnicornMeasurement:
-        return cls.get_measurement(
+        measurement = cls.get_measurement(
             static_docs=static_docs,
             chromatogram_data_cube=assert_not_none(
                 cls.get_data_cube_or_none(
@@ -75,10 +75,15 @@ def create_or_none(
                 DeviceControlDoc(
                     device_type=DEVICE_TYPE,
                     start_time=static_docs.start_time,
+                    device_control_custom_info={},
                 )
             ],
             peaks=cls.get_peaks(handler),
         )
+        cls.add_custom_info(
+            measurement, cls.filter_curve_or_none(elements, cls.get_curve_regex())
+        )
+        return measurement
 
 
 class AbsorbanceMeasurement1(AbsorbanceMeasurement):
@@ -150,8 +155,9 @@ def get_peaks_custom_info(
     def get_peaks(cls, handler: UnicornZipHandler) -> list[Peak]:
         chrom_1 = handler.get_chrom_1()
         peaks = chrom_1.recursive_find_or_none(["PeakTables", "PeakTable", "Peaks"])
-        return [
-            Peak(
+        output = []
+        for idx, peak in enumerate(peaks.findall("Peak") if peaks else [], start=1):
+            peak_obj = Peak(
                 identifier=random_uuid_str(),
                 index=f"Peak {idx}",
                 end=peak.get_sub_float_or_none("EndPeakRetention"),
@@ -172,8 +178,10 @@ def get_peaks(cls, handler: UnicornZipHandler) -> list[Peak]:
                 chromatographic_asymmetry=peak.get_sub_float_or_none("Assymetry"),
                 custom_info=cls.get_peaks_custom_info(peak),
             )
-            for idx, peak in enumerate(peaks.findall("Peak") if peaks else [], start=1)
-        ]
+            if peak_obj.custom_info is not None:
+                peak_obj.custom_info.update(peak.get_unread())
+            output.append(peak_obj)
+        return output
 
 
 class AbsorbanceMeasurement2(AbsorbanceMeasurement):
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/concentration.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/concentration.py
@@ -43,9 +43,13 @@ def create_or_none(
                             unit="%",
                         ),
                     ),
+                    device_control_custom_info={},
                 ),
             ],
         )
+        cls.add_custom_info(
+            measurement, cls.filter_curve_or_none(elements, r"^Conc B$")
+        )
         return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/conductivity.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/conductivity.py
@@ -56,9 +56,11 @@ def create_or_none(
                 DeviceControlDoc(
                     device_type=DEVICE_TYPE,
                     start_time=static_docs.start_time,
+                    device_control_custom_info={},
                 )
             ],
         )
+        cls.add_custom_info(measurement, cls.filter_curve_or_none(elements, r"^Cond$"))
         return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/flow.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/flow.py
@@ -36,6 +36,7 @@ def create_device_control(
             start_time=start_time,
             sample_flow_data_cube=sample_flow_data_cube,
             system_flow_data_cube=system_flow_data_cube,
+            device_control_custom_info={},
         )
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/generic.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/generic.py
@@ -27,6 +27,66 @@
 
 
 class UnicornMeasurement(Measurement):
+    @classmethod
+    def add_custom_info(
+        cls, measurement: Measurement, base_element: StrictXmlElement | None
+    ) -> None:
+        if base_element:
+            device_control_custom_info = {
+                "ColumnVolume": base_element.get_sub_text_or_none("ColumnVolume"),
+                "ScanInterval": base_element.get_sub_text_or_none("ScanInterval"),
+            }
+
+            processed_data_custom_info = {
+                "AmplitudePrecision": base_element.get_sub_text_or_none(
+                    "AmplitudePrecision"
+                ),
+                "AmplitudeUnit": base_element.get_sub_text_or_none("AmplitudeUnit"),
+                "ChromatogramID": base_element.get_sub_text_or_none("ChromatogramID"),
+                "ChromatogramName": base_element.get_sub_text_or_none(
+                    "ChromatogramName"
+                ),
+                "ChromatogramStartTime": base_element.get_sub_text_or_none(
+                    "ChromatogramStartTime"
+                ),
+                "ChromatogramStartTimeUtcOffsetMinutes": base_element.get_sub_text_or_none(
+                    "ChromatogramStartTimeUtcOffsetMinutes"
+                ),
+                "ColumnVolumeUnitName": base_element.get_sub_text_or_none(
+                    "ColumnVolumeUnitName"
+                ),
+                "CurveDataType": base_element.get_sub_text_or_none("CurveDataType"),
+                "CurveNumber": base_element.get_sub_text_or_none("CurveNumber"),
+                "DistanceBetweenPoints": base_element.get_sub_text_or_none(
+                    "DistanceBetweenPoints"
+                ),
+                "DistancetoStartPoints": base_element.get_sub_text_or_none(
+                    "DistancetoStartPoints"
+                ),
+                "IsExternal": base_element.get_sub_text_or_none("IsExternal"),
+                "IsOriginalData": base_element.get_sub_text_or_none("IsOriginalData"),
+                "IsReadOnly": base_element.get_sub_text_or_none("IsReadOnly"),
+                "IsoChroneType": base_element.get_sub_text_or_none("IsoChroneType"),
+                "MethodStartTime": base_element.get_sub_text_or_none("MethodStartTime"),
+                "MethodStartTimeUtcOffsetMinutes": base_element.get_sub_text_or_none(
+                    "MethodStartTimeUtcOffsetMinutes"
+                ),
+                "TimeUnit": base_element.get_sub_text_or_none("TimeUnit"),
+                "VolumeUnit": base_element.get_sub_text_or_none("VolumeUnit"),
+            }
+            if (
+                measurement.device_control_docs[0].device_control_custom_info
+                is not None
+            ):
+                measurement.device_control_docs[0].device_control_custom_info.update(
+                    device_control_custom_info
+                )
+            if measurement.processed_data_custom_info is not None:
+                processed_data_custom_info.update(base_element.get_unread())
+                measurement.processed_data_custom_info.update(
+                    processed_data_custom_info
+                )
+
     @classmethod
     def filter_curve_or_none(
         cls, curve_elements: list[StrictXmlElement], pattern: str
@@ -61,13 +121,15 @@ def get_data_cube_or_none(
 
         if data_cube_handler := cls.get_data_cube_handler_or_none(handler, curve):
             name_element = curve.find("Name")
+
             if name := name_element.get_text_or_none():
-                return create_data_cube(
+                data_cube = create_data_cube(
                     data_cube_handler,
                     name,
                     data_cube_component,
                     transformation,
                 )
+                return data_cube
         return None
 
     @classmethod
@@ -104,6 +166,7 @@ def get_measurement(
                 "sample_volume_3": static_docs.sample_volume_3,
             },
             peaks=peaks,
+            processed_data_custom_info={},
         )
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/ph.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/ph.py
@@ -43,9 +43,11 @@ def create_or_none(
                 DeviceControlDoc(
                     device_type=DEVICE_TYPE,
                     start_time=static_docs.start_time,
+                    device_control_custom_info={},
                 )
             ],
         )
+        cls.add_custom_info(measurement, cls.filter_curve_or_none(elements, r"^pH$"))
         return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/pressure.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/pressure.py
@@ -79,9 +79,13 @@ def create_or_none(
                             unit="MPa",
                         ),
                     ),
+                    device_control_custom_info={},
                 ),
             ],
         )
+        cls.add_custom_info(
+            measurement, cls.filter_curve_or_none(elements, r"^DeltaC pressure$")
+        )
         return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/measurements/temperature.py b/src/allotropy/parsers/cytiva_unicorn/structure/measurements/temperature.py
@@ -43,9 +43,13 @@ def create_or_none(
                             unit="degC",
                         ),
                     ),
+                    device_control_custom_info={},
                 ),
             ],
         )
+        cls.add_custom_info(
+            measurement, cls.filter_curve_or_none(elements, r"^Cond temp$")
+        )
         return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None
 
     @classmethod
diff --git a/src/allotropy/parsers/cytiva_unicorn/structure/metadata.py b/src/allotropy/parsers/cytiva_unicorn/structure/metadata.py
diff --git a/src/allotropy/parsers/utils/strict_xml_element.py b/src/allotropy/parsers/utils/strict_xml_element.py
diff --git a/tests/parsers/cytiva_unicorn/testdata/unicorn_1.json b/tests/parsers/cytiva_unicorn/testdata/unicorn_1.json

Original file line number	Diff line number	Diff line change
`@@ -56,9 +56,11 @@ def create_or_none(`
`56`	`56`	`DeviceControlDoc(`
`57`	`57`	`device_type=DEVICE_TYPE,`
`58`	`58`	`start_time=static_docs.start_time,`
	`59`	`+ device_control_custom_info={},`
`59`	`60`	`)`
`60`	`61`	`],`
`61`	`62`	`)`
	`63`	`+ cls.add_custom_info(measurement, cls.filter_curve_or_none(elements, r"^Cond$"))`
`62`	`64`	`return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None`
`63`	`65`
`64`	`66`	`@classmethod`
Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,7 @@ def create_device_control(`
`36`	`36`	`start_time=start_time,`
`37`	`37`	`sample_flow_data_cube=sample_flow_data_cube,`
`38`	`38`	`system_flow_data_cube=system_flow_data_cube,`
	`39`	`+ device_control_custom_info={},`
`39`	`40`	`)`
`40`	`41`
`41`	`42`	`@classmethod`
Original file line number	Diff line number	Diff line change
`@@ -43,9 +43,11 @@ def create_or_none(`
`43`	`43`	`DeviceControlDoc(`
`44`	`44`	`device_type=DEVICE_TYPE,`
`45`	`45`	`start_time=static_docs.start_time,`
	`46`	`+ device_control_custom_info={},`
`46`	`47`	`)`
`47`	`48`	`],`
`48`	`49`	`)`
	`50`	`+ cls.add_custom_info(measurement, cls.filter_curve_or_none(elements, r"^pH$"))`
`49`	`51`	`return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None`
`50`	`52`
`51`	`53`	`@classmethod`