Add atomic-ehr-codegen Python US Core profiles example

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

@@ -0,0 +1,7 @@
+venv/
+__pycache__/
+*.pyc
+.codegen-cache/
+node_modules/
+.idea/
+bundle.json

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

@@ -0,0 +1,73 @@
+# US Core Profiles in Python with @atomic-ehr/codegen
+
+A small CSV-to-FHIR converter demonstrating [`@atomic-ehr/codegen`](https://github.com/atomic-ehr/codegen) profile class generation for US Core. The Python counterpart of [atomic-ehr-codegen-typescript-us-core-profiles](../atomic-ehr-codegen-typescript-us-core-profiles), generated for **Pydantic** with the **[fhirpy](https://github.com/beda-software/fhirpy)** async client enabled (`fhirpyClient: true`).
+
+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),
+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.
+
+## Files
+
+| File | Purpose |
+|------|---------|
+| `generate.ts` | Runs `@atomic-ehr/codegen` to produce typed profile classes in `fhir_types/` (Node step) |
+| `fhir_types/` | Generated output (committed so you can browse without running the generator) |
+| `patients.csv` | Sample input (5 rows) |
+| `load.py` | Parses CSV, builds the typed Bundle, writes `bundle.json` |
+| `avg.py` | Reads `bundle.json` back, selects US Core BP, computes average BP |
+| `post.py` | Optional: POSTs the bundle to a FHIR server with fhirpy's async client |
+
+## Run It
+
+Code generation runs through the Node tool; the application code is pure Python.
+
+```bash
+npm install
+npx tsx generate.ts                           # regenerate fhir_types/ (optional -- already committed)
+
+python3 -m venv venv && source venv/bin/activate
+pip install -r fhir_types/requirements.txt    # pydantic + fhirpy
+
+python load.py                                # reads patients.csv, writes bundle.json
+python avg.py                                 # reads bundle.json, prints the average BP
+```
+
+Expected output:
+
+```
+$ python load.py
+Loaded 5 rows
+Wrote bundle with 10 entries
+
+$ python avg.py
+Avg BP: 125.2/82.0 mmHg (n=5)
+```
+
+## POST to a FHIR Server with fhirpy (Optional)
+
+Run [Aidbox](https://www.health-samurai.io/fhir-server) locally:
+
+```bash
+curl -JO https://aidbox.app/runme && docker compose up -d
+```
+
+Then POST `bundle.json` and read the stored observations back as typed resources:
+
+```bash
+export AIDBOX_SECRET=$(awk '/BOX_ROOT_CLIENT_SECRET:/{print $2}' docker-compose.yaml)
+python post.py
+```
+
+`post.py` builds an `AsyncFHIRClient`, POSTs the transaction (Aidbox resolves the `urn:uuid` references on commit), then `client.resources(Observation).search(...).fetch()` returns the stored observations deserialized into the generated `Observation` class.
+
+## Notes on the Code
+
+- **The generator is a Node tool; the output is Python.** `generate.ts` runs once to emit `fhir_types/`. After that you only need Python + Pydantic (and fhirpy for `post.py`).
+- **`fhirpyClient: true`** makes the generated resources extend `FhirpyBaseModel`: they expose `resourceType` at class level and serialize via `model_dump`, which is everything fhirpy's typed client needs to `create` / `search` / `fetch` them.
+- **snake_case attributes, FHIR-correct JSON.** Construct with Python names (`birth_date`, `effective_date_time`); `model_dump(by_alias=True, exclude_none=True)` emits camelCase FHIR JSON.
+- **Must-support base fields** (`gender`, `birth_date`) aren't profiled further by US Core, so the profile class emits no `.set_gender()`-style setters. `load.py` sets them on the base `Patient`, then calls `UscorePatientProfile.apply()`. `validate()` warns if a must-support field is missing.
+- **No `is()` type guard.** Unlike the TypeScript API, the Python classes don't ship a `.filter()`-style guard. `avg.py` selects BP observations by `resourceType` + `meta.profile`, then calls `from_resource()`.

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

@@ -0,0 +1,38 @@
+"""Read bundle.json back, select US Core Blood Pressure observations, print average BP."""
+
+import json
+
+from fhir_types.hl7_fhir_r4_core.observation import Observation
+from fhir_types.hl7_fhir_us_core.profiles import UscoreBloodPressureProfile
+
+
+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 [])
+    )
+
+
+def main() -> None:
+    with open("bundle.json") as f:
+        bundle = json.load(f)
+
+    bps = [
+        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]
+
+    print(f"Avg BP: {avg(systolic):.1f}/{avg(diastolic):.1f} mmHg (n={len(bps)})")
+
+
+if __name__ == "__main__":
+    main()