Merge pull request #33 from Aidbox/atomic-ehr-codegen-python-us-core-profiles-update

Atomic ehr codegen python us core profiles update

Author: ryukzak

developer-experience/atomic-ehr-codegen-python-us-core-profiles/README.md

@@ -5,7 +5,7 @@ A small CSV-to-FHIR converter demonstrating [`@atomic-ehr/codegen`](https://gith
 The example:
 
 1. generates profile classes for [US Core Patient](https://www.hl7.org/fhir/us/core/StructureDefinition-us-core-patient.html) and [US Core Blood Pressure](https://www.hl7.org/fhir/us/core/StructureDefinition-us-core-blood-pressure.html) plus base `Bundle` from `hl7.fhir.r4.core`,
-2. loads `patients.csv` (5 rows: MRN, name, demographics, race, one BP reading each),
+2. loads `patients.csv` (each row: MRN, name, demographics, race, one BP reading each),
 3. converts each row into a validated `UscorePatientProfile` + `UscoreBloodPressureProfile`,
 4. packages them as a `Bundle[Patient | Observation]` transaction with `urn:uuid` cross-references,
 5. reads `bundle.json` back, selects the US Core BP observations, and prints the average BP.
@@ -55,6 +55,8 @@ Run [Aidbox](https://www.health-samurai.io/fhir-server) locally:
 curl -JO https://aidbox.app/runme && docker compose up -d
 ```
 
+Open `http://localhost:8080` and issue license if not already done.
+
 Then POST `bundle.json` and read the stored observations back as typed resources:
 
 ```bash

developer-experience/atomic-ehr-codegen-python-us-core-profiles/avg.py

@@ -9,9 +9,10 @@
 def is_us_core_bp(resource: dict) -> bool:
     # Python profiles have no `is()` type guard (unlike the TS API); select on
     # resourceType + meta.profile, then hand the survivors to from_resource().
-    return (
-        resource.get("resourceType") == "Observation"
-        and UscoreBloodPressureProfile.canonical_url in (resource.get("meta", {}).get("profile") or [])
+    return resource.get(
+        "resourceType"
+    ) == "Observation" and UscoreBloodPressureProfile.canonical_url in (
+        resource.get("meta", {}).get("profile") or []
     )
 
 
@@ -20,16 +21,24 @@ def main() -> None:
         bundle = json.load(f)
 
     bps = [
-        UscoreBloodPressureProfile.from_resource(Observation.model_validate(entry["resource"]))
+        UscoreBloodPressureProfile.from_resource(
+            Observation.model_validate(entry["resource"])
+        )
         for entry in bundle.get("entry", [])
         if is_us_core_bp(entry["resource"])
     ]
 
     def avg(xs: list[float]) -> float:
         return sum(xs) / len(xs)
 
-    systolic = [bp.get_systolic()["value"] for bp in bps]
-    diastolic = [bp.get_diastolic()["value"] for bp in bps]
+    systolic = [
+        systolic["value"] for bp in bps if (systolic := bp.get_systolic()) is not None
+    ]
+    diastolic = [
+        diastolic["value"]
+        for bp in bps
+        if (diastolic := bp.get_diastolic()) is not None
+    ]
 
     print(f"Avg BP: {avg(systolic):.1f}/{avg(diastolic):.1f} mmHg (n={len(bps)})")
 

developer-experience/atomic-ehr-codegen-python-us-core-profiles/fhir_types/README.md

@@ -13,7 +13,7 @@ const main = async () => {
           "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient": {},
           "http://hl7.org/fhir/us/core/StructureDefinition/us-core-blood-pressure": {},
         },
-        "hl7.fhir.r4.core": {
+        "hl7.fhir.r4.core": { // hl7.fhir.us.core depends on hl7.fhir.r4.core
           "http://hl7.org/fhir/StructureDefinition/Bundle": {},
         },
       },

developer-experience/atomic-ehr-codegen-python-us-core-profiles/generate.ts

@@ -5,10 +5,17 @@
 import uuid
 import warnings
 
-from fhir_types.hl7_fhir_r4_core.base import Identifier, HumanName, Reference
-from fhir_types.hl7_fhir_r4_core.patient import Patient
-from fhir_types.hl7_fhir_r4_core.observation import Observation
-from fhir_types.hl7_fhir_r4_core.bundle import Bundle, BundleEntry, BundleEntryRequest
+from fhir_types.hl7_fhir_r4_core import (
+    Identifier,
+    HumanName,
+    Reference,
+    Patient,
+    Observation,
+    Bundle,
+    BundleEntry,
+    BundleEntryRequest,
+)
+
 from fhir_types.hl7_fhir_us_core.profiles import (
     UscorePatientProfile,
     UscoreBloodPressureProfile,
@@ -22,16 +29,22 @@
 def row_to_patient(row: dict[str, str]) -> UscorePatientProfile:
     base_patient = Patient(
         resource_type="Patient",
-        identifier=[Identifier(system="http://hospital.example.org/mrn", value=row["mrn"])],
+        identifier=[
+            Identifier(system="http://hospital.example.org/mrn", value=row["mrn"])
+        ],
         name=[HumanName(family=row["family"], given=[row["given"]])],
-        gender=row["gender"],          # Literal-typed; Pydantic validates the value
-        birth_date=row["birthDate"],   # snake_case attr; serializes back to "birthDate"
+        gender=row["gender"],  # Literal-typed; Pydantic validates the value
+        birth_date=row["birthDate"],  # snake_case attr; serializes back to "birthDate"
     )
 
     patient = UscorePatientProfile.apply(base_patient)
     patient.set_race(
         {
-            "ombCategory": {"system": "urn:oid:2.16.840.1.113883.6.238", "code": row["raceCode"], "display": row["raceDisplay"]},
+            "ombCategory": {
+                "system": "urn:oid:2.16.840.1.113883.6.238",
+                "code": row["raceCode"],
+                "display": row["raceDisplay"],
+            },
             "text": row["raceDisplay"],
         }
     )
@@ -46,8 +59,22 @@ def row_to_bp(row: dict[str, str], patient_urn: str) -> UscoreBloodPressureProfi
 
     (
         bp.set_effective_date_time(row["effectiveDateTime"])
-          .set_systolic({"value": float(row["systolic"]), "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]"})
-          .set_diastolic({"value": float(row["diastolic"]), "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]"})
+        .set_systolic(
+            {
+                "value": float(row["systolic"]),
+                "unit": "mmHg",
+                "system": "http://unitsofmeasure.org",
+                "code": "mm[Hg]",
+            }
+        )
+        .set_diastolic(
+            {
+                "value": float(row["diastolic"]),
+                "unit": "mmHg",
+                "system": "http://unitsofmeasure.org",
+                "code": "mm[Hg]",
+            }
+        )
     )
 
     errors = bp.validate()["errors"]
@@ -62,10 +89,16 @@ def row_to_entries(row: dict[str, str]) -> list[BundleEntry[Patient | Observatio
     bp = row_to_bp(row, patient_urn)
 
     return [
-        BundleEntry(full_url=patient_urn, resource=patient.to_resource(),
-                    request=BundleEntryRequest(method="POST", url="Patient")),
-        BundleEntry(full_url=f"urn:uuid:{uuid.uuid4()}", resource=bp.to_resource(),
-                    request=BundleEntryRequest(method="POST", url="Observation")),
+        BundleEntry(
+            full_url=patient_urn,
+            resource=patient.to_resource(),
+            request=BundleEntryRequest(method="POST", url="Patient"),
+        ),
+        BundleEntry(
+            full_url=f"urn:uuid:{uuid.uuid4()}",
+            resource=bp.to_resource(),
+            request=BundleEntryRequest(method="POST", url="Observation"),
+        ),
     ]
 
 
@@ -82,7 +115,10 @@ def main() -> None:
 
     with open("bundle.json", "w") as f:
         json.dump(bundle.model_dump(by_alias=True, exclude_none=True), f, indent=2)
-    print(f"Wrote bundle with {len(bundle.entry)} entries")
+    entries = bundle.entry
+    if entries is None:
+        raise ValueError("entries is None")
+    print(f"Wrote bundle with {len(entries)} entries")
 
 
 if __name__ == "__main__":

developer-experience/atomic-ehr-codegen-python-us-core-profiles/load.py

@@ -8,7 +8,7 @@
     "generate": "tsx generate.ts"
   },
   "devDependencies": {
-    "@atomic-ehr/codegen": "^0.0.15",
+    "@atomic-ehr/codegen": "^0.0.15-canary.20260616131237.0f30eea",
     "@types/node": "^25.6.0",
     "tsx": "^4.21.0",
     "typescript": "^5.9.3"

developer-experience/atomic-ehr-codegen-python-us-core-profiles/package.json

@@ -26,8 +26,14 @@
 
 
 def make_client() -> AsyncFHIRClient:
-    url = os.environ.get("FHIR_URL") or os.environ.get("AIDBOX_URL", "http://localhost:8080/fhir")
-    dump = lambda r: r.model_dump(by_alias=True, exclude_none=True)
+    url = (
+        os.environ.get("FHIR_URL")
+        or os.environ.get("AIDBOX_URL")
+        or "http://localhost:8080/fhir"
+    )
+
+    def dump(r):
+        return r.model_dump(by_alias=True, exclude_none=True)
 
     # Basic auth when a secret is provided (Aidbox); otherwise an open server.
     secret = os.environ.get("AIDBOX_SECRET")
@@ -50,7 +56,11 @@ async def main() -> None:
     print("Transaction committed:", response.get("type"))
 
     # Read the stored US Core Blood Pressure observations back as typed resources.
-    observations = await client.resources(Observation).search(code="http://loinc.org|85354-9").fetch()
+    observations = (
+        await client.resources(Observation)
+        .search(code="http://loinc.org|85354-9")
+        .fetch()
+    )
     print(f"Stored BP observations: {len(observations)}")
     for obs in observations:
         assert isinstance(obs, Observation)

developer-experience/atomic-ehr-codegen-python-us-core-profiles/post.py

@@ -47,7 +47,11 @@ Run [Aidbox](https://www.health-samurai.io/fhir-server) locally and POST `bundle
 ```bash
 curl -JO https://aidbox.app/runme && docker compose up -d
 SECRET=$(awk '/BOX_ROOT_CLIENT_SECRET:/{print $2}' docker-compose.yaml)
+```
+
+Open `http://localhost:8080` and issue license if not already done.
 
+```bash
 curl -u "root:$SECRET" -X POST -H "Content-Type: application/fhir+json" \
   -d @bundle.json http://localhost:8080/fhir
 ```