feat(core): rule engine for alerts/agent actions and virtual demo clock
CI / ruff · pytest (pull_request) Successful in 18s
CI / ruff · pytest (pull_request) Successful in 18s
- app/core/clock.py: now()/today(), virtual offset (advance/set_now/reset), pluggable offset persistence hook (ADR-005) - config/rules.yaml + app/core/rules.py: YAML declares rules (params, actions, levels, reason_ru), Python evaluators implement predicates by id; dedup of open alerts, crisis never suppressed; ADR-006 colours (ADR-007) - tests: per-rule unit tests, dedup, clock, 13-day demo persona trajectory Refs #5 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,122 @@
|
||||
"""The only source of time in the system (ADR-005, guardrail #4).
|
||||
|
||||
Real mode (`DEMO_CLOCK=false`): `now()` is the wall clock in `settings.tz`.
|
||||
Virtual mode (`DEMO_CLOCK=true`): `now()` is the wall clock shifted by an offset. The demo console moves
|
||||
time forward with `advance(days=1)` ("next day"), so 13 days of therapy fit into a 3-minute demo while the
|
||||
scheduler and the rule windows ("3 days in a row", "7 days") keep working on the shifted time.
|
||||
|
||||
The clock keeps an *offset*, not a frozen instant: virtual time keeps ticking between button presses, so
|
||||
scheduled jobs and timestamps remain ordered. The offset can be persisted through a pluggable hook
|
||||
(`register_persistence`) so the DB layer can store it and survive restarts; by default it lives in memory.
|
||||
|
||||
Business logic must never call `datetime.now()` directly — it receives time from here (or as a parameter).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from datetime import date, datetime, timedelta, timezone, tzinfo
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
LoadOffset = Callable[[], timedelta | None]
|
||||
SaveOffset = Callable[[timedelta], None]
|
||||
|
||||
_lock = threading.RLock()
|
||||
_offset = timedelta(0)
|
||||
_save: SaveOffset | None = None
|
||||
|
||||
# Moscow has had a fixed UTC+3 without DST since 2014; used only if the OS has no tz database
|
||||
# (slim Docker images may ship without tzdata).
|
||||
_FALLBACK_OFFSETS = {"Europe/Moscow": timedelta(hours=3)}
|
||||
|
||||
|
||||
def _tz() -> tzinfo:
|
||||
name = get_settings().tz
|
||||
try:
|
||||
return ZoneInfo(name)
|
||||
except ZoneInfoNotFoundError:
|
||||
return timezone(_FALLBACK_OFFSETS.get(name, timedelta(0)), name)
|
||||
|
||||
|
||||
def _virtual() -> bool:
|
||||
return get_settings().demo_clock
|
||||
|
||||
|
||||
def _real_now() -> datetime:
|
||||
# The single sanctioned wall-clock read in the codebase.
|
||||
return datetime.now(_tz())
|
||||
|
||||
|
||||
def _require_virtual() -> None:
|
||||
if not _virtual():
|
||||
raise RuntimeError("Virtual clock is disabled (DEMO_CLOCK=false); time cannot be moved")
|
||||
|
||||
|
||||
def _store(offset: timedelta) -> None:
|
||||
global _offset
|
||||
_offset = offset
|
||||
if _save is not None:
|
||||
_save(offset)
|
||||
|
||||
|
||||
def now() -> datetime:
|
||||
"""Current time, tz-aware in `settings.tz`; shifted by the demo offset in virtual mode."""
|
||||
real = _real_now()
|
||||
if not _virtual():
|
||||
return real
|
||||
with _lock:
|
||||
return real + _offset
|
||||
|
||||
|
||||
def today() -> date:
|
||||
"""Current local date according to `now()`."""
|
||||
return now().date()
|
||||
|
||||
|
||||
def offset() -> timedelta:
|
||||
"""Current virtual offset (always zero in real mode)."""
|
||||
if not _virtual():
|
||||
return timedelta(0)
|
||||
with _lock:
|
||||
return _offset
|
||||
|
||||
|
||||
def advance(days: float = 0, hours: float = 0, minutes: float = 0) -> datetime:
|
||||
"""Move virtual time forward (or backward with negative values). Returns the new `now()`."""
|
||||
_require_virtual()
|
||||
with _lock:
|
||||
_store(_offset + timedelta(days=days, hours=hours, minutes=minutes))
|
||||
return now()
|
||||
|
||||
|
||||
def set_now(dt: datetime) -> datetime:
|
||||
"""Jump virtual time to `dt` (naive values are interpreted in `settings.tz`). Returns the new `now()`."""
|
||||
_require_virtual()
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=_tz())
|
||||
with _lock:
|
||||
_store(dt - _real_now())
|
||||
return now()
|
||||
|
||||
|
||||
def reset() -> None:
|
||||
"""Return virtual time to the wall clock (offset = 0)."""
|
||||
with _lock:
|
||||
_store(timedelta(0))
|
||||
|
||||
|
||||
def register_persistence(load: LoadOffset | None, save: SaveOffset | None) -> None:
|
||||
"""Plug in offset storage (e.g. a DB row). `load` is called once immediately; `save` on every change.
|
||||
|
||||
Pass `(None, None)` to go back to in-memory only.
|
||||
"""
|
||||
global _offset, _save
|
||||
with _lock:
|
||||
_save = save
|
||||
if load is not None:
|
||||
stored = load()
|
||||
if stored is not None:
|
||||
_offset = stored
|
||||
@@ -0,0 +1,481 @@
|
||||
"""Deterministic alert & agent-action engine (ADR-007, guardrail #3).
|
||||
|
||||
The LLM never decides whether to alert. Structured data (daily check-ins, doses, scale totals, C-SSRS) is
|
||||
packed into a `PatientSnapshot`; `evaluate()` runs the rules and returns `Action`s that the orchestrator
|
||||
executes (create an alert, start the crisis script, schedule a scale, send medication info...). The LLM only
|
||||
phrases the reply around an action that has already been chosen.
|
||||
|
||||
How a rule is defined — the YAML/code split:
|
||||
* `config/rules.yaml` declares each rule: `id`, description, thresholds/windows (`params`), the actions it
|
||||
produces (type, level, extra params) and `reason_ru` for the doctor. Clinicians edit only this file.
|
||||
* This module implements the predicate of each rule as a pure function registered under the same id with
|
||||
`@evaluator("<id>")`. It returns `None` when the rule does not fire, or a dict of computed values
|
||||
(e.g. `{"total": 22}`) that is merged into action params and used to format `reason_ru`.
|
||||
* Loading validates both directions: a YAML id without an evaluator is an error (no silently dead rules).
|
||||
|
||||
Determinism: evaluators read time only from `snapshot.as_of` (the virtual "now" from `app.core.clock`),
|
||||
never from the wall clock, so the same snapshot always yields the same actions.
|
||||
|
||||
Dedup: when a rule would emit an `alert` whose code is already open (not acked by the doctor) the whole rule
|
||||
stays silent. `crisis` actions are never suppressed — the orchestrator is responsible for not restarting an
|
||||
already running crisis script.
|
||||
|
||||
Colours (ADR-006): ACUTE → red; HIGH/MODERATE → yellow; INFO only → grey; no open alerts → green.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Iterable
|
||||
from datetime import date, datetime, timedelta
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
import yaml
|
||||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||||
|
||||
from app.config import get_settings
|
||||
from app.core.types import AlertLevel
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- snapshot
|
||||
|
||||
Module = Literal["veteran", "depression", "family"]
|
||||
SleepBucket = Literal["<4", "4-6", "6-8", ">8"]
|
||||
AlcoholBucket = Literal["none", "1-2", "more"]
|
||||
DoseStatus = Literal["taken", "missed", "skipped", "pending"]
|
||||
CssrsRisk = Literal["none", "low", "moderate", "high"]
|
||||
ActionType = Literal[
|
||||
"alert",
|
||||
"crisis",
|
||||
"schedule_scale",
|
||||
"reach_out",
|
||||
"med_info",
|
||||
"suggest_visit",
|
||||
"show_safety_plan",
|
||||
"enhanced_monitoring",
|
||||
]
|
||||
StatusColor = Literal["red", "yellow", "green", "grey"]
|
||||
|
||||
_CSSRS_RANK: dict[str, int] = {"none": 0, "low": 1, "moderate": 2, "high": 3}
|
||||
_LEVELS: dict[str, AlertLevel] = {
|
||||
"info": AlertLevel.INFO,
|
||||
"moderate": AlertLevel.MODERATE,
|
||||
"high": AlertLevel.HIGH,
|
||||
"acute": AlertLevel.ACUTE,
|
||||
}
|
||||
|
||||
|
||||
class DailyCheckin(BaseModel):
|
||||
"""One day of the daily check-in (ADR-002). `None` = not asked/not answered that day."""
|
||||
|
||||
day: date
|
||||
mood: int | None = Field(default=None, ge=0, le=10)
|
||||
med_taken: bool | None = None
|
||||
sleep_hours_bucket: SleepBucket | None = None
|
||||
alcohol: AlcoholBucket | None = None
|
||||
side_effects: list[str] = Field(default_factory=list) # normalised codes from `extract`, e.g. "nausea"
|
||||
quit_intent: bool = False # "I'll stop taking the pills" detected in free text
|
||||
nightmares: bool | None = None
|
||||
|
||||
|
||||
class ScalePoint(BaseModel):
|
||||
"""A completed scale: only the total (+ PHQ-9 item 9) is needed by the rules."""
|
||||
|
||||
scale: str # phq9 | pcl5 | gad7 | auditc | ...
|
||||
ts: datetime
|
||||
total: int = Field(ge=0)
|
||||
item9: int | None = Field(default=None, ge=0, le=3)
|
||||
scheduled: bool = True # False = unscheduled (triggered by a rule)
|
||||
|
||||
|
||||
class DoseRecord(BaseModel):
|
||||
planned_at: datetime
|
||||
status: DoseStatus
|
||||
|
||||
|
||||
class PatientSnapshot(BaseModel):
|
||||
"""Everything the rules may look at, assembled by the caller from the clinical DB (pseudonymised)."""
|
||||
|
||||
token: str
|
||||
as_of: datetime # virtual "now" (app.core.clock.now()); all windows are counted back from here
|
||||
age: int = Field(ge=0)
|
||||
sex: Literal["m", "f"]
|
||||
modules: set[Module] = Field(default_factory=set)
|
||||
therapy_day: int = 0 # day 0 = visit where therapy was prescribed
|
||||
checkins: list[DailyCheckin] = Field(default_factory=list)
|
||||
scale_results: list[ScalePoint] = Field(default_factory=list)
|
||||
doses: list[DoseRecord] = Field(default_factory=list)
|
||||
cssrs_risk: CssrsRisk | None = None
|
||||
weapon_access: bool | None = None
|
||||
last_patient_message_at: datetime | None = None
|
||||
open_alert_codes: set[str] = Field(default_factory=set) # alerts not yet acked by the doctor
|
||||
|
||||
@field_validator("as_of")
|
||||
@classmethod
|
||||
def _aware(cls, v: datetime) -> datetime:
|
||||
if v.tzinfo is None:
|
||||
raise ValueError("as_of must be timezone-aware (take it from app.core.clock.now())")
|
||||
return v
|
||||
|
||||
|
||||
class Action(BaseModel):
|
||||
"""One output of the rule engine, executed by the orchestrator."""
|
||||
|
||||
type: ActionType
|
||||
level: AlertLevel | None = None
|
||||
code: str # stable reason code = rule id (dedup key, stored with the alert)
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
reason_ru: str = "" # short text for the doctor dashboard
|
||||
|
||||
@property
|
||||
def creates_alert(self) -> bool:
|
||||
"""`alert` and `crisis` actions are persisted as an alert record (crisis = acute alert)."""
|
||||
return self.type in ("alert", "crisis")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- config
|
||||
|
||||
|
||||
class ActionSpec(BaseModel):
|
||||
type: ActionType
|
||||
level: Literal["info", "moderate", "high", "acute"] | None = None
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _alert_needs_level(self) -> ActionSpec:
|
||||
if self.type in ("alert", "crisis") and self.level is None:
|
||||
raise ValueError(f"action '{self.type}' requires a level")
|
||||
return self
|
||||
|
||||
|
||||
class RuleSpec(BaseModel):
|
||||
id: str
|
||||
description: str = ""
|
||||
enabled: bool = True
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
actions: list[ActionSpec] = Field(min_length=1)
|
||||
reason_ru: str = ""
|
||||
|
||||
|
||||
class RulesConfig(BaseModel):
|
||||
rules: list[RuleSpec]
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_ids(self) -> RulesConfig:
|
||||
ids = [r.id for r in self.rules]
|
||||
dupes = {i for i in ids if ids.count(i) > 1}
|
||||
if dupes:
|
||||
raise ValueError(f"duplicate rule ids: {sorted(dupes)}")
|
||||
unknown = set(ids) - set(_REGISTRY)
|
||||
if unknown:
|
||||
raise ValueError(f"rules without an evaluator in app/core/rules.py: {sorted(unknown)}")
|
||||
return self
|
||||
|
||||
|
||||
def load_rules(path: Path | None = None) -> RulesConfig:
|
||||
"""Parse and validate a rules YAML (default: `<config_dir>/rules.yaml`, cached)."""
|
||||
if path is None:
|
||||
return _default_rules()
|
||||
return RulesConfig.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
|
||||
|
||||
|
||||
@lru_cache
|
||||
def _default_rules() -> RulesConfig:
|
||||
return load_rules(get_settings().config_dir / "rules.yaml")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- registry
|
||||
|
||||
Context = dict[str, Any]
|
||||
Evaluator = Callable[[PatientSnapshot, dict[str, Any]], Context | None]
|
||||
_REGISTRY: dict[str, Evaluator] = {}
|
||||
|
||||
|
||||
def evaluator(rule_id: str) -> Callable[[Evaluator], Evaluator]:
|
||||
"""Register the predicate implementing YAML rule `rule_id`."""
|
||||
|
||||
def deco(fn: Evaluator) -> Evaluator:
|
||||
if rule_id in _REGISTRY:
|
||||
raise ValueError(f"evaluator for '{rule_id}' already registered")
|
||||
_REGISTRY[rule_id] = fn
|
||||
return fn
|
||||
|
||||
return deco
|
||||
|
||||
|
||||
def registered_rule_ids() -> set[str]:
|
||||
return set(_REGISTRY)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- helpers
|
||||
|
||||
|
||||
def _today(s: PatientSnapshot) -> date:
|
||||
return s.as_of.date()
|
||||
|
||||
|
||||
def _checkins_by_day(s: PatientSnapshot) -> dict[date, DailyCheckin]:
|
||||
return {c.day: c for c in s.checkins if c.day <= _today(s)}
|
||||
|
||||
|
||||
def _anchor_day(s: PatientSnapshot) -> date:
|
||||
"""Last day a "N days in a row" streak may end on.
|
||||
|
||||
Today if today's check-in exists, otherwise yesterday — so a morning evaluation (before today's check-in)
|
||||
still sees yesterday's streak, but an old streak does not keep firing forever.
|
||||
"""
|
||||
today = _today(s)
|
||||
return today if today in _checkins_by_day(s) else today - timedelta(days=1)
|
||||
|
||||
|
||||
def _streak(s: PatientSnapshot, days: int, pred: Callable[[DailyCheckin], bool]) -> list[DailyCheckin] | None:
|
||||
"""Check-ins of `days` consecutive calendar days ending at the anchor day, if all satisfy `pred`.
|
||||
|
||||
A missing day breaks the streak: we never assume a symptom on a day the patient did not report.
|
||||
"""
|
||||
by_day = _checkins_by_day(s)
|
||||
end = _anchor_day(s)
|
||||
run: list[DailyCheckin] = []
|
||||
for i in range(days):
|
||||
c = by_day.get(end - timedelta(days=i))
|
||||
if c is None or not pred(c):
|
||||
return None
|
||||
run.append(c)
|
||||
return run
|
||||
|
||||
|
||||
def _history(s: PatientSnapshot, scale: str) -> list[ScalePoint]:
|
||||
"""Results of `scale` up to `as_of`, oldest first (results "from the future" are ignored)."""
|
||||
return sorted((p for p in s.scale_results if p.scale == scale and p.ts <= s.as_of), key=lambda p: p.ts)
|
||||
|
||||
|
||||
def _recent_item9(s: PatientSnapshot, threshold: int, recent_hours: float) -> int | None:
|
||||
"""Max PHQ-9 item 9 among PHQ-9s completed within `recent_hours`, if it reaches `threshold`.
|
||||
|
||||
Limited to recent results so one old positive answer does not restart the crisis protocol on every
|
||||
evaluation; after that the open acute alert keeps the patient red until the doctor acts.
|
||||
"""
|
||||
since = s.as_of - timedelta(hours=recent_hours)
|
||||
values = [p.item9 for p in _history(s, "phq9") if p.ts > since and p.item9 is not None]
|
||||
top = max(values, default=None)
|
||||
return top if top is not None and top >= threshold else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- evaluators
|
||||
# Each function implements the predicate of the YAML rule with the same id; thresholds come from `p`.
|
||||
|
||||
|
||||
@evaluator("phq9_item9")
|
||||
def _phq9_item9(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
# Item 9 is collected only by buttons (never LLM-interpreted), so the value is the patient's own answer.
|
||||
item9 = _recent_item9(s, p["threshold"], p["recent_hours"])
|
||||
return None if item9 is None else {"item9": item9}
|
||||
|
||||
|
||||
@evaluator("cssrs_high")
|
||||
def _cssrs_high(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
return {} if s.cssrs_risk == "high" else None
|
||||
|
||||
|
||||
@evaluator("weapon_access_risk")
|
||||
def _weapon_access_risk(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
# Access to means alone is not a crisis; combined with any suicide-risk signal it is.
|
||||
if not s.weapon_access:
|
||||
return None
|
||||
item9_signal = _recent_item9(s, p["item9_threshold"], p["recent_hours"]) is not None
|
||||
cssrs_signal = _CSSRS_RANK.get(s.cssrs_risk or "none", 0) >= _CSSRS_RANK[p["cssrs_min"]]
|
||||
return {} if item9_signal or cssrs_signal else None
|
||||
|
||||
|
||||
@evaluator("cssrs_moderate")
|
||||
def _cssrs_moderate(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
return {} if s.cssrs_risk == "moderate" else None
|
||||
|
||||
|
||||
def _latest_at_least(scale: str) -> Evaluator:
|
||||
"""Latest total of `scale` >= params.threshold."""
|
||||
|
||||
def fn(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
hist = _history(s, scale)
|
||||
if hist and hist[-1].total >= p["threshold"]:
|
||||
return {"total": hist[-1].total}
|
||||
return None
|
||||
|
||||
return fn
|
||||
|
||||
|
||||
def _rise_between_last_two(scale: str) -> Evaluator:
|
||||
"""Latest total minus the previous total of `scale` >= params.min_rise (worsening)."""
|
||||
|
||||
def fn(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
hist = _history(s, scale)
|
||||
if len(hist) < 2:
|
||||
return None
|
||||
prev, last = hist[-2].total, hist[-1].total
|
||||
if last - prev >= p["min_rise"]:
|
||||
return {"previous": prev, "total": last, "rise": last - prev}
|
||||
return None
|
||||
|
||||
return fn
|
||||
|
||||
|
||||
evaluator("phq9_severe")(_latest_at_least("phq9"))
|
||||
evaluator("pcl5_severe")(_latest_at_least("pcl5"))
|
||||
evaluator("phq9_rise")(_rise_between_last_two("phq9"))
|
||||
evaluator("pcl5_rise")(_rise_between_last_two("pcl5"))
|
||||
|
||||
|
||||
@evaluator("auditc_positive")
|
||||
def _auditc_positive(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
# Sex-specific cut-off: >= 4 for men, >= 3 for women (lower tolerance / higher risk at equal intake).
|
||||
hist = _history(s, "auditc")
|
||||
threshold = p["threshold_m"] if s.sex == "m" else p["threshold_f"]
|
||||
if hist and hist[-1].total >= threshold:
|
||||
return {"total": hist[-1].total, "threshold": threshold}
|
||||
return None
|
||||
|
||||
|
||||
@evaluator("quit_intent")
|
||||
def _quit_intent(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
# Fires on the day the intent is voiced, and on a later skipped dose within the window
|
||||
# (the intent turned into action).
|
||||
anchor = _anchor_day(s)
|
||||
today = _checkins_by_day(s).get(anchor)
|
||||
if today is None:
|
||||
return None
|
||||
if today.quit_intent:
|
||||
return {}
|
||||
since = anchor - timedelta(days=p["window_days"])
|
||||
intent_before = any(c.quit_intent for c in s.checkins if since <= c.day < anchor)
|
||||
return {} if today.med_taken is False and intent_before else None
|
||||
|
||||
|
||||
@evaluator("missed_doses_7d")
|
||||
def _missed_doses(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
# Counts distinct calendar days (not doses) within the window with a missed/skipped dose,
|
||||
# from dose records and from the check-in answer "did not take" — whichever source has the data.
|
||||
window = p["window_days"]
|
||||
first = _today(s) - timedelta(days=window - 1)
|
||||
tz = s.as_of.tzinfo
|
||||
days = {
|
||||
d.planned_at.astimezone(tz).date()
|
||||
for d in s.doses
|
||||
if d.status in ("missed", "skipped") and d.planned_at <= s.as_of
|
||||
}
|
||||
days |= {c.day for c in s.checkins if c.med_taken is False}
|
||||
count = sum(1 for d in days if first <= d <= _today(s))
|
||||
return {"count": count} if count >= p["threshold"] else None
|
||||
|
||||
|
||||
@evaluator("side_effect_3d")
|
||||
def _side_effect_3d(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
run = _streak(s, p["consecutive_days"], lambda c: bool(c.side_effects))
|
||||
if run is None:
|
||||
return None
|
||||
effects = sorted({e for c in run for e in c.side_effects})
|
||||
return {"effects": ", ".join(effects)}
|
||||
|
||||
|
||||
@evaluator("sleep_lt4_3n")
|
||||
def _sleep_lt4(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
return None if _streak(s, p["consecutive_days"], lambda c: c.sleep_hours_bucket == "<4") is None else {}
|
||||
|
||||
|
||||
@evaluator("alcohol_3d")
|
||||
def _alcohol_3d(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
drank = _streak(s, p["consecutive_days"], lambda c: c.alcohol is not None and c.alcohol != "none")
|
||||
return None if drank is None else {}
|
||||
|
||||
|
||||
@evaluator("mood_low_3d")
|
||||
def _mood_low_3d(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
low = _streak(s, p["consecutive_days"], lambda c: c.mood is not None and c.mood <= p["threshold"])
|
||||
if low is None:
|
||||
return None
|
||||
# Do not ask for an unscheduled PHQ-9 again if one was already taken recently.
|
||||
since = s.as_of - timedelta(days=p["cooldown_days"])
|
||||
if any(not r.scheduled and r.ts > since for r in _history(s, "phq9")):
|
||||
return None
|
||||
return {}
|
||||
|
||||
|
||||
@evaluator("no_contact_3d")
|
||||
def _no_contact(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
if s.last_patient_message_at is None:
|
||||
return None
|
||||
silence = s.as_of - s.last_patient_message_at
|
||||
return {"silent_days": silence.days} if silence > timedelta(days=p["days"]) else None
|
||||
|
||||
|
||||
@evaluator("young_adult_start")
|
||||
def _young_adult_start(s: PatientSnapshot, p: dict[str, Any]) -> Context | None:
|
||||
if s.age < p["age_below"] and 0 <= s.therapy_day <= p["max_therapy_day"]:
|
||||
return {}
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- engine
|
||||
|
||||
|
||||
class _KeepMissing(dict[str, Any]):
|
||||
def __missing__(self, key: str) -> str:
|
||||
return "{" + key + "}"
|
||||
|
||||
|
||||
def _build_actions(rule: RuleSpec, ctx: Context) -> list[Action]:
|
||||
reason = rule.reason_ru.format_map(_KeepMissing({**rule.params, **ctx}))
|
||||
return [
|
||||
Action(
|
||||
type=spec.type,
|
||||
level=_LEVELS[spec.level] if spec.level else None,
|
||||
code=rule.id,
|
||||
params={**ctx, **spec.params},
|
||||
reason_ru=reason,
|
||||
)
|
||||
for spec in rule.actions
|
||||
]
|
||||
|
||||
|
||||
def evaluate(snapshot: PatientSnapshot, config: RulesConfig | None = None) -> list[Action]:
|
||||
"""Run all enabled rules (YAML order) on the snapshot. Pure and deterministic."""
|
||||
cfg = config or load_rules()
|
||||
out: list[Action] = []
|
||||
for rule in cfg.rules:
|
||||
if not rule.enabled:
|
||||
continue
|
||||
ctx = _REGISTRY[rule.id](snapshot, rule.params)
|
||||
if ctx is None:
|
||||
continue
|
||||
actions = _build_actions(rule, ctx)
|
||||
is_crisis = any(a.type == "crisis" for a in actions)
|
||||
already_open = any(a.type == "alert" and a.code in snapshot.open_alert_codes for a in actions)
|
||||
if already_open and not is_crisis:
|
||||
continue
|
||||
out.extend(actions)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------- colours
|
||||
|
||||
|
||||
class PatientStatus(BaseModel):
|
||||
color: StatusColor
|
||||
top_level: AlertLevel | None
|
||||
|
||||
|
||||
def color(levels: Iterable[AlertLevel | None]) -> StatusColor:
|
||||
"""Dashboard colour for a set of open alert levels (ADR-006)."""
|
||||
present = {lvl for lvl in levels if lvl is not None}
|
||||
if AlertLevel.ACUTE in present:
|
||||
return "red"
|
||||
if present & {AlertLevel.HIGH, AlertLevel.MODERATE}:
|
||||
return "yellow"
|
||||
if AlertLevel.INFO in present:
|
||||
return "grey"
|
||||
return "green"
|
||||
|
||||
|
||||
def patient_status(open_alert_levels: Iterable[AlertLevel | None]) -> PatientStatus:
|
||||
"""Colour + highest open level for a patient row on the dashboard."""
|
||||
levels = [lvl for lvl in open_alert_levels if lvl is not None]
|
||||
return PatientStatus(color=color(levels), top_level=max(levels, default=None))
|
||||
Reference in New Issue
Block a user