Skip to main content

functions

Pure helpers for the patient_level_eligibility step.

Patient-identity construction from the EHR cache (falling back to the evaluation's SubjectIdentity), used by patient_level_eligibility_task, kept apart from the task orchestration. No I/O.

Module​

Functions​

build_ehr_lookup​

def build_ehr_lookup(    ehr_patient_data: CacheAccessor | None,) ‑> dict[str, dict[str, typing.Any]]:

Map bitfount_patient_id → EHR identity fields from the EHR cache.

Arguments

  • ehr_patient_data: Optional EHR cache accessor.

Returns A lookup keyed by Bitfount patient ID (empty if no EHR data), each value a {ehr_patient_id, name, mrns, processed_at, error} dict.

display_name_from_scan_name​

def display_name_from_scan_name(scan_name: str) ‑> str:

Format a scan-sourced patient name for display.

Imaging sources record the name in whatever form their vendor uses — DICOM stores Family^Given^Middle^Prefix^Suffix, commonly padded to SMITH^JANE^^^. That form reaches the patient-data API verbatim unless it is formatted here, and SMITH^JANE^^^ is not a name anyone recognises.

Reduces the name to "Given Family". Returns the original string unchanged when it cannot be parsed, so an unusual name is still shown rather than replaced with something emptier.

Arguments

  • scan_name: The name as recorded on the scan.

Returns The display form of the name.

ehr_retrieved_at_for​

def ehr_retrieved_at_for(    bitfount_patient_id: str, ehr_lookup: dict[str, dict[str, Any]],) ‑> datetime.datetime | None:

The value for one patient's patient_level_eligibility.ehr_retrieved_at.

Sourced from the served EHR row's own processed_at (via ehr_lookup, from build_ehr_lookup) rather than a fresh timestamp: the point is when the patient's EHR data was retrieved, which — under best-effort serving — may be long before this run, not this run's own time.

A row whose error is set records a lookup that retrieved nothing — its processed_at is the run that wrote the row, not a moment at which any EHR data was retrieved. There is no retrieval time for such a patient, so this reports None: cache/types/patient_level_eligibility/v2/schema.py already declares NULL to mean "no EHR data at all", and reporting the write time instead would read fresher than the truth for the one population the field exists to flag. A row whose every countable fetch group failed carries error too, and reports None on the same grounds — it holds demographics but no clinical data for a verdict to rest on.

The lookup value comes off a CacheAccessor DataFrame, so it arrives as either the stored ISO-8601 string or an already-parsed timestamp; both are accepted. An unparseable value is dropped rather than stored, because the column is typed and NULL already carries the right meaning for it.

Arguments

  • bitfount_patient_id: The patient's Bitfount ID.
  • ehr_lookup: EHR identity lookup (from build_ehr_lookup).

Returns The stored retrieval time, or None — no EHR accessor, no row for this patient, a row recording a failed lookup, a row with no recorded processed_at, or one whose recorded value cannot be read as a timestamp. Every one of those is "not recorded".

identity_for​

def identity_for(    identity: SubjectIdentity,    bitfount_patient_id: str,    ehr_lookup: dict[str, dict[str, Any]],    now: datetime,    task_hash: str,    run_id: str | None,) ‑> PatientRecord:

Build a patient identity record from EHR data or the evaluation's identity.

The display name is resolved EHR name → the evaluation's identity.name → the bare bitfount_patient_id. That last resort is an undesirable state (a named-less patient) and emits a warning.

The EHR name is already split into parts by build_ehr_lookup; the evaluation's name comes off the scan in its vendor's own form, so it goes through display_name_from_scan_name first.

Arguments

  • identity: The subject identity carried on the evaluation (name/MRN fallback).
  • bitfount_patient_id: The patient's ID, which the caller has already established is present. Taken as its own argument rather than read off identity, whose own field is optional: a PatientRecord keys on this, so there is no record to build without it, and passing it in puts that requirement in the signature instead of leaving this function to re-check what its caller has already checked.
  • ehr_lookup: EHR identity lookup (from build_ehr_lookup).
  • now: The write timestamp.
  • task_hash: Provenance task hash.
  • run_id: Provenance run ID.

Returns A PatientRecord for the patient.