From bd92126384d2dd56d7f959b7dac6f34de2674671 Mon Sep 17 00:00:00 2001 From: Kisa Date: Tue, 7 Jul 2026 05:20:45 -0400 Subject: [PATCH] feat: patient rollup + nested patients payload (plan 02, P3) core/rollup.py implements the locked truth table: worst gradeable line wins; a Plan Type Needed line caps the patient at Action Needed; non-gradeable lines are displayed but never touch the light; no-graded-lines patients carry None (no label invented). /api/upload gains additive patients[] (one entry per patient, nested device lines with doc checklists, citations, timing flags, dedup display fields) and patient_stats (counts patients, not rows). Flat records payload untouched until P6. Independent audit: SHIP; its LOW findings fixed (date coercion on timing keys, empty-rollup guard, severity divergence documented). 151 tests green. Co-Authored-By: Claude Fable 5 --- python-backend/api/main.py | 174 ++++++++++++++-- python-backend/core/rollup.py | 101 +++++++++ python-backend/core/worklist_readiness.py | 31 +-- tests/test_rollup.py | 238 ++++++++++++++++++++++ 4 files changed, 518 insertions(+), 26 deletions(-) create mode 100644 python-backend/core/rollup.py create mode 100644 tests/test_rollup.py diff --git a/python-backend/api/main.py b/python-backend/api/main.py index 13a7ba8..b9ba1e2 100644 --- a/python-backend/api/main.py +++ b/python-backend/api/main.py @@ -185,6 +185,36 @@ class ReadinessItemOut(BaseModel): rule_id: Optional[str] = None +class LineOut(BaseModel): + """One coverage line (patient x device x plan type) in the nested payload.""" + + device_type: str + device_display: str + plan_type: Optional[str] = None # None = plan type needed (never guessed) + gradeable: bool + line_status: Optional[str] = None # LineStatus verbatim; None = ungraded + timing_flag: Optional[str] = None # CoverageFlag of the current shipment + doc_items: Optional[list[ReadinessItemOut]] = None + # Display fields per the locked dedup design (section 6) + order_numbers: list[str] = [] + hcpcs_codes: list[str] = [] + total_quantity: int = 0 + shipment_date: Optional[str] = None # current shipment DOS + payer: Optional[str] = None + prior_shipment_count: int = 0 + + +class PatientOut(BaseModel): + """One patient in the nested payload; tabs count these, not rows (P6).""" + + patient_id: str + rollup_status: Optional[str] = None # None = no graded gradeable lines + plan_type_needed: bool = False + cgm_line_count: int = 0 + total_line_count: int = 0 + lines: list[LineOut] = [] + + class RecordOut(BaseModel): patient_id: str device_type: str @@ -234,6 +264,10 @@ class UploadResponse(BaseModel): mapping_summary: dict batch_id: str | None = None readiness_stats: dict | None = None + # Nested patient payload (P3, additive). The flat `records` list stays + # untouched until P6 switches the UI (locked migration strategy). + patients: list[PatientOut] | None = None + patient_stats: dict | None = None def _build_reason( @@ -333,6 +367,24 @@ class DocStatusRequest(BaseModel): expiry_date: Optional[str] = None +def _readiness_items_out(readiness) -> Optional[list[ReadinessItemOut]]: + """Serialize a ReadinessVerdict's doc items (shared by rows and lines).""" + if not readiness: + return None + return [ + ReadinessItemOut( + doc_type=i.doc_type, + required_state=i.required_state.value, + input_state=i.input_state.value, + quality=i.quality.value, + value=i.value, + satisfied=i.satisfied, + rule_id=i.rule_id, + ) + for i in readiness.doc_items + ] + + def _to_record_out( r, record=None, @@ -422,20 +474,7 @@ def _to_record_out( else None, plan_type=getattr(record, "plan_type", None) if record else None, readiness_status=readiness.line_status.value if readiness else None, - readiness_items=[ - ReadinessItemOut( - doc_type=i.doc_type, - required_state=i.required_state.value, - input_state=i.input_state.value, - quality=i.quality.value, - value=i.value, - satisfied=i.satisfied, - rule_id=i.rule_id, - ) - for i in readiness.doc_items - ] - if readiness - else None, + readiness_items=_readiness_items_out(readiness), ) @@ -461,6 +500,104 @@ def _compute_stats(records: list[RecordOut]) -> dict: } +def _build_patients_payload(index, reachable_line_keys, results): + """Assemble the nested patients payload (P3) from the lines reachable in + THIS response, so patients/lines always reconcile with the flat records. + Rollup rules are locked in core/rollup.py (worst gradeable line; PTN caps + at Action Needed; non-gradeable lines excluded from the light).""" + from core.dedup import UNKNOWN_PLAN_TYPE + from core.rollup import rollup_patients + + reachable = { + k: index.lines[k] for k in reachable_line_keys if k in index.lines + } + if not reachable: + return [], {"patients_total": 0} + + from core.dedup import _as_date + + # Line-level timing flag = the worst-urgency row of the line's CURRENT + # shipment group (twin rows can differ; worst wins, fail-safe). + flag_by_group: dict = {} + for r in results: + gk = (r.patient_id, r.device_type, _as_date(r.last_shipment_date)) + flag_val = r.flag.value if hasattr(r.flag, "value") else str(r.flag) + prev = flag_by_group.get(gk) + if prev is None or r.priority_score > prev[1]: + flag_by_group[gk] = (flag_val, r.priority_score) + + # DISPLAY ordering, intentionally different from rollup.SEVERITY: the + # worklist queues At Risk above Plan Type Needed (urgency), while the + # rollup's capping rule treats PTN as the worst LINE state. Do not unify. + severity = { + "At Risk": 4, + "Plan Type Needed": 3, + "Action Needed": 2, + "On Track": 1, + "Clear to Ship": 0, + } + patients = [] + for pid, ru in rollup_patients(reachable).items(): + lines_out = [] + for line in sorted( + ru.lines, + key=lambda l: ( + -(severity.get(l.line_status, -1)), + l.device_type, + ), + ): + line_key = (line.patient_id, line.device_type, line.plan_type) + verdict = index.by_line.get(line_key) + cur = line.current_shipment + timing = flag_by_group.get( + (line.patient_id, line.device_type, _as_date(cur.shipment_date)) + ) + lines_out.append( + LineOut( + device_type=line.device_type, + device_display=DEVICE_DISPLAY.get( + line.device_type, line.device_type + ), + plan_type=None + if line.plan_type == UNKNOWN_PLAN_TYPE + else line.plan_type, + gradeable=line.gradeable, + line_status=line.line_status, + timing_flag=timing[0] if timing else None, + doc_items=_readiness_items_out(verdict), + order_numbers=cur.order_numbers, + hcpcs_codes=cur.hcpcs_codes, + total_quantity=cur.total_quantity, + shipment_date=cur.shipment_date.isoformat(), + payer=cur.payer or None, + prior_shipment_count=len(line.prior_shipments), + ) + ) + patients.append( + PatientOut( + patient_id=pid, + rollup_status=ru.rollup_status.value + if ru.rollup_status + else None, + plan_type_needed=ru.plan_type_needed, + cgm_line_count=ru.cgm_line_count, + total_line_count=ru.total_line_count, + lines=lines_out, + ) + ) + + # Worklist order: worst patients first, stable by patient_id. + patients.sort( + key=lambda p: (-(severity.get(p.rollup_status, -1)), p.patient_id) + ) + + stats: dict = {"patients_total": len(patients)} + for p in patients: + key = p.rollup_status if p.rollup_status else "ungraded" + stats[key] = stats.get(key, 0) + 1 + return patients, stats + + @app.get("/health") def health(): return {"status": "ok", "service": "signal-api", "version": "1.0.0"} @@ -551,10 +688,13 @@ async def upload_csv( out = [] used_lines = {} + reachable_line_keys = set() for r in results: verdict, line_key, merged = resolve_group( index, r.patient_id, r.device_type, r.last_shipment_date ) + if line_key is not None: + reachable_line_keys.add(line_key) if verdict is not None and line_key is not None: used_lines[line_key] = verdict # Exact row match first (byte-identical legacy path). CoverageResult @@ -610,6 +750,10 @@ async def upload_csv( clerk_org_id=clerk_org_id, ) + patients_out, patient_stats = _build_patients_payload( + index, reachable_line_keys, results + ) + return UploadResponse( records=out, total=len(out), @@ -622,6 +766,8 @@ async def upload_csv( # reconcile with the records beside them (a record calculate_batch # skipped cannot leave a phantom line in the counts). readiness_stats=readiness_stats(used_lines), + patients=patients_out, + patient_stats=patient_stats, ) diff --git a/python-backend/core/rollup.py b/python-backend/core/rollup.py new file mode 100644 index 0000000..1d0097a --- /dev/null +++ b/python-backend/core/rollup.py @@ -0,0 +1,101 @@ +""" +rollup.py +Signal — STTIL Solutions + +Patient-level rollup of coverage-line readiness verdicts (build plan 02, P3). +Design locked in readiness-model-brief section 4 and the phase-2 design Item 3: + +- The light = worst gradeable line. Severity (worst first): + Plan Type Needed > At Risk > Action Needed > On Track > Clear to Ship. +- Any gradeable Plan Type Needed line caps the patient no greener than + Action Needed (the patient IS being served; one unmapped plan type must not + read as the whole patient being ungradeable). +- Non-gradeable lines (non-CGM) never participate in the light or the counts. +- Tabs count patients, not rows. + +A patient with no graded gradeable lines has rollup_status None — the UI +treatment is an open Kisa question; no label is invented here. + +PHI CONTRACT: operates on patient_id only; this module logs nothing. +""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from typing import Optional + +from core.dedup import CoverageLine +from core.readiness import LineStatus + +# Higher = worse. Plan Type Needed ranks worst for LINE severity ordering; +# its patient-level effect is the capping rule below, not raw worstness. +SEVERITY = { + LineStatus.CLEAR_TO_SHIP: 0, + LineStatus.ON_TRACK: 1, + LineStatus.ACTION_NEEDED: 2, + LineStatus.AT_RISK: 3, + LineStatus.PLAN_TYPE_NEEDED: 4, +} + + +@dataclass +class PatientRollup: + patient_id: str + lines: list[CoverageLine] # all lines, incl. non-gradeable + rollup_status: Optional[LineStatus] # None = no graded gradeable lines + cgm_line_count: int # gradeable lines + total_line_count: int # all lines + plan_type_needed: bool # any gradeable line in PTN + + +def rollup_patient(lines: list[CoverageLine]) -> PatientRollup: + """Roll one patient's coverage lines into a single verdict, per the + locked truth table (plan 02 P3): + + | gradeable statuses present | rollup_status | + | none | None | + | all Plan Type Needed | Plan Type Needed | + | some PTN + others | worse of (Action Needed, worst other)| + | no PTN | worst by severity | + """ + if not lines: + raise ValueError("rollup_patient requires at least one CoverageLine") + graded = [ + l for l in lines if l.gradeable and l.line_status is not None + ] + statuses = [LineStatus(l.line_status) for l in graded] + ptn = [s for s in statuses if s == LineStatus.PLAN_TYPE_NEEDED] + + if not statuses: + status: Optional[LineStatus] = None + elif len(ptn) == len(statuses): + status = LineStatus.PLAN_TYPE_NEEDED + elif ptn: + worst_other = max( + (s for s in statuses if s != LineStatus.PLAN_TYPE_NEEDED), + key=SEVERITY.get, + ) + status = max(worst_other, LineStatus.ACTION_NEEDED, key=SEVERITY.get) + else: + status = max(statuses, key=SEVERITY.get) + + return PatientRollup( + patient_id=lines[0].patient_id, + lines=lines, + rollup_status=status, + cgm_line_count=sum(1 for l in lines if l.gradeable), + total_line_count=len(lines), + plan_type_needed=bool(ptn), + ) + + +def rollup_patients(lines: dict) -> dict[str, PatientRollup]: + """Group CoverageLines (any keying) by patient and roll each up.""" + by_patient: dict[str, list[CoverageLine]] = defaultdict(list) + for line in lines.values(): + by_patient[line.patient_id].append(line) + return { + patient_id: rollup_patient(patient_lines) + for patient_id, patient_lines in by_patient.items() + } diff --git a/python-backend/core/worklist_readiness.py b/python-backend/core/worklist_readiness.py index 901c0f2..860a2ae 100644 --- a/python-backend/core/worklist_readiness.py +++ b/python-backend/core/worklist_readiness.py @@ -36,6 +36,7 @@ from typing import Optional from core.coverage_calculator import ShipmentRecord from core.dedup import ( UNKNOWN_PLAN_TYPE, + CoverageLine, MergedShipment, _as_date, assign_to_coverage_lines, @@ -54,11 +55,13 @@ class ReadinessIndex: """Dedup-derived resolution index for one upload batch. by_line: verdicts per gradeable coverage line + lines: every CoverageLine (graded ones carry line_status) line_by_group: dedup group -> the line that group's shipment belongs to merged_by_group: dedup group -> its MergedShipment (display-field truth) """ by_line: dict[LineKey, ReadinessVerdict] = field(default_factory=dict) + lines: dict[LineKey, "CoverageLine"] = field(default_factory=dict) line_by_group: dict[GroupKey, LineKey] = field(default_factory=dict) merged_by_group: dict[GroupKey, MergedShipment] = field(default_factory=dict) @@ -93,22 +96,26 @@ def build_readiness_index( index = ReadinessIndex() for key, line in lines.items(): + line_key = (key.patient_id, key.device_type, key.plan_type_id) + index.lines[line_key] = line if not line.gradeable: continue cur = line.current_shipment - index.by_line[(key.patient_id, key.device_type, key.plan_type_id)] = ( - evaluate_readiness( - line.plan_type if line.plan_type != UNKNOWN_PLAN_TYPE else None, - csv_swo_status=cur.csv_swo_status, - csv_visit_date=cur.csv_visit_date, - confirmed_visit_date=_confirmed_for( - key.patient_id, confirmed_visits - ), - csv_pecos_verified=cur.csv_pecos_verified, - csv_pa_status=cur.csv_pa_status, - csv_diagnosis_on_file=cur.csv_diagnosis_on_file, - ) + verdict = evaluate_readiness( + line.plan_type if line.plan_type != UNKNOWN_PLAN_TYPE else None, + csv_swo_status=cur.csv_swo_status, + csv_visit_date=cur.csv_visit_date, + confirmed_visit_date=_confirmed_for( + key.patient_id, confirmed_visits + ), + csv_pecos_verified=cur.csv_pecos_verified, + csv_pa_status=cur.csv_pa_status, + csv_diagnosis_on_file=cur.csv_diagnosis_on_file, ) + index.by_line[line_key] = verdict + # The grading step owns line_status (dedup leaves it None) — the + # rollup (P3) reads it from the line. + line.line_status = verdict.line_status.value for m in merged: gkey: GroupKey = (m.patient_id, m.device_type, m.shipment_date) diff --git a/tests/test_rollup.py b/tests/test_rollup.py new file mode 100644 index 0000000..1abe5ee --- /dev/null +++ b/tests/test_rollup.py @@ -0,0 +1,238 @@ +""" +Tests for the patient rollup (build plan 02, P3). + +Rollup rules locked in readiness-model-brief section 4 / phase-2 design Item 3: +worst gradeable line; any gradeable Plan Type Needed line caps the patient no +greener than Action Needed; non-gradeable lines never touch the light; a +patient with no graded gradeable lines has rollup_status None (no label +invented). + +PHI CONTRACT: synthetic patient_ids only. +""" + +import io +import sys +from datetime import date, timedelta +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from core.dedup import CoverageLine, MergedShipment # noqa: E402 +from core.readiness import LineStatus # noqa: E402 +from core.rollup import rollup_patient, rollup_patients # noqa: E402 + +TODAY = date.today() + + +def make_line(patient_id="PT-1", device="dexcom_g7", plan="medicare", + status=None, gradeable=True): + shipment = MergedShipment( + patient_id=patient_id, + device_type=device, + shipment_date=TODAY - timedelta(days=5), + total_quantity=3, + components=["sensor"], + order_numbers=[], + hcpcs_codes=[], + payer="Medicare Part B", + plan_type=None if plan == "unknown" else plan, + ) + return CoverageLine( + patient_id=patient_id, + device_type=device, + plan_type=plan, + gradeable=gradeable, + current_shipment=shipment, + prior_shipments=[], + line_status=status, + timing_flag=None, + ) + + +def test_patient_rollup_single_line(): + ru = rollup_patient([make_line(status="On Track")]) + assert ru.rollup_status == LineStatus.ON_TRACK + assert ru.cgm_line_count == 1 + assert ru.total_line_count == 1 + + +def test_patient_rollup_mixed_statuses(): + ru = rollup_patient([ + make_line(device="dexcom_g7", status="Clear to Ship"), + make_line(device="freestyle_libre_2", status="At Risk"), + ]) + assert ru.rollup_status == LineStatus.AT_RISK + + +def test_patient_rollup_plan_type_needed_capping(): + # PTN + Clear to Ship = Action Needed (capped, not PTN) + ru = rollup_patient([ + make_line(device="dexcom_g7", status="Clear to Ship"), + make_line(device="freestyle_libre_2", plan="unknown", + status="Plan Type Needed"), + ]) + assert ru.rollup_status == LineStatus.ACTION_NEEDED + assert ru.plan_type_needed is True + # PTN + At Risk = At Risk (the worse of the two) + ru2 = rollup_patient([ + make_line(device="dexcom_g7", status="At Risk"), + make_line(device="freestyle_libre_2", plan="unknown", + status="Plan Type Needed"), + ]) + assert ru2.rollup_status == LineStatus.AT_RISK + + +def test_patient_rollup_all_lines_plan_type_needed(): + ru = rollup_patient([ + make_line(device="dexcom_g7", plan="unknown", status="Plan Type Needed"), + make_line(device="freestyle_libre_2", plan="unknown", + status="Plan Type Needed"), + ]) + assert ru.rollup_status == LineStatus.PLAN_TYPE_NEEDED + + +def test_patient_rollup_non_cgm_excluded(): + ru = rollup_patient([ + make_line(device="dexcom_g7", status="Clear to Ship"), + make_line(device="omnipod_5", gradeable=False, status=None), + ]) + assert ru.rollup_status == LineStatus.CLEAR_TO_SHIP # pod changes nothing + assert ru.cgm_line_count == 1 + assert ru.total_line_count == 2 + + +def test_patient_rollup_no_gradeable_lines(): + ru = rollup_patient([make_line(device="omnipod_5", gradeable=False)]) + assert ru.rollup_status is None + assert ru.total_line_count == 1 + assert ru.cgm_line_count == 0 + + +def test_rollup_patients_groups_by_patient(): + lines = { + ("A", "dexcom_g7", "medicare"): make_line("A", status="Clear to Ship"), + ("A", "freestyle_libre_2", "medicare"): make_line( + "A", device="freestyle_libre_2", status="Action Needed"), + ("B", "dexcom_g7", "unknown"): make_line( + "B", plan="unknown", status="Plan Type Needed"), + } + rollups = rollup_patients(lines) + assert set(rollups) == {"A", "B"} + assert rollups["A"].rollup_status == LineStatus.ACTION_NEEDED + assert rollups["B"].rollup_status == LineStatus.PLAN_TYPE_NEEDED + + +# --- API level --------------------------------------------------------------- + +httpx = pytest.importorskip("httpx", reason="httpx required for TestClient") +from fastapi.testclient import TestClient # noqa: E402 + +import api.main as api_main # noqa: E402 + +client = TestClient(api_main.app) + +SHIP = (TODAY - timedelta(days=5)).isoformat() +VISIT = (TODAY - timedelta(days=30)).isoformat() +HEADER = ( + "patient_id,device_type,shipment_date,quantity,payer,plan_type," + "swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file" +) + + +def _upload(csv_text: str): + return client.post( + "/api/upload", + files={"file": ("fixture.csv", io.BytesIO(csv_text.encode()), "text/csv")}, + ) + + +def test_upload_response_contains_patients_nested(): + csv_text = ( + HEADER + + f"\nPT-N-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + + f"\nPT-N-1,Libre 2,{SHIP},1,Medicare Part B,medicare," + + "On File,,Yes,,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + body = resp.json() + patients = body["patients"] + assert len(patients) == 1 + p = patients[0] + assert p["patient_id"] == "PT-N-1" + assert p["rollup_status"] == "Action Needed" # worst line (Libre 2 no visit) + assert len(p["lines"]) == 2 + by_dev = {l["device_type"]: l for l in p["lines"]} + assert by_dev["dexcom_g7"]["line_status"] == "Clear to Ship" + assert by_dev["freestyle_libre_2"]["line_status"] == "Action Needed" + items = {i["doc_type"]: i for i in by_dev["freestyle_libre_2"]["doc_items"]} + assert items["visit"]["satisfied"] is False + assert items["visit"]["rule_id"] == "R101" + # Worst line renders first + assert p["lines"][0]["device_type"] == "freestyle_libre_2" + + +def test_patient_stats_count_patients_not_rows(): + csv_text = ( + HEADER + + f"\nPT-S-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + + f"\nPT-S-1,Libre 2,{SHIP},1,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + + f"\nPT-S-2,Dexcom G6,{SHIP},3,Medicaid - GA,," + + f"On File,{VISIT},,,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + stats = resp.json()["patient_stats"] + assert stats["patients_total"] == 2 # 2-device patient counts once + assert stats["Clear to Ship"] == 1 + assert stats["Plan Type Needed"] == 1 + + +def test_flat_records_untouched_by_nested_payload(): + csv_text = ( + HEADER + + f"\nPT-F-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + ) + resp = _upload(csv_text) + body = resp.json() + assert body["total"] == 1 + assert len(body["records"]) == 1 # flat path byte-identical (P6 switches UI) + assert body["records"][0]["readiness_status"] == "Clear to Ship" + + +def test_non_gradeable_only_patient_has_null_rollup(): + csv_text = ( + HEADER + + f"\nPT-P-1,Omnipod 5,{SHIP},1,Highmark BCBS,commercial," + + f"On File,{VISIT},Yes,Approved,Yes" + ) + resp = _upload(csv_text) + body = resp.json() + p = body["patients"][0] + assert p["rollup_status"] is None + assert p["cgm_line_count"] == 0 + assert p["lines"][0]["gradeable"] is False + assert p["lines"][0]["line_status"] is None + assert body["patient_stats"]["ungraded"] == 1 + + +def test_rollup_patient_empty_raises(): + with pytest.raises(ValueError): + rollup_patient([]) + + +def test_line_timing_flag_populated(): + csv_text = ( + HEADER + + f"\nPT-T-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + ) + resp = _upload(csv_text) + line = resp.json()["patients"][0]["lines"][0] + assert line["timing_flag"] == "RESUPPLY_READY"