bd92126384
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 <noreply@anthropic.com>
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""
|
|
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()
|
|
}
|