Skip to content

chore: Cytiva Unicorn - silence the remaining get_unread warnings and clean up logic a bit #1056

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nathan-stender merged 4 commits into from
Aug 20, 2025
Merged

chore: Cytiva Unicorn - silence the remaining get_unread warnings and clean up logic a bit #1056

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a5c3251
No warnings
nathan-stender Aug 15, 2025
51a8d79
Clean up
nathan-stender Aug 15, 2025
26e80ba
More cleanup
nathan-stender Aug 15, 2025
4be8c3b
Merge branch 'main' into nstender-finish-unicorn-unread
nathan-stender Aug 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 12 additions & 4 deletions src/allotropy/parsers/cytiva_unicorn/reader/unicorn_zip_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,9 @@ def get_system_data(self) -> StrictXmlElement:
system_data = self.get_zip_from_pattern("SystemData(.zip)?$")
b_stream = system_data.get_file_from_pattern("^Xml$")
raw_content = self.filter_xml_metadata(b_stream).read()
return StrictXmlElement.create_from_bytes(raw_content)
return StrictXmlElement.create_from_bytes(
raw_content, mark_read={"attr:FormatVersion", "attr:UNICORNVersion"}
)

def get_results(self) -> StrictXmlElement:
b_stream = self.get_file_from_pattern("Result.xml$")
Expand All @@ -77,12 +79,16 @@ def get_instrument_config_data(self) -> StrictXmlElement:
instrument_config_data = self.get_zip_from_pattern(instrument_regex)
b_stream = instrument_config_data.get_file_from_pattern("^Xml$")
raw_content = self.filter_xml_metadata(b_stream).read()
return StrictXmlElement.create_from_bytes(raw_content)
return StrictXmlElement.create_from_bytes(
raw_content, mark_read={"attr:FormatVersion", "attr:UNICORNVersion"}
)

def get_evaluation_log(self) -> StrictXmlElement:
b_stream = self.get_file_from_pattern("EvaluationLog.xml$")
raw_content = self.filter_xml_metadata(b_stream).read()
return StrictXmlElement.create_from_bytes(raw_content)
return StrictXmlElement.create_from_bytes(
raw_content, mark_read={"attr:FormatVersion", "attr:UNICORNVersion"}
)

def get_chrom_1(self) -> StrictXmlElement:
b_stream = self.get_file_from_pattern("Chrom.1.Xml$")
Expand All @@ -93,4 +99,6 @@ def get_column_type_data(self) -> StrictXmlElement:
column_type_data = self.get_zip_from_pattern("ColumnTypeData(.zip)?$")
b_stream = column_type_data.get_file_from_pattern("^Xml$")
raw_content = self.filter_xml_metadata(b_stream).read()
return StrictXmlElement.create_from_bytes(raw_content)
return StrictXmlElement.create_from_bytes(
raw_content, mark_read={"attr:FormatVersion", "attr:UNICORNVersion"}
)
38 changes: 19 additions & 19 deletions src/allotropy/parsers/cytiva_unicorn/structure/measurement_group.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,34 +62,32 @@ def create_measurement_groups(
FlowMeasurement.create_or_none(handler, elements, static_docs),
TemperatureMeasurement.create_or_none(handler, elements, static_docs),
]
# Some curves may not be read due not matching any supported measurement type, mark these as read.
for element in elements:
element.mark_read("attr:CurveDataType")

chrom_1.mark_read(
{
"element:TimeUnit",
"element:VolumeUnit",
"element:IsReadonly",
"attr:UNICORNVersion",
}
)

measurement_group = [
return [
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(
{
measurement_aggregate_custom_info={
"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
**dict(sorted(chrom_1.get_unread().items())),
},
)

return measurement_group
]


def create_fractions(event_curves: StrictXmlElement) -> list[Fraction]:
Expand Down Expand Up @@ -142,10 +140,12 @@ def create_logs(event_curves: StrictXmlElement) -> list[Log]:
"Method",
"Manual",
]:
event.get_unread()
continue

event_text = event.get_sub_text_or_none("EventText")
if event_text is None:
event.get_unread()
continue

event_subtype = event.get_attr_or_none("EventSubType")
Expand Down
15 changes: 10 additions & 5 deletions src/allotropy/parsers/cytiva_unicorn/structure/measurements/absorbance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,16 @@ def create_or_none(
elements: list[StrictXmlElement],
static_docs: StaticDocs,
) -> UnicornMeasurement:
element = assert_not_none(
cls.filter_curve_or_none(elements, cls.get_curve_regex()),
"Unable to find curve data for absorbance measurement.",
)
measurement = cls.get_measurement(
static_docs=static_docs,
chromatogram_data_cube=assert_not_none(
cls.get_data_cube_or_none(
handler,
cls.filter_curve_or_none(elements, cls.get_curve_regex()),
element,
DataCubeComponent(
type_=FieldComponentDatatype.float,
concept="absorbance",
Expand All @@ -75,13 +79,13 @@ def create_or_none(
DeviceControlDoc(
device_type=DEVICE_TYPE,
start_time=static_docs.start_time,
device_control_custom_info={},
device_control_custom_info=cls.get_device_control_custom_info(
element
),
)
],
peaks=cls.get_peaks(handler),
)
cls.add_custom_info(
measurement, cls.filter_curve_or_none(elements, cls.get_curve_regex())
processed_data_custom_info=cls.get_processed_data_custom_info(element),
)
return measurement

Expand Down Expand Up @@ -154,6 +158,7 @@ def get_peaks_custom_info(
@classmethod
def get_peaks(cls, handler: UnicornZipHandler) -> list[Peak]:
chrom_1 = handler.get_chrom_1()
chrom_1.get_unread() # chrom_1 data is read in create_measurement_groups
peaks = chrom_1.recursive_find_or_none(["PeakTables", "PeakTable", "Peaks"])
output = []
for idx, peak in enumerate(peaks.findall("Peak") if peaks else [], start=1):
Expand Down
11 changes: 6 additions & 5 deletions src/allotropy/parsers/cytiva_unicorn/structure/measurements/concentration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ def create_or_none(
elements: list[StrictXmlElement],
static_docs: StaticDocs,
) -> UnicornMeasurement | None:
element = cls.filter_curve_or_none(elements, r"^Conc B$")
measurement = cls.get_measurement(
static_docs=static_docs,
device_control_docs=[
Expand All @@ -36,19 +37,19 @@ def create_or_none(
start_time=static_docs.start_time,
solvent_conc_data_cube=cls.get_data_cube_or_none(
handler,
cls.filter_curve_or_none(elements, r"^Conc B$"),
element,
DataCubeComponent(
type_=FieldComponentDatatype.float,
concept="solvent concentration",
unit="%",
),
),
device_control_custom_info={},
device_control_custom_info=cls.get_device_control_custom_info(
element
),
),
],
)
cls.add_custom_info(
measurement, cls.filter_curve_or_none(elements, r"^Conc B$")
processed_data_custom_info=cls.get_processed_data_custom_info(element),
)
return measurement if cls.is_valid(cls.get_data_cubes(measurement)) else None

Expand Down
9 changes: 6 additions & 3 deletions src/allotropy/parsers/cytiva_unicorn/structure/measurements/conductivity.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,12 @@ def create_or_none(
elements: list[StrictXmlElement],
static_docs: StaticDocs,
) -> UnicornMeasurement | None:
element = cls.filter_curve_or_none(elements, r"^Cond$")
measurement = cls.get_measurement(
static_docs=static_docs,
chromatogram_data_cube=cls.get_data_cube_or_none(
handler,
cls.filter_curve_or_none(elements, r"^Cond$"),
element,
DataCubeComponent(
type_=FieldComponentDatatype.float,
concept="electric conductivity",
Expand All @@ -56,11 +57,13 @@ def create_or_none(
DeviceControlDoc(
device_type=DEVICE_TYPE,
start_time=static_docs.start_time,
device_control_custom_info={},
device_control_custom_info=cls.get_device_control_custom_info(
element
),
)
],
processed_data_custom_info=cls.get_processed_data_custom_info(element),
)
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
Expand Down
1 change: 0 additions & 1 deletion src/allotropy/parsers/cytiva_unicorn/structure/measurements/flow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ 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
Expand Down
112 changes: 54 additions & 58 deletions src/allotropy/parsers/cytiva_unicorn/structure/measurements/generic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

from re import search
from typing import Any

from allotropy.allotrope.schema_mappers.adm.liquid_chromatography.benchling._2023._09.liquid_chromatography import (
DeviceControlDoc,
Expand Down Expand Up @@ -28,64 +29,58 @@

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"),
}
def get_device_control_custom_info(
cls, base_element: StrictXmlElement | None
) -> dict[str, Any]:
if not base_element:
return {}
return {
"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 get_processed_data_custom_info(
cls, base_element: StrictXmlElement | None
) -> dict[str, Any]:
if not base_element:
return {}
return {
"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_attr_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"),
**base_element.get_unread(),
}

@classmethod
def filter_curve_or_none(
Expand Down Expand Up @@ -141,6 +136,7 @@ def get_measurement(
processed_data_chromatogram_data_cube: DataCube | None = None,
derived_column_pressure_data_cube: DataCube | None = None,
peaks: list[Peak] | None = None,
processed_data_custom_info: dict[str, Any] | None = None,
) -> UnicornMeasurement:
return UnicornMeasurement(
measurement_identifier=random_uuid_str(),
Expand All @@ -166,7 +162,7 @@ def get_measurement(
"sample_volume_3": static_docs.sample_volume_3,
},
peaks=peaks,
processed_data_custom_info={},
processed_data_custom_info=processed_data_custom_info,
)

@classmethod
Expand Down
9 changes: 6 additions & 3 deletions src/allotropy/parsers/cytiva_unicorn/structure/measurements/ph.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,11 +28,12 @@ def create_or_none(
elements: list[StrictXmlElement],
static_docs: StaticDocs,
) -> UnicornMeasurement | None:
element = cls.filter_curve_or_none(elements, r"^pH$")
measurement = cls.get_measurement(
static_docs=static_docs,
chromatogram_data_cube=cls.get_data_cube_or_none(
handler,
cls.filter_curve_or_none(elements, r"^pH$"),
element,
DataCubeComponent(
type_=FieldComponentDatatype.float,
concept="pH",
Expand All @@ -43,11 +44,13 @@ def create_or_none(
DeviceControlDoc(
device_type=DEVICE_TYPE,
start_time=static_docs.start_time,
device_control_custom_info={},
device_control_custom_info=cls.get_device_control_custom_info(
element
),
)
],
processed_data_custom_info=cls.get_processed_data_custom_info(element),
)
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
Expand Down
Loading
Loading