refs/refs/fhir-x-omop/fhir_x_omop/to_fhir/condition.py

lines 1–72 73 lines · py
 1from fhir.resources.condition import Condition
 2from omop_pydantic import ConditionOccurrence
 3
 4from chidian import DataMapping, Mapper
 5import chidian.partials as p
 6
 7condition_mapper = Mapper(
 8    lambda src: {
 9        "resourceType": "Condition",
10        "id": (p.get("condition_occurrence_id") | p.str())(src),
11        "identifier": [{
12            "system": "http://omop.org/condition",
13            "value": (p.get("condition_occurrence_id") | p.str())(src),
14        }],
15        "clinicalStatus": {
16            "coding": [{
17                "system": "http://terminology.hl7.org/CodeSystem/condition-clinical",
18                "code": p.case(p.get("condition_status_concept_id")(src), {
19                    32893: "active",
20                    32897: "resolved",
21                    32896: "inactive",
22                }, default="active")
23            }]
24        },
25        "verificationStatus": {
26            "coding": [{
27                "system": "http://terminology.hl7.org/CodeSystem/condition-ver-status",
28                "code": "confirmed"
29            }]
30        },
31        "category": [{
32            "coding": [{
33                "system": "http://terminology.hl7.org/CodeSystem/condition-category",
34                "code": p.case(p.get("condition_type_concept_id")(src), {
35                    32817: "encounter-diagnosis",
36                    32818: "problem-list-item",
37                    32819: "health-concern",
38                }, default="encounter-diagnosis")
39            }]
40        }],
41        "code": {
42            "coding": [{
43                "system": p.case(p.get("condition_type_concept_id")(src), {
44                    32817: "http://hl7.org/fhir/sid/icd-10-cm",
45                    32818: "http://snomed.info/sct",
46                }, default="http://hl7.org/fhir/sid/icd-10-cm"),
47                "code": p.get("condition_source_value")(src),
48                "display": p.get("condition_source_value")(src)
49            }]
50        },
51        "subject": {
52            "reference": (p.get("person_id") | p.format("Patient/{}"))(src)
53        },
54        "encounter": {
55            "reference": (p.get("visit_occurrence_id") | p.format("Encounter/{}"))(src)
56        },
57        "onsetDateTime": (p.get("condition_start_datetime") | p.str())(src),
58        "abatementDateTime": (p.get("condition_end_datetime") | p.str())(src),
59        "recordedDate": (p.get("condition_start_datetime") | p.str())(src),
60        "recorder": {
61            "reference": (p.get("provider_id") | p.format("Practitioner/{}"))(src)
62        },
63        "note": [{
64            "text": p.get("stop_reason")(src)
65        }]
66    }
67)
68
69to_fhir_condition = DataMapping(
70    mapper=condition_mapper,
71    input_schema=ConditionOccurrence,
72    output_schema=Condition,
73)