Cavell SDK — Lab Results Ingestion Demo¶

Ingest structured lab results into FHIR — deterministically. Unlike the note extraction demos, no LLM is involved anywhere: the CSV already carries every field an Observation needs (value, unit, reference range, collection time, references), so the Prism API builds the resources by plain code. That means zero token spend, instant responses, and byte-for-byte reproducible output.

Three properties this demo shows off:

1. Fail-closed references — every patient_id, encounter_id and practitioner_id must already exist in your FHIR server. Rows referencing anything unknown are rejected and reported, never created for.
2. Skip bad rows, report them — a malformed datetime or blank value rejects that row with a reason; the rest of the feed proceeds. The final report merges client-side and API-side rejections with input row numbers.
3. Idempotency — each row's lab_result_id becomes the Observation's urn:cavell:lab-result identifier and the bundle uses conditional creates on it, so re-running the whole feed creates nothing twice.

Prerequisites:

• Docker running (docker compose up -d)
• A Prism API URL and an LLM Gateway key (only used to authenticate — this endpoint spends no tokens)
• The hospitalization demo has been run first (hospitalization_extraction_demo.ipynb): this dataset layers labs onto its patients, practitioners and admissions, and a guard cell below checks they are in place.

All data in this demo is fully synthetic. Names, identifiers and results were generated for demonstration purposes and do not describe real patients.

Dataset: structured labs for the hospitalization cohort¶

lab_results.csv holds 251 rows — the structured twin of the prose "LABORATORY RESULTS" notes in hospitalizations.csv, same patients (MRN-20001…MRN-20017), same admissions (V-001…V-018, plus the backdated V-902/V-903), same practitioners, and values consistent with what those notes state (the V-009 STEMI troponins really do run 480 → 5,200 ng/L, timestamped hours apart — which is why collected_datetime keeps its time and timezone).

Beyond the in-stay draws it adds:

• 29 pre-admission / post-discharge rows with a blank encounter_id — GP bloods foreshadowing each admission (the creatinine "baseline 40" the V-003 admission note refers to, the iron-deficiency picture that triggered V-006's cancer workup, HbA1c drifting up before the V-013 diabetic foot) and a post-discharge CEA after V-006's hemicolectomy.
• 7 rows without a LOINC code (dipstick, cultures, FIT…) — these get a text-only code instead of a LOINC coding.
• 5 comparator values (<5, <14, >=90) — kept as FHIR comparator.
• Qualitative results ("No growth after 5 days", "Escherichia coli >10^5 CFU/mL") — kept verbatim as valueString.

The last 7 rows are deliberately broken, to exercise the rejection report:

Rows	Problem	Rejected by
LAB-0245, LAB-0246	unknown patient MRN-99999	client (reference)
LAB-0247, LAB-0248	unknown encounter V-999	client (reference)
LAB-0249	unknown practitioner DOC-999	client (reference)
LAB-0250	empty value	client (validation)
LAB-0251	datetime without timezone offset	API (server)

1. Install¶

In [ ]:

%pip install cavell-prism-client

%pip install cavell-prism-client

In [ ]:

import logging

logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s")
logging.getLogger("cavell_client").setLevel(logging.INFO)

import logging logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s") logging.getLogger("cavell_client").setLevel(logging.INFO)

2. Check FHIR server¶

In [ ]:

import httpx

FHIR_BASE_URL = "http://localhost:8090"

resp = httpx.get(f"{FHIR_BASE_URL}/fhir/metadata", timeout=5)
assert resp.status_code == 200, "FHIR server not reachable — run: docker compose up -d"
print(f"FHIR server OK at {FHIR_BASE_URL}")

import httpx FHIR_BASE_URL = "http://localhost:8090" resp = httpx.get(f"{FHIR_BASE_URL}/fhir/metadata", timeout=5) assert resp.status_code == 200, "FHIR server not reachable — run: docker compose up -d" print(f"FHIR server OK at {FHIR_BASE_URL}")

3. Check prerequisites¶

Labs attach to existing patients, practitioners and admissions — this pipeline never creates them. Everything it references was put there by the hospitalization demo, so verify its state is actually in this FHIR server before spending any effort.

In [ ]:

def _exists(resource_type: str, system: str, value: str) -> bool:
    r = httpx.get(
        f"{FHIR_BASE_URL}/fhir/{resource_type}",
        params={"identifier": f"{system}|{value}"},
        timeout=10,
    )
    return r.status_code == 200 and bool(r.json().get("entry"))


MISSING = [
    f"{rt} {value}"
    for rt, system, value in [
        ("Patient", "urn:cavell:patient", "MRN-20001"),
        ("Practitioner", "urn:cavell:practitioner", "DOC-201"),
        ("Encounter", "urn:cavell:encounter", "V-001"),
    ]
    if not _exists(rt, system, value)
]
assert not MISSING, (
    f"Missing from FHIR: {', '.join(MISSING)}. Run "
    "hospitalization_extraction_demo.ipynb first — this demo layers structured "
    "labs onto the patients, practitioners and admissions it creates."
)
print(
    "Hospitalization demo state found — patients, practitioners, admissions in place."
)

def _exists(resource_type: str, system: str, value: str) -> bool: r = httpx.get( f"{FHIR_BASE_URL}/fhir/{resource_type}", params={"identifier": f"{system}|{value}"}, timeout=10, ) return r.status_code == 200 and bool(r.json().get("entry")) MISSING = [ f"{rt} {value}" for rt, system, value in [ ("Patient", "urn:cavell:patient", "MRN-20001"), ("Practitioner", "urn:cavell:practitioner", "DOC-201"), ("Encounter", "urn:cavell:encounter", "V-001"), ] if not _exists(rt, system, value) ] assert not MISSING, ( f"Missing from FHIR: {', '.join(MISSING)}. Run " "hospitalization_extraction_demo.ipynb first — this demo layers structured " "labs onto the patients, practitioners and admissions it creates." ) print( "Hospitalization demo state found — patients, practitioners, admissions in place." )

4. Load the CSV¶

In [ ]:

import csv

CSV_PATH = "lab_results.csv"  # structured labs for the hospitalization cohort

with open(CSV_PATH, newline="", encoding="utf-8-sig") as f:
    rows = list(csv.DictReader(f))

print(f"{len(rows)} rows")
print(f"Columns: {list(rows[0].keys())}")
rows[0]

import csv CSV_PATH = "lab_results.csv" # structured labs for the hospitalization cohort with open(CSV_PATH, newline="", encoding="utf-8-sig") as f: rows = list(csv.DictReader(f)) print(f"{len(rows)} rows") print(f"Columns: {list(rows[0].keys())}") rows[0]

5. Configure column mapping¶

Only five columns are required. Everything else is optional — map a field to None if your export does not have that column.

In [ ]:

# Required
COL_LAB_RESULT_ID = "lab_result_id"  # your LIS accession id — the idempotency key
COL_PATIENT_ID = "patient_id"  # MRN (urn:cavell:patient), resolved against FHIR
COL_TEST_NAME = "test_name"
COL_VALUE = "value"  # numeric (comparators ok) -> valueQuantity; else valueString
COL_COLLECTED = "collected_datetime"  # date, or datetime WITH timezone offset

# Optional — set to None if your CSV doesn't have these columns
COL_LOINC = "loinc_code"  # present -> LOINC coding; absent -> text-only code
COL_UNIT = "unit"
COL_REF_LOW = "reference_low"
COL_REF_HIGH = "reference_high"
COL_ENCOUNTER_ID = "encounter_id"  # admission the draw belongs to; blank = none
COL_PRACTITIONER_ID = "practitioner_id"  # performing/ordering practitioner

# Required COL_LAB_RESULT_ID = "lab_result_id" # your LIS accession id — the idempotency key COL_PATIENT_ID = "patient_id" # MRN (urn:cavell:patient), resolved against FHIR COL_TEST_NAME = "test_name" COL_VALUE = "value" # numeric (comparators ok) -> valueQuantity; else valueString COL_COLLECTED = "collected_datetime" # date, or datetime WITH timezone offset # Optional — set to None if your CSV doesn't have these columns COL_LOINC = "loinc_code" # present -> LOINC coding; absent -> text-only code COL_UNIT = "unit" COL_REF_LOW = "reference_low" COL_REF_HIGH = "reference_high" COL_ENCOUNTER_ID = "encounter_id" # admission the draw belongs to; blank = none COL_PRACTITIONER_ID = "practitioner_id" # performing/ordering practitioner

6. Build LabResult objects¶

In [ ]:

from cavell_client import LabResult

results = LabResult.from_rows(
    rows,
    columns={
        "lab_result_id": COL_LAB_RESULT_ID,
        "patient_identifier": COL_PATIENT_ID,
        "test_name": COL_TEST_NAME,
        "value": COL_VALUE,
        "collected_datetime": COL_COLLECTED,
        "loinc_code": COL_LOINC,
        "unit": COL_UNIT,
        "reference_low": COL_REF_LOW,
        "reference_high": COL_REF_HIGH,
        "encounter_id": COL_ENCOUNTER_ID,
        "practitioner_id": COL_PRACTITIONER_ID,
    },
)
print(f"{len(results)} lab results")

from cavell_client import LabResult results = LabResult.from_rows( rows, columns={ "lab_result_id": COL_LAB_RESULT_ID, "patient_identifier": COL_PATIENT_ID, "test_name": COL_TEST_NAME, "value": COL_VALUE, "collected_datetime": COL_COLLECTED, "loinc_code": COL_LOINC, "unit": COL_UNIT, "reference_low": COL_REF_LOW, "reference_high": COL_REF_HIGH, "encounter_id": COL_ENCOUNTER_ID, "practitioner_id": COL_PRACTITIONER_ID, }, ) print(f"{len(results)} lab results")

Data profile¶

In [ ]:

from collections import Counter

patients = Counter(r.patient_identifier for r in results)
stays = Counter(r.encounter_id for r in results if r.encounter_id)
no_encounter = sum(1 for r in results if not r.encounter_id)
no_loinc = sum(1 for r in results if not r.loinc_code)
comparators = sum(1 for r in results if str(r.value).startswith(("<", ">")))

print(f"{len(results)} results | {len(patients)} patients | {len(stays)} stays")
print(f"{no_encounter} rows without an encounter (pre-admission / post-discharge)")
print(f"{no_loinc} rows without a LOINC code, {comparators} comparator values")
print()
top_tests = Counter(r.test_name for r in results).most_common(8)
width = max(n for _, n in top_tests)
for name, n in top_tests:
    print(f"{name:32s} {'#' * (40 * n // width):40s} {n}")

from collections import Counter patients = Counter(r.patient_identifier for r in results) stays = Counter(r.encounter_id for r in results if r.encounter_id) no_encounter = sum(1 for r in results if not r.encounter_id) no_loinc = sum(1 for r in results if not r.loinc_code) comparators = sum(1 for r in results if str(r.value).startswith(("<", ">"))) print(f"{len(results)} results | {len(patients)} patients | {len(stays)} stays") print(f"{no_encounter} rows without an encounter (pre-admission / post-discharge)") print(f"{no_loinc} rows without a LOINC code, {comparators} comparator values") print() top_tests = Counter(r.test_name for r in results).most_common(8) width = max(n for _, n in top_tests) for name, n in top_tests: print(f"{name:32s} {'#' * (40 * n // width):40s} {n}")

7. Connect to FHIR and Cavell API¶

In [ ]:

import getpass
import os

from cavell_client import CavellClient

# Point CAVELL_API_URL at your Prism deployment and provide your LLM Gateway
# key (prompted below when the CAVELL_API_KEY environment variable is unset).
# The key only authenticates — deterministic ingestion spends no tokens.
CAVELL_API_URL = os.environ.get("CAVELL_API_URL", "https://prd.prism.cavell.app/api")
CAVELL_API_KEY = os.environ.get("CAVELL_API_KEY") or getpass.getpass(
    "LLM Gateway key: "
)

client = CavellClient(
    api_url=CAVELL_API_URL,
    api_key=CAVELL_API_KEY,
    fhir_base_url=FHIR_BASE_URL,
    fhir_api_path="/fhir",
)

print("Connected.")

import getpass import os from cavell_client import CavellClient # Point CAVELL_API_URL at your Prism deployment and provide your LLM Gateway # key (prompted below when the CAVELL_API_KEY environment variable is unset). # The key only authenticates — deterministic ingestion spends no tokens. CAVELL_API_URL = os.environ.get("CAVELL_API_URL", "https://prd.prism.cavell.app/api") CAVELL_API_KEY = os.environ.get("CAVELL_API_KEY") or getpass.getpass( "LLM Gateway key: " ) client = CavellClient( api_url=CAVELL_API_URL, api_key=CAVELL_API_KEY, fhir_base_url=FHIR_BASE_URL, fhir_api_path="/fhir", ) print("Connected.")

8. Ingest¶

One call: validate rows, resolve references fail-closed, send per patient, persist the returned bundles. No tier to pick and no batching to think about — the endpoint is deterministic and cheap.

In [ ]:

from cavell_client import LabIngestionPipeline

pipeline = LabIngestionPipeline(client)
outcome = pipeline.ingest(results)

print(outcome)
for mrn, persist in outcome.persistence:
    print(f"  {mrn}: {persist.created} created, {persist.updated} already present")

from cavell_client import LabIngestionPipeline pipeline = LabIngestionPipeline(client) outcome = pipeline.ingest(results) print(outcome) for mrn, persist in outcome.persistence: print(f" {mrn}: {persist.created} created, {persist.updated} already present")

9. Rejection report¶

Every skipped row, with its input position, id, stage and reason — client-side validation, client-side reference resolution and API-side content checks all land in the same list.

In [ ]:

from collections import Counter

for r in outcome.rejected:
    print(f"row {r.index:3d}  {r.lab_result_id or '—':9s} [{r.stage:10s}] {r.reason}")

stages = Counter(r.stage for r in outcome.rejected)
print(f"\n{dict(stages)}")

# The 7 deliberately broken rows, and nothing else
assert len(outcome.rejected) == 7, f"expected 7 rejections, got {len(outcome.rejected)}"
assert stages == Counter({"reference": 5, "validation": 1, "server": 1})
assert outcome.accepted == len(results) - 7
assert outcome.success
print("All 244 valid rows accepted; all 7 broken rows rejected with reasons.")

from collections import Counter for r in outcome.rejected: print(f"row {r.index:3d} {r.lab_result_id or '—':9s} [{r.stage:10s}] {r.reason}") stages = Counter(r.stage for r in outcome.rejected) print(f"\n{dict(stages)}") # The 7 deliberately broken rows, and nothing else assert len(outcome.rejected) == 7, f"expected 7 rejections, got {len(outcome.rejected)}" assert stages == Counter({"reference": 5, "validation": 1, "server": 1}) assert outcome.accepted == len(results) - 7 assert outcome.success print("All 244 valid rows accepted; all 7 broken rows rejected with reasons.")

10. Inspect what landed¶

Pull the hyperkalaemia readmission (V-014, MRN-20003) and look at the Observations: LOINC codings with display names resolved offline, UCUM units, reference ranges, H/L/N interpretations, timestamps intact — and no unvalidated tag, because structured source data is not an extraction awaiting review. Then contrast a pre-admission draw (no encounter) with an in-stay one, and a text-only code with a LOINC-coded one.

In [ ]:

LAB_IDENTIFIER_SYSTEM = "urn:cavell:lab-result"


def lab_observations(mrn: str) -> list[dict]:
    fhir_id = client.find_patient_id(mrn)
    assert fhir_id, f"patient {mrn} not found"
    return [
        o
        for o in client.get_patient_resources(fhir_id, "Observation")
        if any(
            i.get("system") == LAB_IDENTIFIER_SYSTEM for i in o.get("identifier", [])
        )
    ]


def show(o: dict) -> str:
    code_ = o["code"]
    coding = (code_.get("coding") or [{}])[0]
    label = f"{code_['text']} [{coding.get('code', 'text-only')}]"
    if "valueQuantity" in o:
        q = o["valueQuantity"]
        value = f"{q.get('comparator', '')}{q['value']} {q.get('unit', '')}".strip()
    else:
        value = repr(o.get("valueString"))
    interp = (o.get("interpretation") or [{}])[0].get("coding", [{}])[0].get("code", "")
    enc = o.get("encounter", {}).get("reference", "(no encounter)")
    when = o.get("effectiveDateTime", "")
    return f"{when:26s} {label:44s} {value:22s} {interp:2s} {enc}"


almeida = lab_observations("MRN-20003")
print(f"MRN-20003 has {len(almeida)} lab Observations\n")
for o in sorted(almeida, key=lambda o: o.get("effectiveDateTime", "")):
    print(show(o))

# Structured source data: final status, no 'unvalidated' meta tag
assert all(o["status"] == "final" for o in almeida)
assert not any(
    tag.get("code") == "unvalidated"
    for o in almeida
    for tag in o.get("meta", {}).get("tag", [])
)

# Pre-admission draws (no encounter) sit next to in-stay ones for the same patient
assert any("encounter" not in o for o in almeida), "expected pre-admission draws"
assert any("encounter" in o for o in almeida), "expected in-stay draws"

# The V-009 serial troponins kept their timestamps (same day, hours apart)
nowak = lab_observations("MRN-20009")
troponins = sorted(
    (o for o in nowak if (o["code"].get("coding") or [{}])[0].get("code") == "67151-1"),
    key=lambda o: o["effectiveDateTime"],
)
print("\nSerial troponins (V-009):")
for o in troponins:
    print(f"  {o['effectiveDateTime']}  {o['valueQuantity']['value']} ng/L")
assert len(troponins) >= 4

# Text-only coding fallback for the no-LOINC wound swab
sandberg = lab_observations("MRN-20013")
swab = next(o for o in sandberg if o["code"]["text"] == "Wound swab culture")
assert not swab["code"].get("coding"), "no-LOINC row should have a text-only code"
assert swab.get("valueString"), "qualitative result should be a valueString"
print(f"\nText-only code: {swab['code']}\n  value: {swab['valueString']}")

LAB_IDENTIFIER_SYSTEM = "urn:cavell:lab-result" def lab_observations(mrn: str) -> list[dict]: fhir_id = client.find_patient_id(mrn) assert fhir_id, f"patient {mrn} not found" return [ o for o in client.get_patient_resources(fhir_id, "Observation") if any( i.get("system") == LAB_IDENTIFIER_SYSTEM for i in o.get("identifier", []) ) ] def show(o: dict) -> str: code_ = o["code"] coding = (code_.get("coding") or [{}])[0] label = f"{code_['text']} [{coding.get('code', 'text-only')}]" if "valueQuantity" in o: q = o["valueQuantity"] value = f"{q.get('comparator', '')}{q['value']} {q.get('unit', '')}".strip() else: value = repr(o.get("valueString")) interp = (o.get("interpretation") or [{}])[0].get("coding", [{}])[0].get("code", "") enc = o.get("encounter", {}).get("reference", "(no encounter)") when = o.get("effectiveDateTime", "") return f"{when:26s} {label:44s} {value:22s} {interp:2s} {enc}" almeida = lab_observations("MRN-20003") print(f"MRN-20003 has {len(almeida)} lab Observations\n") for o in sorted(almeida, key=lambda o: o.get("effectiveDateTime", "")): print(show(o)) # Structured source data: final status, no 'unvalidated' meta tag assert all(o["status"] == "final" for o in almeida) assert not any( tag.get("code") == "unvalidated" for o in almeida for tag in o.get("meta", {}).get("tag", []) ) # Pre-admission draws (no encounter) sit next to in-stay ones for the same patient assert any("encounter" not in o for o in almeida), "expected pre-admission draws" assert any("encounter" in o for o in almeida), "expected in-stay draws" # The V-009 serial troponins kept their timestamps (same day, hours apart) nowak = lab_observations("MRN-20009") troponins = sorted( (o for o in nowak if (o["code"].get("coding") or [{}])[0].get("code") == "67151-1"), key=lambda o: o["effectiveDateTime"], ) print("\nSerial troponins (V-009):") for o in troponins: print(f" {o['effectiveDateTime']} {o['valueQuantity']['value']} ng/L") assert len(troponins) >= 4 # Text-only coding fallback for the no-LOINC wound swab sandberg = lab_observations("MRN-20013") swab = next(o for o in sandberg if o["code"]["text"] == "Wound swab culture") assert not swab["code"].get("coding"), "no-LOINC row should have a text-only code" assert swab.get("valueString"), "qualitative result should be a valueString" print(f"\nText-only code: {swab['code']}\n value: {swab['valueString']}")

11. Idempotency¶

Run the exact same feed again. The conditional creates match every existing Observation, so nothing is created twice — the outcome reports them as already present instead.

In [ ]:

rerun = pipeline.ingest(results)

print(rerun)
assert rerun.created == 0, "a re-run must create nothing"
assert rerun.skipped_existing == rerun.accepted
assert rerun.success
print("Re-run created nothing — every accepted row matched its existing Observation.")

rerun = pipeline.ingest(results) print(rerun) assert rerun.created == 0, "a re-run must create nothing" assert rerun.skipped_existing == rerun.accepted assert rerun.success print("Re-run created nothing — every accepted row matched its existing Observation.")