ElebobaandClaude Opus 5.5
2582c73d12
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>
86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
"""Live alerts across all pages: polls the API every few seconds, toasts new alerts, pins open acute ones on top.
|
|
|
|
Uses the built-in `st.fragment(run_every=...)`: only this fragment re-runs on the timer, the page is untouched
|
|
unless a new alert arrives on a doctor page (then the whole app reruns so the list and the card pick it up).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from html import escape
|
|
|
|
import streamlit as st
|
|
|
|
from app.api.schemas import AlertOut
|
|
from dashboard import theme
|
|
from dashboard.api_client import ApiError
|
|
from dashboard.components import helpers as h
|
|
from dashboard.components import state, ui
|
|
|
|
POLL_SECONDS = 2.5
|
|
MAX_BANNER = 3
|
|
|
|
|
|
def _toast(a: AlertOut) -> None:
|
|
icon = theme.TONES[h.level_tone(a.level)].icon
|
|
st.toast(f"**{a.patient_name}** — {a.reason}", icon=icon, duration="long" if a.level == "acute" else "short")
|
|
|
|
|
|
def flush_pending_toasts() -> None:
|
|
"""Toasts queued right before a full-app rerun are shown on the next run (a rerun drops pending deltas)."""
|
|
for a in st.session_state.pop("pending_toasts", []):
|
|
_toast(a)
|
|
|
|
|
|
def _banner(acute: list[AlertOut]) -> None:
|
|
for a in acute[:MAX_BANNER]:
|
|
left, right = st.columns([8, 1.6], vertical_alignment="center")
|
|
with left:
|
|
st.html(
|
|
ui.banner(
|
|
f"<b>Кризисный сигнал · {escape(a.patient_name)}</b> — {escape(a.reason)}"
|
|
f'<div style="font-size:.8rem;margin-top:2px">{escape(h.when_label(a.ts))} · '
|
|
"по протоколу — связаться с пациентом сегодня</div>",
|
|
tone="red",
|
|
pulse=True,
|
|
)
|
|
)
|
|
with right:
|
|
if st.button(
|
|
"Карточка", key=f"acute-open-{a.id}", type="primary", icon=":material/arrow_forward:", width="stretch"
|
|
):
|
|
st.switch_page(state.PATIENTS_PAGE, query_params={"patient": a.token})
|
|
if len(acute) > MAX_BANNER:
|
|
st.caption(f"Ещё кризисных сигналов: {len(acute) - MAX_BANNER}")
|
|
|
|
|
|
def render(*, rerun_app_on_new: bool = True) -> None:
|
|
"""Place once per run (app.py does it above every page)."""
|
|
|
|
@st.fragment(run_every=POLL_SECONDS)
|
|
def _poll() -> None:
|
|
ss = st.session_state
|
|
try:
|
|
alerts = state.client().alerts(open_only=True)
|
|
except ApiError:
|
|
return # pages show their own "API недоступен" banner; the poller just retries next tick
|
|
acked = state.local_acks()
|
|
alerts = [a for a in alerts if a.acked_at is None and a.id not in acked]
|
|
|
|
seen: set[int] | None = ss.get("seen_alert_ids")
|
|
if seen is None:
|
|
# First poll of the session: what is already open is not "new" — no toast storm on page load.
|
|
ss["seen_alert_ids"] = {a.id for a in alerts}
|
|
new: list[AlertOut] = []
|
|
else:
|
|
new = h.split_new(alerts, seen)
|
|
seen.update(a.id for a in new)
|
|
|
|
if new and rerun_app_on_new:
|
|
ss["pending_toasts"] = ss.get("pending_toasts", []) + new
|
|
st.rerun(scope="app")
|
|
for a in new:
|
|
_toast(a)
|
|
_banner(h.sort_alerts(a for a in alerts if a.level == "acute"))
|
|
|
|
_poll()
|