Diagnostic report resource added (#65)

Create DiagnosticReport & tests

fhirflat/__init__.py

@@ -16,9 +16,10 @@
     Procedure,
     ResearchSubject,
     Specimen,
+    DiagnosticReport,
 )
-from .ingest import convert_data_to_flat
+from .ingest import convert_data_to_flat, validate
 
 # Update this when bumping version in pyproject.toml!
 __version__ = "0.1.0"
-__all__ = ["convert_data_to_flat"]
+__all__ = ["convert_data_to_flat", "validate"]

fhirflat/resources/__init__.py

@@ -22,6 +22,7 @@
 from .procedure import Procedure
 from .researchsubject import ResearchSubject
 from .specimen import Specimen
+from .diagnosticreport import DiagnosticReport
 
 __all__ = [
     "Condition",
@@ -36,4 +37,5 @@
     "Procedure",
     "ResearchSubject",
     "Specimen",
+    "DiagnosticReport",
 ]

fhirflat/resources/diagnosticreport.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from typing import ClassVar, Union
+
+from fhir.resources import fhirtypes
+from fhir.resources.diagnosticreport import (
+    DiagnosticReport as _DiagnosticReport,
+)
+from fhir.resources.diagnosticreport import (
+    DiagnosticReportMedia,
+    DiagnosticReportSupportingInfo,
+)
+from pydantic.v1 import Field, validator
+
+from fhirflat.flat2fhir import expand_concepts
+
+from .base import FHIRFlatBase
+from .extension_types import timingPhaseType
+from .extensions import timingPhase
+
+
+class DiagnosticReport(_DiagnosticReport, FHIRFlatBase):
+    extension: list[Union[timingPhaseType, fhirtypes.ExtensionType]] = Field(
+        None,
+        alias="extension",
+        title="List of `Extension` items (represented as `dict` in JSON)",
+        description=(
+            """
+            Contains the Global.health 'timingPhase' extension,
+            and allows extensions from other implementations to be included.
+            """
+        ),
+        # if property is element of this resource.
+        element_property=True,
+        # this trys to match the type of the object to each of the union types
+        union_mode="smart",
+    )
+
+    # attributes to exclude from the flat representation
+    flat_exclusions: ClassVar[set[str]] = FHIRFlatBase.flat_exclusions | {
+        "identifier",
+    }
+
+    # required attributes that are not present in the FHIRflat representation
+    flat_defaults: ClassVar[list[str]] = [*FHIRFlatBase.flat_defaults, "status"]
+
+    backbone_elements: ClassVar[dict] = {
+        "supportingInfo": DiagnosticReportSupportingInfo,
+        "media": DiagnosticReportMedia,
+    }
+
+    @validator("extension")
+    def validate_extension_contents(cls, extensions):
+        tim_phase_count = sum(isinstance(item, timingPhase) for item in extensions)
+
+        if tim_phase_count > 1:
+            raise ValueError("timingPhase can only appear once.")
+
+        return extensions
+
+    @classmethod
+    def cleanup(cls, data: dict) -> dict:
+        """
+        Apply resource-specific changes to references and default values
+        """
+
+        for field in (
+            {
+                "basedOn",
+                "subject",
+                "performer",
+                "resultsInterpreter",
+                "specimen",
+                "result",
+                "study",
+                "composition",
+            }
+            | {x for x in data.keys() if x.endswith(".reference")}
+        ).intersection(data.keys()):
+            data[field] = {"reference": data[field]}
+
+        # add default status back in
+        data["status"] = "final"
+
+        data = expand_concepts(data, cls)
+
+        return data

tests/test_diagnosticreport_resource.py

@@ -0,0 +1,177 @@
+import pandas as pd
+from pandas.testing import assert_frame_equal
+import os
+from fhirflat.resources.diagnosticreport import DiagnosticReport
+import datetime
+
+
+DICT_INPUT = {
+    "resourceType": "DiagnosticReport",
+    "id": "f001",
+    "identifier": [
+        {
+            "use": "official",
+            "system": "http://www.bmc.nl/zorgportal/identifiers/reports",
+            "value": "nr1239044",
+        }
+    ],
+    "extension": [
+        {
+            "url": "timingPhase",
+            "valueCodeableConcept": {
+                "coding": [{"system": "timing.com", "code": "1234"}]
+            },
+        },
+    ],
+    "basedOn": [{"reference": "ServiceRequest/req"}],
+    "status": "final",
+    "category": [
+        {
+            "coding": [
+                {
+                    "system": "http://snomed.info/sct",
+                    "code": "252275004",
+                    "display": "Haematology test",
+                },
+                {
+                    "system": "http://terminology.hl7.org/CodeSystem/v2-0074",
+                    "code": "HM",
+                },
+            ]
+        }
+    ],
+    "code": {
+        "coding": [
+            {
+                "system": "http://loinc.org",
+                "code": "58410-2",
+                "display": "Complete blood count (hemogram) panel - Blood by Automated count",  # noqa: E501
+            }
+        ]
+    },
+    "subject": {"reference": "Patient/f001", "display": "P. van den Heuvel"},
+    "issued": "2013-05-15T19:32:52+01:00",
+    "performer": [
+        {
+            "reference": "Organization/f001",
+            "display": "Burgers University Medical Centre",
+        }
+    ],
+    "result": [
+        {"reference": "Observation/f001"},
+        {"reference": "Observation/f002"},
+        {"reference": "Observation/f003"},
+        {"reference": "Observation/f004"},
+        {"reference": "Observation/f005"},
+    ],
+    "conclusion": "Core lab",
+}
+
+FLAT = {
+    "resourceType": "DiagnosticReport",
+    "id": "f001",
+    "basedOn": "ServiceRequest/req",
+    "category.code": [
+        "http://snomed.info/sct|252275004",
+        "http://terminology.hl7.org/CodeSystem/v2-0074|HM",
+    ],
+    "category.text": ["Haematology test", None],
+    "issued": datetime.datetime(
+        2013,
+        5,
+        15,
+        19,
+        32,
+        52,
+        tzinfo=datetime.timezone(datetime.timedelta(minutes=60)),
+    ),
+    "performer": "Organization/f001",
+    "result_dense": [
+        {"reference": "Observation/f001"},
+        {"reference": "Observation/f002"},
+        {"reference": "Observation/f003"},
+        {"reference": "Observation/f004"},
+        {"reference": "Observation/f005"},
+    ],
+    "conclusion": "Core lab",
+    "code.code": ["http://loinc.org|58410-2"],
+    "code.text": ["Complete blood count (hemogram) panel - Blood by Automated count"],
+    "subject": "Patient/f001",
+    "extension.timingPhase.code": "timing.com|1234",
+    "extension.timingPhase.text": None,
+}
+
+DICT_OUT = {
+    "resourceType": "DiagnosticReport",
+    "id": "f001",
+    "extension": [
+        {
+            "url": "timingPhase",
+            "valueCodeableConcept": {
+                "coding": [{"system": "timing.com", "code": "1234"}]
+            },
+        },
+    ],
+    "basedOn": [{"reference": "ServiceRequest/req"}],
+    "status": "final",
+    "category": [
+        {
+            "coding": [
+                {
+                    "system": "http://snomed.info/sct",
+                    "code": "252275004",
+                    "display": "Haematology test",
+                },
+                {
+                    "system": "http://terminology.hl7.org/CodeSystem/v2-0074",
+                    "code": "HM",
+                },
+            ]
+        }
+    ],
+    "code": {
+        "coding": [
+            {
+                "system": "http://loinc.org",
+                "code": "58410-2",
+                "display": "Complete blood count (hemogram) panel - Blood by Automated count",  # noqa: E501
+            }
+        ]
+    },
+    "subject": {"reference": "Patient/f001"},
+    "issued": "2013-05-15T19:32:52+01:00",
+    "performer": [
+        {
+            "reference": "Organization/f001",
+        }
+    ],
+    "result": [
+        {"reference": "Observation/f001"},
+        {"reference": "Observation/f002"},
+        {"reference": "Observation/f003"},
+        {"reference": "Observation/f004"},
+        {"reference": "Observation/f005"},
+    ],
+    "conclusion": "Core lab",
+}
+
+
+def test_observation_to_flat():
+    report = DiagnosticReport(**DICT_INPUT)
+
+    report.to_flat("test_diagnosticreport.parquet")
+
+    report_flat = pd.read_parquet("test_diagnosticreport.parquet")
+    expected = pd.DataFrame([FLAT], index=[0])
+    # v, e = DiagnosticReport.validate_fhirflat(expected)
+    assert_frame_equal(report_flat, expected, check_dtype=False)
+
+    os.remove("test_diagnosticreport.parquet")
+
+
+def test_observation_from_flat():
+    report = DiagnosticReport(**DICT_OUT)
+
+    flat_report = DiagnosticReport.from_flat("tests/data/diagnosticreport_flat.parquet")
+
+    assert report == flat_report