Used calc docs library for implementation

Author: felipenarv

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

@@ -183,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),
@@ -199,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(
@@ -246,9 +250,6 @@ def _get_measurement_document(
             ),
         )
 
-        measurement_doc.sample_document = add_custom_information_document(
-            measurement_doc.sample_document, measurement.sample_custom_info
-        )
         return add_custom_information_document(measurement_doc, measurement.custom_info)
 
     def _get_calculated_data_aggregate_document(

src/allotropy/parsers/qiacuity_dpcr/constants.py

@@ -1,35 +1,30 @@
-from dataclasses import dataclass
-
-
-@dataclass(frozen=True)
-class CalculatedDataConfig:
-    name: str
-    key: str
-    unit: str
-    feature: str | None = None
-
-
-CONFIGS: list[CalculatedDataConfig] = [
-    CalculatedDataConfig(
-        name="CI (95%)", key="CI (95%)", unit="%", feature="Positive Partition Count"
-    ),
-    CalculatedDataConfig(
-        name="SD", key="SD", unit="(unitless)", feature="Number Concentration (#/μL)"
-    ),
-    CalculatedDataConfig(
-        name="CV%", key="CV%", unit="%", feature="Number Concentration (#/μL)"
-    ),
-    CalculatedDataConfig(
-        name="Mean Concentration",
-        key="Mean conc. [copies/μL]",
-        unit="#/μL",
-        feature="Mean Concentration (#/μL)",
-    ),
+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)",
+    },
 ]
 
-CONFIGS_KEYS: set[str] = {config.key for config in CONFIGS}
-
-
 BRAND_NAME = "Qiacuity Digital PCR System"
 PRODUCT_MANUFACTURER = "Qiagen"
 SOFTWARE_NAME = "Qiacuity Software Suite"

src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_calculated_data.py

@@ -6,42 +6,73 @@
     CalculatedDataItem,
     DataSource,
 )
-from allotropy.parsers.qiacuity_dpcr.constants import CONFIGS
+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_items(row: SeriesData) -> Iterable[CalculatedDataItem]:
+def _iter_row_calculated_docs(row: SeriesData) -> Iterable[UtilsCalculatedDocument]:
     measurement_identifier = row.get(str, "_measurement_identifier")
     if not measurement_identifier:
         return []
 
-    items: list[CalculatedDataItem] = []
+    docs: list[UtilsCalculatedDocument] = []
 
-    for conf in CONFIGS:
-        value = row.get(float, conf.key)
+    measurement_ref = UtilsReferenceable(uuid=measurement_identifier)
+
+    for conf in CALCULATED_DATA_CONFIGS:
+        value = row.get(float, conf["keys"])
         if value is None:
             continue
-        items.append(
-            CalculatedDataItem(
-                identifier=random_uuid_str(),
-                name=conf.name,
+        docs.append(
+            UtilsCalculatedDocument(
+                uuid=random_uuid_str(),
+                name=conf["name"],
                 value=float(value),
-                unit=conf.unit,
+                unit=conf["unit"],
                 data_sources=[
-                    DataSource(
-                        identifier=measurement_identifier,
-                        feature=conf.feature or conf.name,
+                    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]:
-    calculated: list[CalculatedDataItem] = []
+    docs: list[UtilsCalculatedDocument] = []
     for row in rows:
-        calculated.extend(list(_iter_row_calculated_items(row)))
-    return calculated
+        docs.extend(list(_iter_row_calculated_docs(row)))
+    return _docs_to_benchling_items(docs)