feat: Thermo Fisher Scientific SkanIt - add unread data (#1104)

Author: joshua-benchling

src/allotropy/allotrope/schema_mappers/adm/plate_reader/rec/_2025/_03/plate_reader.py

@@ -220,7 +220,8 @@ class Metadata:
     equipment_serial_number: str | None = None
     product_manufacturer: str | None = None
     file_name: str | None = None
-    custom_info_doc: dict[str, Any] | None = None
+    metadata_custom_info: dict[str, Any] | None = None
+    device_control_custom_info: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)
@@ -253,7 +254,7 @@ def map_model(self, data: Data) -> Model:
                         ASM_converter_name=self.converter_name,
                         ASM_converter_version=ASM_CONVERTER_VERSION,
                     ),
-                    data.metadata.custom_info_doc,
+                    data.metadata.metadata_custom_info,
                 ),
                 plate_reader_document=[
                     self._get_technique_document(measurement_group)

src/allotropy/parsers/thermo_skanit/thermo_skanit_structure.py

@@ -80,22 +80,66 @@ def create_metadata(
         )
         general_info_data = ThermoSkanItMetadata._get_general_info_data(general_info_df)
         path = Path(file_path)
+
+        device_identifier = str(instrument_info_data.pop("device_identifier"))
+        model_number = str(instrument_info_data.pop("model_number"))
+
+        equipment_serial_number = None
+        if "equipment_serial_number" in instrument_info_data:
+            equipment_serial_number = str(
+                instrument_info_data.pop("equipment_serial_number")
+            )
+
+        software_name = None
+        if general_info_data.get("software_name") is not None:
+            software_name = str(general_info_data.pop("software_name"))
+        else:
+            general_info_data.pop("software_name", None)
+
+        software_version = None
+        if general_info_data.get("software_version") is not None:
+            software_version = str(general_info_data.pop("software_version"))
+        else:
+            general_info_data.pop("software_version", None)
+
+        # Merge custom_info dicts, ensuring we only unpack dict types
+        instrument_custom_info = instrument_info_data.get("custom_info", {})
+        device_control_custom_info_raw = instrument_info_data.get(
+            "device_control_custom_info"
+        )
+        general_custom_info = general_info_data.get("custom_info", {})
+
+        custom_info: dict[str, Any] = {}
+        if isinstance(instrument_custom_info, dict):
+            custom_info.update(instrument_custom_info)
+        if isinstance(general_custom_info, dict):
+            custom_info.update(general_custom_info)
+
+        # Ensure device_control_custom_info is the correct type
+        device_control_custom_info = (
+            device_control_custom_info_raw
+            if isinstance(device_control_custom_info_raw, dict)
+            else None
+        )
+
         return Metadata(
             asm_file_identifier=path.with_suffix(".json").name,
             data_system_instance_id=NOT_APPLICABLE,
             file_name=path.name,
             unc_path=file_path,
-            device_identifier=instrument_info_data["device_identifier"],
-            model_number=instrument_info_data["model_number"],
-            equipment_serial_number=instrument_info_data.get("equipment_serial_number"),
-            software_name=general_info_data["software_name"],
-            software_version=general_info_data["software_version"],
+            device_identifier=device_identifier,
+            model_number=model_number,
+            equipment_serial_number=equipment_serial_number,
+            software_name=software_name,
+            software_version=software_version,
+            metadata_custom_info=custom_info,
+            device_control_custom_info=device_control_custom_info,
         )
 
     @staticmethod
     def _get_general_info_data(
         general_info_df: pd.DataFrame | None,
-    ) -> dict[str, str | None]:
+    ) -> dict[str, str | None | dict[str, str]]:
         if general_info_df is None:
             return dict.fromkeys(GENERAL_INFO_KEYS, None)
 
@@ -106,15 +150,21 @@ def _get_general_info_data(
         software_name, software_version_txt = software_info.split(",", 1)
         match = re.search(r"\d+(?:\.\d+)*", software_version_txt)
         software_version = match.group() if match else None
+        unread_data = general_info_data.get_unread(skip={"Filter", "Position"})
+        # Filter out "nan" keys from custom info and ensure string values
+        custom_info = {
+            k: str(v) for k, v in unread_data.items() if k != "nan" and v is not None
+        }
         return {
             "software_name": software_name,
             "software_version": software_version,
+            "custom_info": custom_info,
         }
 
     @staticmethod
     def _get_instrument_data(
         instrument_info_df: pd.DataFrame | None,
-    ) -> dict[str, str]:
+    ) -> dict[str, str | None | dict[str, str] | dict[str, float | str | None]]:
         if instrument_info_df is None:
             return {
                 "device_identifier": NOT_APPLICABLE,
@@ -139,11 +189,45 @@ def _get_instrument_data(
             "model_number": ("Name", NOT_APPLICABLE),
             "equipment_serial_number": ("Serial number", None),
         }
-        return {
+        # Get the raw data first
+        raw_data = {
             key: value
             for key, (label, default) in lookups.items()
             if (value := instrument_info_data.get(str, label, default)) is not None
         }
+        device_control_custom_info = instrument_info_data.get_custom_keys(
+            {
+                "Optical response compensation",
+                "Plate adapter number",
+                "Plate adapter name",
+            }
+        )
+        unread_data = instrument_info_data.get_unread(
+            skip={
+                "Wavelength",
+                "Incubator",
+                "Date and time of definition",
+                "Top optics",
+                "Bottom optics",
+                "Dispenser 1",
+                "Dispenser 2",
+                "Module's name",
+                "Module's serial number",
+                "Instrument modules",
+                "Gas control",
+                "Bandwidth",
+                "Position",
+                "Filter",
+            }
+        )
+        custom_info = {
+            k: str(v) for k, v in unread_data.items() if k != "nan" and v is not None
+        }
+        return {
+            **raw_data,
+            "custom_info": custom_info,
+            "device_control_custom_info": device_control_custom_info,
+        }
 
 
 @dataclass
@@ -158,6 +242,7 @@ def create(
         well_plate_identifier: str | None,
         error_documents: list[ErrorDocument] | None = None,
         experimental_data_identifier: str | None = None,
+        device_control_custom_info: dict[str, Any] | None = None,
     ) -> Measurement:
         measurement_type_str = MEASUREMENT_TYPES[type_]
         error_docs = error_documents or []
@@ -184,6 +269,7 @@ def create(
             else None,
             experimental_data_identifier=experimental_data_identifier,
             error_document=error_docs if error_docs else None,
+            device_control_custom_info=device_control_custom_info,
         )
 
     @staticmethod
@@ -202,14 +288,19 @@ def create(
         sheet_df: pd.DataFrame,
         type_: MeasurementType,
         session_info_df: pd.DataFrame | None,
+        device_control_custom_info: dict[str, Any] | None = None,
     ) -> list[MeasurementGroup]:
         plates = ThermoSkanItMeasurementGroups.identify_data_and_sample_dfs(sheet_df)
 
         session_name = exec_time = None
+        custom_info = {}
         if session_info_df is not None:
             session_info_data = df_to_series_data(parse_header_row(session_info_df.T))
             session_name = session_info_data.get(str, "Session name")
             exec_time = session_info_data.get(str, "Execution time")
+            unread_data = session_info_data.get_unread(skip={"Executed with"})
+            # Filter out "nan" keys from custom info
+            custom_info = {k: v for k, v in unread_data.items() if k != "nan"}
 
         if not exec_time:
             exec_time = sheet_df.iloc[1].iloc[0]
@@ -273,6 +364,7 @@ def create(
                         experimental_data_identifier=session_name.replace(".skax", "")
                         if session_name
                         else None,
+                        device_control_custom_info=device_control_custom_info,
                     )
 
                     if well_key not in well_measurements:
@@ -288,6 +380,7 @@ def create(
                     measurements=measurements,
                     plate_well_count=plate_well_counts[plate_id],
                     measurement_time=exec_time,
+                    custom_info=custom_info,
                 )
             )
 
@@ -458,10 +551,15 @@ def create(sheet_data: dict[str, pd.DataFrame], file_path: str) -> Data:
             general_info_df=clean_data.get("General information"),
             file_path=file_path,
         )
+
+        # Extract device control custom info from metadata
+        device_control_custom_info = metadata.device_control_custom_info
+
         measurement_groups = ThermoSkanItMeasurementGroups.create(
             sheet_df=measurement_df,
             session_info_df=clean_data.get("Session information"),
             type_=_type,
+            device_control_custom_info=device_control_custom_info,
         )
 
         return Data(

src/allotropy/parsers/unchained_labs_lunatic_stunner/unchained_labs_lunatic_stunner_structure.py

@@ -463,7 +463,7 @@ def create_metadata(header: SeriesData, file_path: str) -> Metadata:
         unc_path=file_path,
         asm_file_identifier=asm_file_identifier.name,
         data_system_instance_id=NOT_APPLICABLE,
-        custom_info_doc=_filter_empty_string_values(
+        metadata_custom_info=_filter_empty_string_values(
             header.get_unread(
                 # Skip already mapped columns from well plate data (repeated in CSV no header cases)
                 skip={