"""Align optional time-spent-with-patient to specialty duration rules. Paediatrics (26) has hard Schedule minima for special / extended / neurodevelopmental consults and developmental K-units. Family Practice (00) assessments are content-defined (not hard minute floors); time spent only steers BM25 toward the matching office ladder rung. When time spent is omitted, no duration filtering is applied. """ from __future__ import annotations import re from dataclasses import dataclass from .specialty_registry import normalize_specialty_code # Hard minimum direct-contact minutes from PAEDIATRICS (26) / GENERAL LISTINGS. _MIN_MINUTES_BY_CODE: dict[str, int] = { # Special paediatric consultation "A260": 75, "C260": 75, "W260": 75, # Extended special paediatric consultation "A662": 90, "C662": 90, "W662": 90, # Neurodevelopmental consultation "A667": 90, "A695": 90, "C667": 90, "C695": 90, "W667": 90, "W695": 90, # Developmental / behavioural care — first unit (≥ 20 min of a 30-min unit) "K122": 20, "K123": 20, } # Description cues for codes whose indexed text mentions a minute floor but # may not be in the allowlist above (future SoB additions). _DESC_MIN_PATTERNS: list[tuple[re.Pattern[str], int]] = [ ( re.compile( r"minimum\s+of\s+ninety\s+minutes|minimum\s+of\s+90\s+minutes|" r"spends?\s+a\s+minimum\s+of\s+90\s+minutes", re.I, ), 90, ), ( re.compile( r"minimum\s+of\s+seventy[- ]?five\s+minutes|" r"minimum\s+of\s+75\s+minutes|" r"spends?\s+a\s+minimum\s+of\s+75\s+minutes", re.I, ), 75, ), ( re.compile( r"minimum\s+of\s+fifty\s+minutes|minimum\s+of\s+50\s+minutes", re.I, ), 50, ), ( re.compile(r"\bper\s+unit\b", re.I), 20, # GP ½-hour unit services: first unit needs ≥ 20 minutes ), ] @dataclass(frozen=True) class DurationAlignment: minutes: int label: str keywords: str def align_time_spent( minutes: int | None, specialty_code: str | None = None, ) -> DurationAlignment | None: """Build steering keywords / label for clinician-reported time with patient.""" if minutes is None or minutes < 0: return None specialty = normalize_specialty_code(specialty_code) if specialty == "00": return _align_time_spent_family_practice(minutes) return _align_time_spent_paediatrics(minutes) def _align_time_spent_family_practice(minutes: int) -> DurationAlignment: """Soft ladder steer for FP — no invented hard minute floors.""" if minutes < 15: label = f"{minutes} minutes with patient (brief)" keywords = ( f"brief minor assessment A001 mini assessment A008 " f"Family Practice {minutes} minutes" ) elif minutes < 35: label = f"{minutes} minutes with patient" keywords = ( f"intermediate assessment A007 minor assessment A001 " f"Family Practice {minutes} minutes" ) else: label = f"{minutes} minutes with patient" keywords = ( f"general assessment A003 intermediate assessment A007 " f"periodic health K017 Family Practice {minutes} minutes" ) return DurationAlignment(minutes=minutes, label=label, keywords=keywords) def _align_time_spent_paediatrics(minutes: int) -> DurationAlignment: if minutes < 20: label = f"{minutes} minutes with patient (brief)" keywords = ( "brief level 1 paediatric assessment short visit " f"{minutes} minutes" ) elif minutes < 75: label = f"{minutes} minutes with patient" keywords = ( "level 1 level 2 paediatric assessment medical specific " f"consultation {minutes} minutes direct contact" ) elif minutes < 90: label = f"{minutes} minutes with patient (special consult eligible)" keywords = ( "special paediatric consultation minimum 75 minutes " f"level 2 medical specific {minutes} minutes direct contact" ) else: label = f"{minutes} minutes with patient (extended/neurodevelopmental eligible)" keywords = ( "extended special paediatric consultation neurodevelopmental " "consultation minimum 90 minutes developmental behavioural care " f"{minutes} minutes direct contact" ) return DurationAlignment(minutes=minutes, label=label, keywords=keywords) def code_min_minutes(doc: dict) -> int | None: """Return the Schedule minimum direct-contact minutes for this code, if any.""" code = (doc.get("billing_code") or "").upper() if code in _MIN_MINUTES_BY_CODE: return _MIN_MINUTES_BY_CODE[code] text = " ".join( [ doc.get("description_text") or "", doc.get("description") or "", doc.get("rules_and_constraints") or "", ] ) if not text.strip(): return None for pattern, mins in _DESC_MIN_PATTERNS: if pattern.search(text): return mins return None def code_applies_to_duration(doc: dict, minutes: int) -> bool: """True if this code is eligible given time spent with the patient. Codes without a Schedule minimum always apply. Minimum-time services are eligible only when ``minutes`` meets or exceeds their floor. """ required = code_min_minutes(doc) if required is None: return True return minutes >= required def filter_docs_for_duration(docs: list[dict], minutes: int) -> list[dict]: """Drop minimum-time codes the reported duration cannot support.""" return [d for d in docs if code_applies_to_duration(d, minutes)] def k_units_for_minutes(minutes: int) -> int: """How many ½-hour developmental/behavioural units the time supports. General Preamble table (30-minute units, major part thereof): 1→20, 2→46, 3→76, 4→106, 5→136, 6→166, 7→196, 8→226 minutes. """ thresholds = (20, 46, 76, 106, 136, 166, 196, 226) units = 0 for i, need in enumerate(thresholds, start=1): if minutes >= need: units = i else: break return units def duration_hint_for_llm( align: DurationAlignment, specialty_code: str | None = None, ) -> str: """Clinician-facing / LLM hint describing duration eligibility.""" specialty = normalize_specialty_code(specialty_code) if specialty == "00": return _duration_hint_family_practice(align) return _duration_hint_paediatrics(align) def _duration_hint_family_practice(align: DurationAlignment) -> str: mins = align.minutes lines = [ f"{align.label}.", "Family Practice office assessments are content-defined (not hard " "minute floors). Prefer the ladder rung that matches the documented " "service: mini A008 → minor A001 → intermediate A007 → general A003 " "→ re-assessment A004 / periodic health K017.", ] if mins < 15: lines.append( "Brief contact — prefer minor A001 or mini A008 unless the note " "documents a fuller intermediate or general assessment." ) elif mins < 35: lines.append( "Prefer intermediate A007 or minor A001; do not recommend general " "assessment A003 / periodic K017 unless the note documents those " "elements." ) else: lines.append( "Longer visit — general assessment A003 or periodic health K017 " "may apply when the note documents those elements." ) return " ".join(lines) def _duration_hint_paediatrics(align: DurationAlignment) -> str: mins = align.minutes lines = [ f"{align.label}.", "Only recommend Paediatrics codes whose Schedule minimum direct-contact " "time is met (or that have no minimum-time requirement).", "Hard floors from the current Schedule:", " • Special paediatric consultation (A260/C260/W260): ≥ 75 minutes", " • Extended special paediatric consultation (A662/C662/W662): ≥ 90 minutes", " • Neurodevelopmental consultation (A667/A695 and setting mirrors): ≥ 90 minutes", " • Developmental/behavioural care (K122/K123): ≥ 20 minutes for 1 unit", ] units = k_units_for_minutes(mins) if units: lines.append( f"Reported time supports up to {units} K122/K123 unit(s) " "(½ hour or major part thereof)." ) else: lines.append( "Reported time is below 20 minutes — do not recommend K122/K123 " "unit services." ) if mins < 75: lines.append( "Do not recommend special / extended / neurodevelopmental " "consultations — time spent is below their Schedule minima." ) elif mins < 90: lines.append( "Special paediatric consultation (≥ 75 min) may apply; extended " "special and neurodevelopmental consultations still require ≥ 90 min." ) else: lines.append( "Time spent meets minima for special, extended special, and " "neurodevelopmental consultations when other elements are met." ) if mins < 20: lines.append( "Prefer Level 1 paediatric assessment (A261) for brief encounters " "unless the note clearly documents a more extensive Level 2 exam." ) return " ".join(lines)