PTSRaiAgent/app/services/garden.py

"""The patient's garden (#118, slide 13): light gamification over all modules.

Deterministic, no LLM (guardrail 3), rules in `config/garden.yaml` (ADR-008). Computed on the fly from existing
activity rows — no own table, no migration (ADR-003), and changing the rules replays the whole history.

«Without pressure»: points only add up (a skipped day takes nothing away), growth counts the fact of an activity
and never its content (scale scores, mood), and there is no comparison between patients. Only the patient sees it.
An honest «Пропущу» on a dose grows the garden like «Принял(а)», so the garden never rewards a false «taken» (#120).
After a recent acute/high alert the garden is `quiet`: growth stays, celebration does not.
"""

from collections import defaultdict
from collections.abc import Iterable
from datetime import date, datetime, timedelta
from functools import lru_cache
from pathlib import Path
from typing import Literal

import yaml
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import clock
from app.core.intake_questions import DIARY_KINDS
from app.db.clinical import Alert, DailyCheckin, Dose, ScaleResult
from app.db.intake import QuestionnaireRecord

Activity = Literal["meds", "diary", "questionnaire"]


class ActivityRule(BaseModel):
    points: int
    daily_cap: int


class TitleRule(BaseModel):
    from_steps: int
    title: str


class QuietRule(BaseModel):
    alert_levels: list[str]
    hours: int


class GardenConfig(BaseModel):
    activities: dict[Activity, ActivityRule]
    answered_dose_statuses: list[str]
    quiet: QuietRule
    step_points: int
    stages_per_plant: int
    max_plants: int
    titles: list[TitleRule]
    today: dict[Activity, str]


@lru_cache
def load_config(path: Path | None = None) -> GardenConfig:
    from app.config import get_settings

    path = path or get_settings().config_dir / "garden.yaml"
    return GardenConfig.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))


class Plant(BaseModel):
    stage: int  # 0 = not yet sprouted … stages_per_plant = in bloom


class TodayItem(BaseModel):
    activity: Activity
    title: str
    done: bool
    detail: str | None = None  # e.g. "1 из 2" for doses


class Garden(BaseModel):
    points: int
    steps: int
    title: str
    step_points: int
    points_in_step: int  # progress towards the next growth step: «8 из 10»
    plants: list[Plant]
    today: list[TodayItem]
    quiet: bool = False  # recent acute/high alert: no «Сад подрос», no grow-in animation


def compute(events: Iterable[tuple[Activity, date]], today: date, cfg: GardenConfig | None = None) -> Garden:
    """Points from (activity, day) events: per-day caps, events after `today` (demo clock) ignored."""
    cfg = cfg or load_config()
    per_day: dict[tuple[Activity, date], int] = defaultdict(int)
    for kind, day in events:
        if day <= today:
            per_day[(kind, day)] += 1
    points = sum(
        min(n, cfg.activities[kind].daily_cap) * cfg.activities[kind].points for (kind, _), n in per_day.items()
    )
    steps = points // cfg.step_points
    plants = [
        Plant(stage=max(0, min(cfg.stages_per_plant, steps - i * cfg.stages_per_plant))) for i in range(cfg.max_plants)
    ]
    title = next(t.title for t in sorted(cfg.titles, key=lambda t: -t.from_steps) if steps >= t.from_steps)
    return Garden(
        points=points,
        steps=steps,
        title=title,
        step_points=cfg.step_points,
        points_in_step=points % cfg.step_points,
        plants=plants,
        today=[],
    )


def _day(ts: datetime, tz) -> date:
    return ts.astimezone(tz).date()


async def garden_for(s: AsyncSession, token: str) -> Garden:
    cfg = load_config()
    now = clock.now()
    tz, today = now.tzinfo, now.date()
    events: list[tuple[Activity, date]] = []

    doses = (await s.scalars(select(Dose).where(Dose.token == token))).all()
    for d in doses:
        if d.status in cfg.answered_dose_statuses:
            events.append(("meds", _day(d.answered_at or d.planned_at, tz)))
    checkins = (await s.scalars(select(DailyCheckin).where(DailyCheckin.token == token))).all()
    for c in checkins:
        if c.mood is not None:
            events.append(("diary", c.day))
    records = (await s.scalars(select(QuestionnaireRecord).where(QuestionnaireRecord.token == token))).all()
    for r in records:
        if r.status != "complete":
            continue
        if r.kind in DIARY_KINDS:
            events.append(("diary", r.day))
        else:
            events.append(("questionnaire", _day(r.completed_at, tz) if r.completed_at else r.day))
    scales = (await s.scalars(select(ScaleResult).where(ScaleResult.token == token))).all()
    for sc in scales:
        if sc.complete:
            events.append(("questionnaire", _day(sc.ts, tz)))

    garden = compute(events, today, cfg)
    since = now - timedelta(hours=cfg.quiet.hours)
    recent_alert = await s.scalar(
        select(Alert.id)
        .where(Alert.token == token, Alert.level.in_(cfg.quiet.alert_levels), Alert.ts >= since, Alert.ts <= now)
        .limit(1)
    )
    garden.quiet = recent_alert is not None
    done_today = {kind for kind, day in events if day == today}

    items: list[TodayItem] = []
    due = [d for d in doses if _day(d.planned_at, tz) == today]
    if due:
        answered = sum(d.status in cfg.answered_dose_statuses for d in due)
        due_now = sum(d.planned_at <= now for d in due)
        items.append(
            TodayItem(
                activity="meds",
                title=cfg.today["meds"],
                done=answered > 0 and answered >= due_now,
                detail=f"{answered} из {len(due)}",
            )
        )
    items.append(TodayItem(activity="diary", title=cfg.today["diary"], done="diary" in done_today))
    # A questionnaire is not due every day: show it only once done, so an empty day is never a «missed» item.
    if "questionnaire" in done_today:
        items.append(TodayItem(activity="questionnaire", title=cfg.today["questionnaire"], done=True))
    garden.today = items
    return garden