300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""Pure presentation logic of the dashboard (no Streamlit calls), unit-tested in tests/test_dashboard_helpers.py.
|
||
|
||
Clinical meaning stays on the backend: colours come from ADR-006 levels computed by the rules engine and the
|
||
discontinuation-risk index is computed server-side. Here we only order, label and reshape that data for display.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import random
|
||
import re
|
||
from collections.abc import Iterable, Sequence
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from functools import cache
|
||
from pathlib import Path
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import yaml
|
||
|
||
from app.api.schemas import AlertOut, DiscontinuationRisk, DoseOut, PatientListItem, SeriesPoint
|
||
|
||
TZ = ZoneInfo("Europe/Moscow")
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
# ADR-006 ordering for the doctor's attention: acute first, then high/moderate, then info-only, then quiet.
|
||
STATUS_ORDER = {"red": 0, "yellow": 1, "grey": 2, "green": 3}
|
||
|
||
MODULE_LABELS = {"veteran": "ветеран", "depression": "депрессия", "family": "семья"}
|
||
LEVEL_LABELS = {"acute": "Кризис", "high": "Высокий", "moderate": "Умеренный", "info": "Инфо"}
|
||
MONTHS_SHORT = ["янв", "фев", "мар", "апр", "мая", "июн", "июл", "авг", "сен", "окт", "ноя", "дек"]
|
||
|
||
DRUGS: dict[str, list[str]] = {
|
||
"сертралин": ["25 мг", "50 мг", "100 мг", "150 мг"],
|
||
"пароксетин": ["10 мг", "20 мг", "30 мг"],
|
||
"эсциталопрам": ["5 мг", "10 мг", "20 мг"],
|
||
"флуоксетин": ["20 мг", "40 мг"],
|
||
}
|
||
|
||
|
||
# ------------------------------------------------------------------ list: ordering, filtering, KPIs
|
||
def sort_patients(items: Iterable[PatientListItem]) -> list[PatientListItem]:
|
||
"""red → yellow → grey → green; inside a colour the higher discontinuation risk goes first."""
|
||
return sorted(items, key=lambda p: (STATUS_ORDER[p.status_color], -p.discontinuation_risk.score, p.display_name))
|
||
|
||
|
||
def filter_patients(
|
||
items: Iterable[PatientListItem], color: str | None = None, query: str = ""
|
||
) -> list[PatientListItem]:
|
||
q = query.strip().lower()
|
||
return [
|
||
p
|
||
for p in items
|
||
if (color is None or p.status_color == color)
|
||
and (not q or q in p.display_name.lower() or q in (p.drug or "").lower())
|
||
]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Kpis:
|
||
total: int
|
||
red: int
|
||
yellow: int
|
||
grey: int
|
||
green: int
|
||
high_risk: int
|
||
|
||
|
||
def kpis(items: Sequence[PatientListItem]) -> Kpis:
|
||
by = {c: sum(1 for p in items if p.status_color == c) for c in STATUS_ORDER}
|
||
high = sum(1 for p in items if p.discontinuation_risk.level == "high")
|
||
return Kpis(len(items), by["red"], by["yellow"], by["grey"], by["green"], high)
|
||
|
||
|
||
# ------------------------------------------------------------------ small visual mappings
|
||
@dataclass(frozen=True)
|
||
class View:
|
||
text: str
|
||
tone: str # key of theme.TONES
|
||
hint: str = ""
|
||
|
||
|
||
def trend_view(trend: str) -> View:
|
||
"""Direction of symptom scores; "up" means symptoms got worse (schemas.PatientListItem.trend)."""
|
||
return {
|
||
"up": View("↑", "red", "симптомы нарастают"),
|
||
"down": View("↓", "green", "симптомы снижаются"),
|
||
"flat": View("→", "grey", "без динамики"),
|
||
}.get(trend, View("·", "grey", "мало данных"))
|
||
|
||
|
||
def risk_view(risk: DiscontinuationRisk) -> View:
|
||
"""Badge for the rule-based discontinuation-risk index (score 0–100) with factors as a tooltip."""
|
||
label, tone = {"high": ("Высокий", "red"), "medium": ("Средний", "yellow"), "low": ("Низкий", "green")}[risk.level]
|
||
factors = "\n".join(f"• {f}" for f in risk.factors) or "• значимых факторов нет"
|
||
return View(f"{label} · {risk.score}", tone, f"Риск прерывания терапии: {risk.score}/100\n{factors}")
|
||
|
||
|
||
def adherence_view(adherence: float | None) -> View:
|
||
"""7-day adherence. ≥ 80 % is the usual "adherent" cut-off for antidepressants; < 60 % needs a call."""
|
||
if adherence is None:
|
||
return View("—", "grey")
|
||
pct = round(adherence * 100)
|
||
tone = "green" if adherence >= 0.8 else "yellow" if adherence >= 0.6 else "red"
|
||
return View(f"{pct}%", tone)
|
||
|
||
|
||
def status_view(color: str, top_level: str | None = None) -> View:
|
||
if color == "red":
|
||
return View("Срочно", "red", "Кризисное уведомление")
|
||
if color == "yellow":
|
||
return View("Внимание", "yellow", "врачу ≤ 24 ч или к визиту")
|
||
if color == "grey":
|
||
return View("Нет ответа", "grey", "молчание / только инфо-флаги")
|
||
return View("Без уведомлений", "green", "открытых сигналов нет")
|
||
|
||
|
||
def level_tone(level: str) -> str:
|
||
return {"acute": "red", "high": "yellow", "moderate": "yellow"}.get(level, "grey")
|
||
|
||
|
||
# ------------------------------------------------------------------ scales
|
||
@dataclass(frozen=True)
|
||
class Band:
|
||
lo: float
|
||
hi: float
|
||
code: str
|
||
label: str
|
||
|
||
|
||
SCALE_MAX = {"phq9": 27, "pcl5": 80}
|
||
|
||
|
||
@cache
|
||
def severity_bands(scale_id: str) -> tuple[Band, ...]:
|
||
"""Severity gradations from config/scales/<id>.yaml (ADR-008: thresholds live in config, not code)."""
|
||
path = REPO_ROOT / "config" / "scales" / f"{scale_id}.yaml"
|
||
try:
|
||
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
except OSError:
|
||
return ()
|
||
rows = data.get("severity", [])
|
||
return tuple(Band(float(b["min"]), float(b["max"]), str(b["code"]), str(b["label"])) for b in rows)
|
||
|
||
|
||
def severity_label(scale_id: str, value: float | None) -> str:
|
||
if value is None:
|
||
return ""
|
||
for b in severity_bands(scale_id):
|
||
if b.lo <= value <= b.hi:
|
||
return b.label
|
||
return ""
|
||
|
||
|
||
def metric_points(series: Iterable[SeriesPoint], metric: str) -> list[SeriesPoint]:
|
||
return sorted((p for p in series if p.metric == metric), key=lambda p: p.day)
|
||
|
||
|
||
def last_and_prev(series: Iterable[SeriesPoint], metric: str) -> tuple[float | None, float | None]:
|
||
pts = metric_points(series, metric)
|
||
last = pts[-1].value if pts else None
|
||
prev = pts[-2].value if len(pts) > 1 else None
|
||
return last, prev
|
||
|
||
|
||
# ------------------------------------------------------------------ dose calendar
|
||
@dataclass
|
||
class DoseGrid:
|
||
"""Heat-strip of doses: one column per calendar day, one row per scheduled time of day."""
|
||
|
||
days: list[date] = field(default_factory=list)
|
||
slots: list[str] = field(default_factory=list)
|
||
cells: dict[tuple[str, date], str] = field(default_factory=dict)
|
||
counts: dict[str, int] = field(default_factory=dict)
|
||
|
||
@property
|
||
def adherence(self) -> float | None:
|
||
"""Share of resolved doses taken (pending doses are not counted yet)."""
|
||
resolved = sum(self.counts.get(s, 0) for s in ("taken", "missed", "skipped"))
|
||
return self.counts.get("taken", 0) / resolved if resolved else None
|
||
|
||
|
||
def _local(dt: datetime) -> datetime:
|
||
return dt.astimezone(TZ) if dt.tzinfo else dt
|
||
|
||
|
||
def dose_grid(doses: Iterable[DoseOut]) -> DoseGrid:
|
||
grid = DoseGrid(counts={"taken": 0, "missed": 0, "skipped": 0, "pending": 0})
|
||
items = sorted(doses, key=lambda d: d.planned_at)
|
||
if not items:
|
||
return grid
|
||
days: set[date] = set()
|
||
slots: set[str] = set()
|
||
for d in items:
|
||
local = _local(d.planned_at)
|
||
day, slot = local.date(), local.strftime("%H:%M")
|
||
days.add(day)
|
||
slots.add(slot)
|
||
grid.cells[(slot, day)] = d.status
|
||
grid.counts[d.status] += 1
|
||
first, last = min(days), max(days)
|
||
grid.days = [date.fromordinal(o) for o in range(first.toordinal(), last.toordinal() + 1)]
|
||
grid.slots = sorted(slots)
|
||
return grid
|
||
|
||
|
||
# ------------------------------------------------------------------ alerts
|
||
def apply_local_acks(alerts: Iterable[AlertOut], acks: dict[int, tuple[str, datetime]]) -> list[AlertOut]:
|
||
"""Overlay acks done in this session (so the UI reflects the click even before the next poll)."""
|
||
out = []
|
||
for a in alerts:
|
||
if a.acked_at is None and a.id in acks:
|
||
by, at = acks[a.id]
|
||
a = a.model_copy(update={"acked_by": by, "acked_at": at})
|
||
out.append(a)
|
||
return out
|
||
|
||
|
||
def split_new(alerts: Iterable[AlertOut], seen: set[int]) -> list[AlertOut]:
|
||
return sorted((a for a in alerts if a.id not in seen), key=lambda a: a.id)
|
||
|
||
|
||
def sort_alerts(alerts: Iterable[AlertOut]) -> list[AlertOut]:
|
||
"""Open first (acute before others), then newest first."""
|
||
rank = {"acute": 0, "high": 1, "moderate": 2, "info": 3}
|
||
return sorted(alerts, key=lambda a: (a.acked_at is not None, rank.get(a.level, 9), -a.ts.timestamp()))
|
||
|
||
|
||
# ------------------------------------------------------------------ Russian formatting
|
||
def plural_ru(n: int, one: str, few: str, many: str) -> str:
|
||
n = abs(n)
|
||
if n % 10 == 1 and n % 100 != 11:
|
||
return one
|
||
if 2 <= n % 10 <= 4 and not 12 <= n % 100 <= 14:
|
||
return few
|
||
return many
|
||
|
||
|
||
def age_label(age: int) -> str:
|
||
if age <= 0:
|
||
return "возраст не указан"
|
||
return f"{age} {plural_ru(age, 'год', 'года', 'лет')}"
|
||
|
||
|
||
def short_date(d: date) -> str:
|
||
return f"{d.day} {MONTHS_SHORT[d.month - 1]}"
|
||
|
||
|
||
def when_label(dt: datetime, today: date | None = None) -> str:
|
||
local = _local(dt)
|
||
day = "сегодня" if today and local.date() == today else short_date(local.date())
|
||
return f"{day}, {local:%H:%M}"
|
||
|
||
|
||
def visit_label(visit: date | None, today: date | None) -> str:
|
||
if visit is None:
|
||
return "визит не назначен"
|
||
if today is None:
|
||
return f"визит {short_date(visit)}"
|
||
delta = (visit - today).days
|
||
if delta == 0:
|
||
return "визит сегодня"
|
||
if delta < 0:
|
||
return f"визит {short_date(visit)} (просрочен)"
|
||
return f"визит {short_date(visit)} · через {delta} {plural_ru(delta, 'день', 'дня', 'дней')}"
|
||
|
||
|
||
# ------------------------------------------------------------------ connect form
|
||
def absolute_link(link: str, public_url: str) -> str:
|
||
"""The API may return a relative web-console link (`/console?code=…`); a QR needs an absolute URL."""
|
||
if link.startswith(("http://", "https://", "tg://")):
|
||
return link
|
||
return public_url.rstrip("/") + "/" + link.lstrip("/")
|
||
|
||
|
||
def snils_checksum(digits: str) -> int:
|
||
"""Control number of a SNILS (Pension Fund algorithm): Σ digit·(9−i), then mod 101, 100/101 → 0."""
|
||
s = sum(int(ch) * (9 - i) for i, ch in enumerate(digits))
|
||
if s < 100:
|
||
return s
|
||
if s in (100, 101):
|
||
return 0
|
||
s %= 101
|
||
return 0 if s == 100 else s
|
||
|
||
|
||
def format_snils(digits: str, check: int) -> str:
|
||
return f"{digits[0:3]}-{digits[3:6]}-{digits[6:9]} {check:02d}"
|
||
|
||
|
||
def is_valid_snils(value: str) -> bool:
|
||
d = re.sub(r"\D", "", value)
|
||
return len(d) == 11 and snils_checksum(d[:9]) == int(d[9:])
|
||
|
||
|
||
def synthetic_snils(rng: random.Random | None = None) -> str:
|
||
"""Random checksum-valid SNILS for synthetic test patients. Starts with 000 so it never looks like a real one."""
|
||
r = rng or random.Random()
|
||
digits = "000" + "".join(str(r.randint(0, 9)) for _ in range(6))
|
||
return format_snils(digits, snils_checksum(digits))
|