PTSRaiAgent/dashboard/components/helpers.py

"""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))