refs/refs/fhir-x-omop/fhir_x_omop/to_omop/observation.py

 1from fhir.resources.observation import Observation as FHIRObservation
 2from omop_pydantic import Observation
 3
 4from chidian import DataMapping, Mapper
 5import chidian.partials as p
 6
 7def get_first_existing(src, paths):
 8    for path in paths:
 9        value = p.get(path)(src)
10        if value:
11            return value
12    return None
13
14observation_mapper = Mapper(
15    lambda src: {
16        "observation_id": (p.get("id") | p.int())(src),
17        "person_id": p.get("subject.reference", getter=lambda x: int(x.split('/')[1]) if x else None)(src),
18        "observation_concept_id": 0,  # Would need concept mapping in production
19        "observation_date": p.get("effectiveDateTime", getter=lambda x: x.split('T')[0] if x else None)(src),
20        "observation_datetime": p.get("effectiveDateTime")(src),
21        "observation_type_concept_id": p.case(p.get("code.coding[0].system")(src), {
22            "http://loinc.org": 32817,
23            "http://snomed.info/sct": 32818,
24        }, default=32819),
25        "value_as_number": (p.get("valueQuantity.value") | p.float())(src),
26        "value_as_string": p.get("valueString")(src),
27        "value_as_concept_id": (p.get("valueCodeableConcept.coding[0].code") | p.int())(src),
28        "qualifier_concept_id": 0,
29        "unit_concept_id": 0,
30        "provider_id": p.get("performer[0].reference", getter=lambda x: int(x.split('/')[1]) if x else None)(src),
31        "visit_occurrence_id": p.get("encounter.reference", getter=lambda x: int(x.split('/')[1]) if x else None)(src),
32        "visit_detail_id": p.get("encounter.reference", getter=lambda x: int(x.split('/')[1]) if x else None)(src),
33        "observation_source_value": p.get("code.coding[0].code")(src),
34        "observation_source_concept_id": 0,
35        "unit_source_value": p.get("valueQuantity.unit")(src),
36        "qualifier_source_value": p.get("code.coding[0].display")(src),
37        "value_source_value": (p.get(getter=lambda s: str(get_first_existing(s, ["valueCodeableConcept.coding[0].display", "valueString", "valueQuantity.value"]) or '')))(src),
38    }
39)
40
41to_omop_observation = DataMapping(
42    mapper=observation_mapper,
43    input_schema=FHIRObservation,
44    output_schema=Observation,
45)