Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ class MeasurementType(Enum):
class DeviceDocument:
device_type: str
device_identifier: str
device_custom_info: DictType | None = None


@dataclass(frozen=True)
Expand All @@ -79,6 +80,7 @@ class Metadata:
lot_number: str | None = None
sensor_chip_custom_info: DictType | None = None
data_system_custom_info: DictType | None = None
device_system_custom_info: DictType | None = None


@dataclass(frozen=True)
Expand Down Expand Up @@ -172,23 +174,29 @@ def map_model(self, data: Data) -> Model:
),
data.metadata.data_system_custom_info,
),
device_system_document=DeviceSystemDocument(
device_identifier=data.metadata.device_identifier,
model_number=data.metadata.model_number,
brand_name=data.metadata.brand_name,
product_manufacturer=data.metadata.product_manufacturer,
equipment_serial_number=data.metadata.equipment_serial_number,
device_document=(
[
DeviceDocumentItem(
device_type=device_document_item.device_type,
device_identifier=device_document_item.device_identifier,
)
for device_document_item in data.metadata.device_document
]
if data.metadata.device_document
else None
device_system_document=add_custom_information_document(
DeviceSystemDocument(
device_identifier=data.metadata.device_identifier,
model_number=data.metadata.model_number,
brand_name=data.metadata.brand_name,
product_manufacturer=data.metadata.product_manufacturer,
equipment_serial_number=data.metadata.equipment_serial_number,
device_document=(
[
add_custom_information_document(
DeviceDocumentItem(
device_type=device_document_item.device_type,
device_identifier=device_document_item.device_identifier,
),
custom_info_doc=device_document_item.device_custom_info,
)
for device_document_item in data.metadata.device_document
]
if data.metadata.device_document
else None
),
),
data.metadata.device_system_custom_info,
),
binding_affinity_analyzer_document=[
self._get_technique_document(measurement_group, data.metadata)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,16 +68,44 @@ def _get_sensorgram_datacube(sensorgram_data: pd.DataFrame) -> DataCube:


def _get_device_control_custom_info(data: Data) -> DictType:
custom_ifo: dict[str, Any] = {
application_template_details = data.application_template_details

custom_info: dict[str, Any] = {
"number of flow cells": data.chip_data.number_of_flow_cells,
"number of spots": data.chip_data.number_of_spots,
"buffer volume": quantity_or_none(
TQuantityValueMilliliter, data.run_metadata.buffer_volume
),
}
if detection_setting := data.run_metadata.detection_setting:
custom_ifo.update({detection_setting.key: detection_setting.value})
return custom_ifo
custom_info.update({detection_setting.key: detection_setting.value})

detection_info = application_template_details.get_nested("detection")
custom_info.update(
detection_info.get(key={"FlowCellSingle", "FlowCellDual", "FlowCellMulti"})
)

temp_info = application_template_details.get_nested("RackTemperature")
custom_info.update(
temp_info.get_keys_as_dict(
{
"minimum operating temperature": (float, "min", None),
"maximum operating temperature": (float, "max", None),
}
)
)

system_preparations = application_template_details.get_nested("system_preparations")
custom_info.update(
system_preparations.get_keys_as_dict(
{
"analysis temperature": (float, "AnalTemp", None),
"prime": (bool, "Prime", None),
"normalize": (bool, "Normalize", None),
}
)
)
return custom_info


def create_metadata(data: Data, named_file_contents: NamedFileContents) -> Metadata:
Expand All @@ -103,15 +131,19 @@ def create_metadata(data: Data, named_file_contents: NamedFileContents) -> Metad
lot_number=chip_data.lot_number,
sensor_chip_identifier=chip_data.sensor_chip_identifier,
device_document=[
DeviceDocument(device.type_, device.identifier)
DeviceDocument(device.type_, device.identifier, device.custom_info)
for device in run_metadata.devices
],
sensor_chip_custom_info=chip_data.custom_info,
data_system_custom_info=system_information.data_system_custom_info,
device_system_custom_info=system_information.device_system_custom_info,
)


def create_measurements(
measurements_data: list[MeasurementData], device_control_custom_info: DictType
measurements_data: list[MeasurementData],
device_control_custom_info: DictType,
data: Data,
) -> list[Measurement]:
return [
Measurement(
Expand All @@ -129,13 +161,25 @@ def create_measurements(
flow_rate=measurement.flow_rate,
contact_time=measurement.contact_time,
dilution=measurement.dilution,
device_control_custom_info=device_control_custom_info,
device_control_custom_info=dict(
device_control_custom_info
), # copy to avoid modifying the original
)
],
sample_custom_info={
**data.application_template_details.get_nested(
"racks"
).get_keys_as_dict(
{
"Rack1": (str, "_Rack1", None),
"Rack2": (str, "_Rack2", None),
"Lock Positions": (bool, "_LockPositions", None),
}
),
"molecular weight": quantity_or_none(
TQuantityValueDalton, measurement.molecular_weight
)
),
**measurement.sample_custom_info,
},
sensorgram_data_cube=_get_sensorgram_datacube(measurement.sensorgram_data),
report_point_data=(
Expand Down Expand Up @@ -168,7 +212,7 @@ def create_measurement_groups(data: Data) -> list[MeasurementGroup]:
MeasurementGroup(
measurement_time=system_information.measurement_time,
measurements=create_measurements(
measurements_data, device_control_custom_info
measurements_data, device_control_custom_info, data
),
experiment_type=system_information.experiment_type,
analytical_method_identifier=system_information.analytical_method_identifier,
Expand All @@ -181,6 +225,10 @@ def create_measurement_groups(data: Data) -> list[MeasurementGroup]:
TQuantityValueHertz,
try_float_or_none(data.run_metadata.data_collection_rate),
),
**data.application_template_details.get_nested(
"properties"
).get_keys_as_dict({"Run Type": (str, "TypeName", None)}),
Comment thread
james-leinas marked this conversation as resolved.
**system_information.measurement_aggregate_custom_info,
},
)
for measurements_data in data.sample_data.measurements.values()
Expand Down
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
import json
from pathlib import Path, PureWindowsPath
import warnings

from allotropy.allotrope.models.adm.binding_affinity_analyzer.wd._2024._12.binding_affinity_analyzer import (
Model,
)
Expand Down Expand Up @@ -30,9 +34,26 @@ class CytivaBiacoreT200ControlParser(VendorParser[MapperData, Model]):
SCHEMA_MAPPER = Mapper

def create_data(self, named_file_contents: NamedFileContents) -> MapperData:
data = Data.create(DictData(decode_data(named_file_contents)))
return MapperData(
base_data = DictData(decode_data(named_file_contents))
data = Data.create(base_data)
mapper_data = MapperData(
metadata=create_metadata(data, named_file_contents),
measurement_groups=create_measurement_groups(data),
calculated_data=create_calculated_data(data),
)
try:
unread = base_data.get_unread_deep()
original_path = named_file_contents.original_file_path
# Derive file stem robustly for both POSIX and Windows-style paths
stem = (
PureWindowsPath(original_path).stem
if "\\" in original_path and "/" not in original_path
else Path(original_path).stem
)
out_name = f"{stem}data_unread_2.json"
with open(out_name, "w", encoding="utf-8") as f:
json.dump(unread, f, ensure_ascii=False, indent=2)
except Exception as e:
# Best-effort debug artifact; do not break parsing on failure
warnings.warn(f"Failed to write unread debug file: {e!s}", stacklevel=1)
return mapper_data
Loading
Loading