feat: Qiacuity dPCR - Added unread data (#1103)

Author: felipenarv

src/allotropy/allotrope/schema_mappers/adm/pcr/BENCHLING/_2023/_09/dpcr.py

@@ -90,6 +90,7 @@ class Measurement:
     errors: list[Error] | None = None
 
     # custom
+    sample_custom_info: dict[str, Any] | None = None
     custom_info: dict[str, Any] | None = None
 
 
@@ -182,6 +183,15 @@ def _get_technique_document(
     def _get_measurement_document(
         self, measurement: Measurement, metadata: Metadata
     ) -> MeasurementDocumentItem:
+        sample_document = SampleDocument(
+            sample_identifier=measurement.sample_identifier,
+            well_location_identifier=measurement.location_identifier,
+            well_plate_identifier=measurement.plate_identifier,
+            sample_role_type=measurement.sample_role_type,
+        )
+        sample_document = add_custom_information_document(
+            sample_document, measurement.sample_custom_info
+        )
         measurement_doc = MeasurementDocumentItem(
             measurement_identifier=measurement.identifier,
             measurement_time=self.get_date_time(measurement.measurement_time),
@@ -198,12 +208,7 @@ def _get_measurement_document(
             error_aggregate_document=self._get_error_aggregate_document(
                 measurement.errors
             ),
-            sample_document=SampleDocument(
-                sample_identifier=measurement.sample_identifier,
-                well_location_identifier=measurement.location_identifier,
-                well_plate_identifier=measurement.plate_identifier,
-                sample_role_type=measurement.sample_role_type,
-            ),
+            sample_document=sample_document,
             device_control_aggregate_document=DeviceControlAggregateDocument(
                 device_control_document=[
                     DeviceControlDocumentItem(
@@ -244,6 +249,7 @@ def _get_measurement_document(
                 ]
             ),
         )
+
         return add_custom_information_document(measurement_doc, measurement.custom_info)
 
     def _get_calculated_data_aggregate_document(

src/allotropy/parsers/qiacuity_dpcr/constants.py

@@ -1,3 +1,30 @@
+CALCULATED_DATA_CONFIGS: list[dict[str, str]] = [
+    {
+        "name": "CI (95%)",
+        "keys": "CI (95%)",
+        "unit": "%",
+        "feature": "Positive Partition Count",
+    },
+    {
+        "name": "SD",
+        "keys": "SD",
+        "unit": "(unitless)",
+        "feature": "Number Concentration",
+    },
+    {
+        "name": "CV%",
+        "keys": "CV%",
+        "unit": "%",
+        "feature": "Number Concentration (#/μL)",
+    },
+    {
+        "name": "Mean Concentration",
+        "keys": "Mean conc. [copies/μL]",
+        "unit": "#/μL",
+        "feature": "Mean Concentration (#/μL)",
+    },
+]
+
 BRAND_NAME = "Qiacuity Digital PCR System"
 PRODUCT_MANUFACTURER = "Qiagen"
 SOFTWARE_NAME = "Qiacuity Software Suite"

src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_calculated_data.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from allotropy.allotrope.schema_mappers.adm.pcr.BENCHLING._2023._09.dpcr import (
+    CalculatedDataItem,
+    DataSource,
+)
+from allotropy.parsers.qiacuity_dpcr.constants import CALCULATED_DATA_CONFIGS
+from allotropy.parsers.utils.calculated_data_documents.definition import (
+    CalculatedDocument as UtilsCalculatedDocument,
+    DataSource as UtilsDataSource,
+    Referenceable as UtilsReferenceable,
+)
+from allotropy.parsers.utils.pandas import SeriesData
+from allotropy.parsers.utils.uuids import random_uuid_str
+
+
+def _iter_row_calculated_docs(row: SeriesData) -> Iterable[UtilsCalculatedDocument]:
+    measurement_identifier = row.get(str, "_measurement_identifier")
+    if not measurement_identifier:
+        return []
+
+    docs: list[UtilsCalculatedDocument] = []
+
+    measurement_ref = UtilsReferenceable(uuid=measurement_identifier)
+
+    for conf in CALCULATED_DATA_CONFIGS:
+        value = row.get(float, conf["keys"])
+        if value is None:
+            continue
+        docs.append(
+            UtilsCalculatedDocument(
+                uuid=random_uuid_str(),
+                name=conf["name"],
+                value=float(value),
+                unit=conf["unit"],
+                data_sources=[
+                    UtilsDataSource(
+                        feature=conf.get("feature", conf["name"]),
+                        reference=measurement_ref,
+                    )
+                ],
+            )
+        )
+
+    return docs
+
+
+def _docs_to_benchling_items(
+    docs: Iterable[UtilsCalculatedDocument],
+) -> list[CalculatedDataItem]:
+    items: list[CalculatedDataItem] = []
+    for doc in docs:
+        for flat_doc in doc.iter_struct():
+            items.append(
+                CalculatedDataItem(
+                    identifier=flat_doc.uuid,
+                    name=flat_doc.name,
+                    value=flat_doc.value,
+                    unit=flat_doc.unit or "",
+                    data_sources=[
+                        DataSource(
+                            identifier=ds.reference.uuid,
+                            feature=ds.feature,
+                        )
+                        for ds in flat_doc.data_sources
+                    ],
+                )
+            )
+    return items
+
+
+def create_calculated_data(rows: list[SeriesData]) -> list[CalculatedDataItem]:
+    docs: list[UtilsCalculatedDocument] = []
+    for row in rows:
+        docs.extend(list(_iter_row_calculated_docs(row)))
+    return _docs_to_benchling_items(docs)

src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_parser.py

@@ -7,13 +7,17 @@
     MeasurementGroup,
 )
 from allotropy.named_file_contents import NamedFileContents
+from allotropy.parsers.qiacuity_dpcr.qiacuity_dpcr_calculated_data import (
+    create_calculated_data as create_qiacuity_calculated_data,
+)
 from allotropy.parsers.qiacuity_dpcr.qiacuity_dpcr_reader import QiacuitydPCRReader
 from allotropy.parsers.qiacuity_dpcr.qiacuity_dpcr_structure import (
     create_measurements,
     create_metadata,
 )
 from allotropy.parsers.release_state import ReleaseState
-from allotropy.parsers.utils.pandas import map_rows
+from allotropy.parsers.utils.pandas import SeriesData
+from allotropy.parsers.utils.uuids import random_uuid_str
 from allotropy.parsers.vendor_parser import VendorParser
 
 
@@ -25,14 +29,25 @@ class QiacuitydPCRParser(VendorParser[Data, Model]):
 
     def create_data(self, named_file_contents: NamedFileContents) -> Data:
         reader = QiacuitydPCRReader(named_file_contents)
+        # Assign stable measurement identifiers per row for data source linkage
+        reader.well_data["_measurement_identifier"] = [
+            random_uuid_str() for _ in range(len(reader.well_data))
+        ]
+        # Build SeriesData list once and reuse
+        series_rows: list[SeriesData] = [
+            SeriesData(row) for _, row in reader.well_data.iterrows()
+        ]
+        calculated_data = create_qiacuity_calculated_data(series_rows)
+        measurement_groups = [
+            MeasurementGroup(
+                measurements=[create_measurements(row) for row in series_rows],
+                # TODO: Hardcoded plate well count to 0 since it's a required field
+                #  ASM will be modified to optional in future version
+                plate_well_count=0,
+            )
+        ]
         return Data(
             create_metadata(named_file_contents.original_file_path),
-            measurement_groups=[
-                MeasurementGroup(
-                    measurements=map_rows(reader.well_data, create_measurements),
-                    # TODO: Hardcoded plate well count to 0 since it's a required field
-                    #  ASM will be modified to optional in future version
-                    plate_well_count=0,
-                )
-            ],
+            measurement_groups=measurement_groups,
+            calculated_data=calculated_data,
         )

src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_structure.py

@@ -29,8 +29,15 @@ def create_measurements(data: SeriesData) -> Measurement:
             "sample type", sample_role_type, SAMPLE_ROLE_TYPE_MAPPING
         )
 
+    identifier = data.get(str, "_measurement_identifier") or random_uuid_str()
+
+    sample_custom_info = data.get_custom_keys({"IC", "Control type"})
+    for key in sample_custom_info:
+        if sample_custom_info[key] in ("", "-", "-", "--"):
+            sample_custom_info[key] = None
+
     return Measurement(
-        identifier=random_uuid_str(),
+        identifier=identifier,
         measurement_time=DEFAULT_EPOCH_TIMESTAMP,
         sample_identifier=data[str, "Sample/NTC/Control"],
         sample_role_type=sample_role_type,
@@ -42,6 +49,8 @@ def create_measurements(data: SeriesData) -> Measurement:
         positive_partition_count=data[int, "Partitions (positive)"],
         negative_partition_count=data.get(int, "Partitions (negative)"),
         fluorescence_intensity_threshold_setting=data.get(float, "Threshold"),
+        sample_custom_info=sample_custom_info,
+        custom_info=data.get_unread(),
     )