a53728e424
Ends the override leak: an SWO/PA save now applies to one device's line, not
every device the patient has. doc_status gains device_type ('' = legacy
patient-scoped row, applies until re-saved); the old unique key is dropped by
introspected name and replaced with the 4-col scope key. Staff overrides now
FEED THE VERDICT (device-scoped, staff wins over the CSV cell) instead of
display-only; visit overrides keep their patient-scoped fan-out via
confirmed_visits. PUT /api/doc-status and /api/confirm-visit accept an
optional echo of the patient's lines and return {lines, rollup_status}
recomputed through the same engine (stateless, no batch rebuild). Frontend
minimal fix: doc-status state and writes keyed patient:device. Independent
audit: SHIP (migration reproduced empirically on both old-key shapes; guard
scoped per its nit). 162 tests green.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
"""
|
|
overrides.py
|
|
Signal — STTIL Solutions
|
|
|
|
Device-keyed staff overrides (build plan 02, P4; phase-2 design Item 4).
|
|
|
|
Pilot scope is LOCKED to two override kinds:
|
|
- visit: patient-scoped (the visit is the person). It rides the existing
|
|
confirmed_visits path, which already fans out to every line of the patient
|
|
by construction (evaluate_readiness receives the confirmed date per
|
|
patient). This module does not duplicate that mechanism.
|
|
- swo / pa: line-scoped. They apply only to lines whose device_type matches;
|
|
device_type '' marks a legacy patient-scoped row and matches all devices
|
|
until staff re-saves it device-scoped.
|
|
|
|
Overrides feed the GRADING inputs (staff knowledge wins over the CSV cell,
|
|
matching the existing display precedence), then affected lines re-grade.
|
|
|
|
PHI CONTRACT: overrides carry patient_id_hash only (the storage key). Lines
|
|
carry raw patient_id; matching hashes it. Nothing here logs patient data.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime
|
|
from typing import Optional
|
|
|
|
# Stored status -> display/grading label. Single source of truth; the API's
|
|
# display map imports this. The labels feed the readiness vocabularies
|
|
# ("On File" grades GOOD, "Denied" grades BAD, "Requested" grades PENDING).
|
|
STORED_STATUS_LABELS = {
|
|
"pending": "Pending",
|
|
"requested": "Requested",
|
|
"on_file": "On File",
|
|
"not_required": "Not Required",
|
|
"approved": "Approved",
|
|
"denied": "Denied",
|
|
}
|
|
|
|
# Legacy patient-scoped rows carry an empty device_type and match all devices.
|
|
LEGACY_ALL_DEVICES = ""
|
|
|
|
OVERRIDABLE_DOC_TYPES = ("swo", "pa")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OverrideKey:
|
|
org_id: str
|
|
patient_hash: str
|
|
device_type: str # '' = legacy patient-scoped, applies to all devices
|
|
doc_type: str # "swo" | "pa" (visit rides confirmed_visits)
|
|
|
|
|
|
@dataclass
|
|
class Override:
|
|
key: OverrideKey
|
|
status: str # stored value, e.g. "on_file" | "approved" | "denied"
|
|
status_date: Optional[date] = None
|
|
expiry_date: Optional[date] = None
|
|
confirmed_by: str = "staff"
|
|
confirmed_at: Optional[datetime] = None
|
|
|
|
|
|
def _hash_pid(patient_id: str) -> str:
|
|
return hashlib.sha256(patient_id.encode()).hexdigest()
|
|
|
|
|
|
def grading_value(override: Override) -> str:
|
|
"""The label fed to the readiness vocabularies for this override."""
|
|
return STORED_STATUS_LABELS.get(override.status, override.status)
|
|
|
|
|
|
def overrides_from_saved(saved: dict) -> list[Override]:
|
|
"""Build Override objects from the persistence load shape:
|
|
patient_hash -> device_type -> doc_type -> {status, status_date, ...}."""
|
|
out: list[Override] = []
|
|
for patient_hash, by_device in (saved or {}).items():
|
|
for device_type, by_doc in by_device.items():
|
|
for doc_type, row in by_doc.items():
|
|
if doc_type not in OVERRIDABLE_DOC_TYPES:
|
|
continue
|
|
out.append(
|
|
Override(
|
|
key=OverrideKey(
|
|
org_id="",
|
|
patient_hash=patient_hash,
|
|
device_type=device_type,
|
|
doc_type=doc_type,
|
|
),
|
|
status=row.get("status", ""),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def apply_overrides(lines: dict, overrides: list[Override]) -> set[str]:
|
|
"""Apply swo/pa overrides onto the grading inputs of matching lines.
|
|
|
|
Mutates each matching line's current_shipment csv fields (staff knowledge
|
|
wins over the CSV cell). Device-scoped rows win over legacy '' rows.
|
|
Returns the raw patient_ids of touched lines. Grading (or re-grading)
|
|
happens in the caller — build_readiness_index grades after application.
|
|
"""
|
|
if not overrides:
|
|
return set()
|
|
|
|
by_patient: dict[str, list[Override]] = {}
|
|
for o in overrides:
|
|
by_patient.setdefault(o.key.patient_hash, []).append(o)
|
|
|
|
touched: set[str] = set()
|
|
for line in lines.values():
|
|
patient_overrides = by_patient.get(_hash_pid(line.patient_id))
|
|
if not patient_overrides:
|
|
continue
|
|
# Legacy '' first so a device-scoped row applied after it wins.
|
|
for o in sorted(
|
|
patient_overrides, key=lambda o: o.key.device_type != LEGACY_ALL_DEVICES
|
|
):
|
|
if o.key.device_type not in (LEGACY_ALL_DEVICES, line.device_type):
|
|
continue
|
|
label = grading_value(o)
|
|
if o.key.doc_type == "swo":
|
|
line.current_shipment.csv_swo_status = label
|
|
elif o.key.doc_type == "pa":
|
|
line.current_shipment.csv_pa_status = label
|
|
touched.add(line.patient_id)
|
|
return touched
|