Skip to content

Commit

Permalink
Add option to process tract without address in FHIR ETL
Browse files Browse the repository at this point in the history
  • Loading branch information
sallybg committed Dec 4, 2023
1 parent 5dc3023 commit 968ee3d
Showing 1 changed file with 19 additions and 9 deletions.
28 changes: 19 additions & 9 deletions lib/id3c/cli/command/etl/fhir.py
Original file line number Diff line number Diff line change
Expand Up @@ -798,15 +798,17 @@ def process_locations(db: DatabaseSession, encounter_id: int, encounter: Encount
def process_location(db: DatabaseSession, encounter_id: int, location: Location):
"""
Process a FHIR *location* and attach it to an *encounter_id*.
First looks for a parent location's tract hierarchy;
if no parent location is available, assumes that the location entry itself
is a tract and looks for a tract hierarchy.
"""
# XXX FIXME: This function assumes we always have an address, and never
# just a tract. It also assumes the scales tract and address and their
# XXX FIXME: This function assumes the scales tract and address and their
# relationship in the hierarchy instead of being agnostic to the scale.
# -trs, 19 Dec 2019

def get_tract_hierarchy(location: Location) -> Optional[str]:
def get_tract(location: Location) -> Optional[Any]:
"""
Given a *location*, returns its tract hierarchy if it exists, else None.
Given a *location*, returns its tract location entry if it exists, else None.
"""
scale = 'tract'
tract_identifier = identifier(location, f"{INTERNAL_SYSTEM}/location/{scale}")
Expand All @@ -817,7 +819,7 @@ def get_tract_hierarchy(location: Location) -> Optional[str]:
tract = find_location(db, scale, tract_identifier)
assert tract, f"Tract «{tract_identifier}» is unknown"

return tract.hierarchy
return tract

def process_address(location: Location, tract_hierarchy: str) -> Optional[Any]:
"""
Expand All @@ -844,12 +846,19 @@ def process_address(location: Location, tract_hierarchy: str) -> Optional[Any]:

try:
parent_location = location.partOf.resolved(Location)
tract_hierarchy = get_tract_hierarchy(parent_location) # TODO do we only want parent hierarchy?
tract_hierarchy = get_tract(parent_location).hierarchy # TODO do we only want parent hierarchy?

except AttributeError:
LOG.info(f"No parent location found for Location with code «{code}», " + \
f"relation «{relation}».")
tract_hierarchy = None
f"relation «{relation}». " + \
"Assuming location is a tract.")
try: # handle encounter locations that are only tracts
tract = get_tract(location)
tract_hierarchy = tract.hierarchy
except (AttributeError, AssertionError):
LOG.info(f"No tract found for Location with code «{code}», " + \
f"relation «{relation}».")
tract_hierarchy = None

address = process_address(location, tract_hierarchy)

Expand All @@ -860,7 +869,8 @@ def process_address(location: Location, tract_hierarchy: str) -> Optional[Any]:
upsert_encounter_location(db,
encounter_id = encounter_id,
relation = relation,
location_id = address.id)
location_id = address.id if address else tract.id)
# if there is an address, link encounter location to address; if only tract, link encounter location to tract


def location_code(location: Location) -> str:
Expand Down

0 comments on commit 968ee3d

Please sign in to comment.