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

 1from fhir.resources.practitioner import Practitioner
 2from omop_pydantic import Provider
 3
 4from chidian import DataMapping, Mapper
 5import chidian.partials as p
 6
 7def get_provider_name(name_list):
 8    if not name_list:
 9        return None
10    name = name_list[0]
11    given = " ".join(name.given) if name.given else ""
12    family = name.family or ""
13    return f"{given} {family}".strip()
14
15def get_npi(identifier_list):
16    if not identifier_list:
17        return None
18    for identifier in identifier_list:
19        if identifier.system == 'http://hl7.org/fhir/sid/us-npi':
20            return identifier.value
21    return None
22
23provider_mapper = Mapper(
24    lambda src: {
25        "provider_id": (p.get("id") | p.int())(src),
26        "provider_source_value": p.get("id")(src),
27        "provider_name": p.get("name", getter=get_provider_name)(src),
28        "npi": p.get("identifier", getter=get_npi)(src),
29        "gender_source_value": p.get("gender", getter=lambda x: x.upper() if x else None)(src),
30        "year_of_birth": p.get("birthDate", getter=lambda x: int(x.split('-')[0]) if x else None)(src),
31        "specialty_source_value": p.get("qualification[0].code.coding[0].code")(src),
32    }
33)
34
35to_omop_provider = DataMapping(
36    mapper=provider_mapper,
37    input_schema=Practitioner,
38    output_schema=Provider,
39)