""" dedup.py Signal — STTIL Solutions Groups normalized shipment rows into deduped shipments and gradeable coverage lines. This is the structural foundation of the readiness model: it turns a flat list of CSV-derived rows into the patient -> device -> shipment hierarchy the worklist needs, BEFORE any grading happens. Design: docs/dedup-design-2026-06-24.md (Pi, locked by Kisa 2026-06-24) docs/readiness-model-brief-2026-06-23.md sections 3, 5, 7 LOCKED DEDUP KEY: (patient_id, device_type, date_of_service) order_number is a displayed attribute, NEVER part of the dedup key. Monthly resupplies are separate dates of service, NOT duplicates. PLAN TYPE: carried through verbatim from the client-mapped field. It is NEVER derived from the payer name. None means "plan type needed" -> keyed as 'unknown' so the line still forms but cannot be graded green later. PHI CONTRACT: operates only on patient_id, device_type, shipment_date, quantity, payer (display), plan_type, component, order_number, hcpcs, and the csv_* doc-status fields. No names, SSNs, DOBs, or contact fields. """ from __future__ import annotations import hashlib import logging from collections import defaultdict from dataclasses import dataclass, field from datetime import date, datetime from typing import Optional from core.coverage_calculator import ShipmentRecord, CoverageFlag logger = logging.getLogger(__name__) def _hash_pid(patient_id: str) -> str: """Short hashed patient_id for log lines — raw patient_id is never logged.""" return hashlib.sha256(patient_id.encode()).hexdigest()[:12] def _as_date(value) -> date: """Coerce a datetime to a pure date so same-day timestamps group together.""" return value.date() if isinstance(value, datetime) else value # CGM device types graded this phase. Pumps and other DMEPOS categories are # preserved and displayed but not graded (gradeable=False). Mirrors the CGM # devices in payer_rules.json; moves into addressable config in a later step. CGM_DEVICE_TYPES = frozenset( {"dexcom_g6", "dexcom_g7", "freestyle_libre_2", "freestyle_libre_3"} ) UNKNOWN_PLAN_TYPE = "unknown" def is_cgm_device(device_type: str) -> bool: """True only for CGM device types graded in the current phase.""" return device_type in CGM_DEVICE_TYPES @dataclass class MergedShipment: """One deduped shipment: all rows sharing (patient_id, device_type, DOS).""" patient_id: str device_type: str shipment_date: date total_quantity: int components: list[str] # combined, deduped, first-seen order order_numbers: list[str] # collected, deduped — display only hcpcs_codes: list[str] # collected, deduped — display only payer: str # raw display NAME, latest non-null plan_type: Optional[str] # client-mapped; NEVER from payer name # csv_* doc fields — latest non-null wins across the group csv_visit_date: Optional[date] = None csv_swo_status: Optional[str] = None csv_pecos_verified: Optional[str] = None csv_pa_status: Optional[str] = None csv_diagnosis_on_file: Optional[str] = None csv_transfer_from: Optional[str] = None @dataclass(frozen=True) class CoverageLineKey: patient_id: str device_type: str plan_type_id: str @dataclass class CoverageLine: """A gradeable unit: one patient's one device under one plan type. line_status and timing_flag are computed later by the grading step (gradeLine); a freshly assigned line is intentionally ungraded. """ patient_id: str device_type: str plan_type: str gradeable: bool current_shipment: MergedShipment prior_shipments: list[MergedShipment] = field(default_factory=list) line_status: Optional[str] = None timing_flag: Optional[CoverageFlag] = None def _qty(record: ShipmentRecord) -> int: """A row without a quantity is still a shipment -> default to 1.""" return 1 if record.quantity is None else record.quantity def _merge_quantities(rows: list[ShipmentRecord]) -> int: """Sum quantities across DISTINCT component configs, max within a config. Within one component config, identical rows are likely data-entry duplicates, so take the MAX (guards against doubling). Across distinct configs (a real component split, e.g. sensor + transmitter), SUM. A disagreement within one config (e.g. qty 3 and qty 100 for the same component) collapses to the max and leaves a warning, so silently dropping quantity always leaves an audit breadcrumb. """ by_config: dict[frozenset, list[int]] = defaultdict(list) for row in rows: by_config[frozenset({row.component})].append(_qty(row)) total = 0 for config, qtys in by_config.items(): if len(set(qtys)) > 1: logger.warning( "Quantity disagreement in dedup group patient=%s device=%s " "component=%s qtys=%s -> taking max", _hash_pid(rows[0].patient_id), rows[0].device_type, set(config), qtys, ) total += max(qtys) return total def _merge_plan_type(rows: list[ShipmentRecord]) -> Optional[str]: """Plan type is grading-critical and must NEVER be guessed. If a dedup group carries more than one distinct non-null plan type, the data is ambiguous, so return None ('plan type needed') rather than pick one. """ distinct = {r.plan_type for r in rows if r.plan_type} if len(distinct) > 1: logger.warning( "Conflicting plan_type in dedup group patient=%s device=%s: %s " "-> routing to 'plan type needed'", _hash_pid(rows[0].patient_id), rows[0].device_type, sorted(distinct), ) return None return next(iter(distinct)) if distinct else None def _dedup_preserve_order(values: list[Optional[str]]) -> list[str]: """Collect non-empty values, dedup, preserve first-seen order.""" seen: dict[str, None] = {} for v in values: if v: seen.setdefault(v, None) return list(seen.keys()) def _latest_non_null(rows: list[ShipmentRecord], attr: str): """Last meaningful value for an attribute, in ingestion order. Empty / whitespace-only strings are treated as null so a blank later row never clobbers an earlier real value (does not rely on upstream nulling). """ result = None for row in rows: value = getattr(row, attr) if value is None: continue if isinstance(value, str) and value.strip() == "": continue result = value return result def dedup(records: list[ShipmentRecord]) -> list[MergedShipment]: """Group rows by (patient_id, device_type, shipment_date) into MergedShipments.""" groups: dict[tuple, list[ShipmentRecord]] = {} for record in records: key = (record.patient_id, record.device_type, _as_date(record.shipment_date)) groups.setdefault(key, []).append(record) merged: list[MergedShipment] = [] for (patient_id, device_type, shipment_date), rows in groups.items(): merged.append( MergedShipment( patient_id=patient_id, device_type=device_type, shipment_date=shipment_date, total_quantity=_merge_quantities(rows), components=_dedup_preserve_order([r.component for r in rows]), order_numbers=_dedup_preserve_order([r.order_number for r in rows]), hcpcs_codes=_dedup_preserve_order([r.hcpcs for r in rows]), payer=_latest_non_null(rows, "payer") or "", plan_type=_merge_plan_type(rows), csv_visit_date=_latest_non_null(rows, "csv_visit_date"), csv_swo_status=_latest_non_null(rows, "csv_swo_status"), csv_pecos_verified=_latest_non_null(rows, "csv_pecos_verified"), csv_pa_status=_latest_non_null(rows, "csv_pa_status"), csv_diagnosis_on_file=_latest_non_null(rows, "csv_diagnosis_on_file"), csv_transfer_from=_latest_non_null(rows, "csv_transfer_from"), ) ) merged.sort(key=lambda m: (m.patient_id, m.device_type, m.shipment_date)) return merged def assign_to_coverage_lines( merged_shipments: list[MergedShipment], ) -> dict[CoverageLineKey, CoverageLine]: """Group MergedShipments into CoverageLines by (patient_id, device_type, plan_type). plan_type is the client-mapped field. None is keyed as 'unknown' (plan type needed) so the line still forms but cannot be graded green later. The latest DOS in each line is the current shipment; earlier ones are collapsed history. """ by_key: dict[CoverageLineKey, list[MergedShipment]] = {} for shipment in merged_shipments: key = CoverageLineKey( patient_id=shipment.patient_id, device_type=shipment.device_type, plan_type_id=shipment.plan_type or UNKNOWN_PLAN_TYPE, ) by_key.setdefault(key, []).append(shipment) lines: dict[CoverageLineKey, CoverageLine] = {} for key, shipments in by_key.items(): shipments.sort(key=lambda s: s.shipment_date, reverse=True) lines[key] = CoverageLine( patient_id=key.patient_id, device_type=key.device_type, plan_type=key.plan_type_id, gradeable=is_cgm_device(key.device_type), current_shipment=shipments[0], prior_shipments=shipments[1:], line_status=None, timing_flag=None, ) return lines