CI / ruff · pytest (pull_request) Successful in 2m1s
- Patient list sorted red -> yellow -> grey -> green with KPI row, trend arrows, 7-day adherence bars, discontinuation-risk badge with factor tooltip. - Patient card: scales with severity bands from config/scales (unscheduled marked), mood/sleep/alcohol small multiples, dose heat-strip, alerts feed with ack, safety plan, AI summary with disclaimer. - Live alerts: st.fragment(run_every=2.5s) polling, toast + pinned acute banner on every page; full rerun on new alert only on doctor pages. - Connect form -> POST /patients/connect -> code, QR, deep link. - Thin httpx ApiClient over the frozen schemas with graceful "API недоступен". - app.py discovers dashboard/pages/*.py automatically (st.navigation); demo pages of #17 plug in by adding a file. Refs #16 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
"""Session-level plumbing shared by all dashboard pages: API client, demo "today", local acks, error banners."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from collections.abc import Callable
|
||
from datetime import date, datetime
|
||
from html import escape
|
||
from typing import TypeVar
|
||
|
||
import streamlit as st
|
||
|
||
from dashboard.api_client import ApiClient, ApiError, ApiUnavailable
|
||
from dashboard.components import ui
|
||
|
||
T = TypeVar("T")
|
||
|
||
# Short doctor names for the header; no other doctor PII is shown.
|
||
DOCTORS = {"doc-smirnova": ("д-р Смирнова Е. А.", "д-ра Смирновой Е. А.")}
|
||
|
||
PATIENTS_PAGE = "pages/1_Пациенты.py"
|
||
CONNECT_PAGE = "pages/1b_Подключить_пациента.py"
|
||
|
||
|
||
@st.cache_resource
|
||
def _client(base_url: str | None, doctor_id: str | None) -> ApiClient:
|
||
return ApiClient(base_url, doctor_id)
|
||
|
||
|
||
def client() -> ApiClient:
|
||
return _client(os.environ.get("API_BASE_URL"), os.environ.get("DOCTOR_ID"))
|
||
|
||
|
||
def doctor_names() -> tuple[str, str]:
|
||
did = client().doctor_id
|
||
return DOCTORS.get(did, (did, did))
|
||
|
||
|
||
def now() -> datetime | None:
|
||
"""Demo clock from the API (ADR-005), so day 13 of the scenario is "today" on screen; None if unknown."""
|
||
c = client().clock()
|
||
return c.now if c else None
|
||
|
||
|
||
def today() -> date | None:
|
||
n = now()
|
||
return n.date() if n else None
|
||
|
||
|
||
def local_acks() -> dict[int, tuple[str, datetime]]:
|
||
return st.session_state.setdefault("acked_local", {})
|
||
|
||
|
||
def call(fn: Callable[[], T]) -> T | None:
|
||
"""Run an API call; on failure show a calm banner instead of a traceback and return None."""
|
||
try:
|
||
return fn()
|
||
except ApiUnavailable as exc:
|
||
st.html(ui.banner(f"<b>API недоступен.</b> {escape(exc.message)}. Данные обновятся автоматически.", "grey"))
|
||
except ApiError as exc:
|
||
st.html(ui.banner(f"<b>Не удалось загрузить данные.</b> {escape(exc.message)}", "grey"))
|
||
return None
|