Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add option to process tract without address in FHIR ETL #331

Merged
merged 1 commit into from
Dec 4, 2023
Merged

Add option to process tract without address in FHIR ETL #331

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 19 additions & 9 deletions lib/id3c/cli/command/etl/fhir.py
View file
Open in desktop
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
Loading