455 lines
20 KiB
Python
455 lines
20 KiB
Python
"""Streamlit widgets shared by the demo console and the split screen (#17).
|
||
|
||
Robustness rules for the live demo: every API call goes through `ConsoleApi` (raises `ApiError` only),
|
||
callbacks never raise — errors are queued as flash messages and shown on the next render — and the chat
|
||
transcript / playback progress live in `st.session_state`, so both pages share them within one browser tab.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import contextlib
|
||
import os
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
import streamlit as st
|
||
|
||
from .api import ApiError, ConsoleApi, api_base_url, doctor_id, personas_or_fallback
|
||
from .chatlog import ChatLog
|
||
from .playback import PlaybackRunner, PlaybackState, StepLog, parse_clock_now, target_datetime
|
||
from .scenario import Scenario, default_scenario_path, load_scenario, parse_scenario
|
||
|
||
COLOR_EMOJI = {"red": "🔴", "yellow": "🟡", "green": "🟢", "grey": "⚪"}
|
||
LEVEL_EMOJI = {"acute": "🔴", "high": "🟠", "moderate": "🟡", "info": "⚪"}
|
||
WEEKDAYS = ("пн", "вт", "ср", "чт", "пт", "сб", "вс")
|
||
POLL_SECONDS = 2
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ state
|
||
@st.cache_resource
|
||
def _api_for(base_url: str) -> ConsoleApi:
|
||
return ConsoleApi(base_url)
|
||
|
||
|
||
def get_api() -> ConsoleApi:
|
||
return _api_for(api_base_url())
|
||
|
||
|
||
def chat_for(token: str) -> ChatLog:
|
||
key = f"console_chat:{token}"
|
||
if key not in st.session_state:
|
||
entries: list[dict[str, Any]] = []
|
||
st.session_state[key] = entries
|
||
with contextlib.suppress(ApiError): # history is a nice-to-have: the chat may start empty
|
||
ChatLog(entries).add_history(get_api().history(token))
|
||
return ChatLog(st.session_state[key])
|
||
|
||
|
||
def playback_state(token: str) -> PlaybackState:
|
||
key = f"console_playback:{token}"
|
||
if key not in st.session_state:
|
||
st.session_state[key] = PlaybackState()
|
||
return st.session_state[key]
|
||
|
||
|
||
def flash(message: str, level: str = "error") -> None:
|
||
"""Report a problem from a callback: a toast now plus a persistent note shown by `show_flash`."""
|
||
icon = {"error": "❌", "warning": "⚠️", "success": "✅"}.get(level, "ℹ️")
|
||
# Outside a script run context (tests, threads) a toast may fail — the persistent note below is enough.
|
||
with contextlib.suppress(Exception):
|
||
st.toast(message, icon=icon)
|
||
st.session_state["console_last_note"] = (level, message)
|
||
|
||
|
||
def show_flash(container: Any = None) -> None:
|
||
"""Last error/notice of the session, kept until the next one (does not vanish with the 2 s poll)."""
|
||
note = st.session_state.get("console_last_note")
|
||
if not note:
|
||
return
|
||
level, message = note
|
||
where = container or st.sidebar
|
||
{"error": where.error, "warning": where.warning, "success": where.success}.get(level, where.info)(message)
|
||
|
||
|
||
def forget_demo_state() -> None:
|
||
"""After «Сброс демо»: drop transcripts, playback progress and the alert baseline of every persona."""
|
||
for key in list(st.session_state.keys()):
|
||
if str(key).startswith(("console_chat:", "console_playback:", "console_alerts_")):
|
||
del st.session_state[key]
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ persona
|
||
@st.cache_data(ttl=15, show_spinner=False)
|
||
def _personas(base_url: str) -> tuple[list[dict[str, Any]], str | None]:
|
||
return personas_or_fallback(get_api())
|
||
|
||
|
||
def select_persona(container: Any = None) -> dict[str, Any]:
|
||
"""Persona selector (sidebar by default); the choice is shared by all console pages."""
|
||
where = container or st.sidebar
|
||
personas, error = _personas(get_api().base_url)
|
||
if error:
|
||
where.caption(f"⚠️ {error}")
|
||
tokens = [p["token"] for p in personas]
|
||
current = st.session_state.get("console_token")
|
||
index = tokens.index(current) if current in tokens else 0
|
||
labels = {p["token"]: f"{p.get('display_name') or p['token']} ({p['token'][:12]})" for p in personas}
|
||
token = where.selectbox("Пациент демо", tokens, index=index, format_func=lambda t: labels.get(t, t))
|
||
st.session_state["console_token"] = token
|
||
return next(p for p in personas if p["token"] == token)
|
||
|
||
|
||
# Synthetic demo hero (docs/product/01_demo_persona.md); the SNILS is fake and used only to derive the token.
|
||
DEMO_HERO = {
|
||
"display_name": "Андрей С.",
|
||
"birth_year": 1992,
|
||
"sex": "m",
|
||
"snils": "000-000-017 17",
|
||
"drug": "сертралин",
|
||
"dose": "50 мг",
|
||
"times": ["08:00"],
|
||
"instructions": "1 раз в сутки утром после еды",
|
||
"modules": ["veteran", "depression"],
|
||
}
|
||
|
||
|
||
def _connect_hero() -> None:
|
||
name = st.session_state.get("console_hero_name") or DEMO_HERO["display_name"]
|
||
body = {**DEMO_HERO, "doctor_id": doctor_id(), "display_name": name}
|
||
try:
|
||
out = get_api().connect(body)
|
||
except ApiError as exc:
|
||
hint = " — пациент уже подключён, выберите его в списке" if "409" in str(exc) else ""
|
||
flash(f"Подключение не выполнено: {exc}{hint}", "warning" if hint else "error")
|
||
return
|
||
token = str(out.get("token", ""))
|
||
st.session_state["console_token"] = token
|
||
st.session_state[f"console_code:{token}"] = str(out.get("code", ""))
|
||
_personas.clear()
|
||
flash(f"Пациент подключён, код {out.get('code', '')}. Нажмите «Начать» в чате.", "success")
|
||
|
||
|
||
def render_connect_hero(container: Any = None) -> None:
|
||
"""Fallback when the demo seed is missing: connect the synthetic hero through the doctor API."""
|
||
where = container or st.sidebar
|
||
with where.expander("Подключить демо-пациента"):
|
||
st.caption("Если сид демо не загружен: врач подключает Андрея (синтетика) — как на приёме.")
|
||
st.text_input("Имя", value="Андрей С.", key="console_hero_name")
|
||
st.button("Подключить", key="console-connect", on_click=_connect_hero)
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ clock
|
||
def _drain_quiet(api: ConsoleApi, token: str) -> None:
|
||
try:
|
||
chat_for(token).add_replies(api.outbox(token))
|
||
except ApiError as exc:
|
||
flash(f"Не удалось забрать входящие: {exc}", "warning")
|
||
|
||
|
||
def _advance(token: str, days: float = 0, hours: float = 0) -> None:
|
||
api = get_api()
|
||
try:
|
||
api.advance(days=days, hours=hours)
|
||
except ApiError as exc:
|
||
flash(f"Часы не сдвинулись: {exc}")
|
||
return
|
||
_drain_quiet(api, token)
|
||
|
||
|
||
def _advance_to(token: str, hhmm: str) -> None:
|
||
api = get_api()
|
||
try:
|
||
now = parse_clock_now(api.clock())
|
||
hours = (target_datetime(now, hhmm, None) - now).total_seconds() / 3600
|
||
if hours > 0:
|
||
api.advance(hours=hours)
|
||
except (ApiError, ValueError) as exc:
|
||
flash(f"Часы не сдвинулись: {exc}")
|
||
return
|
||
_drain_quiet(api, token)
|
||
|
||
|
||
def _reset() -> None:
|
||
try:
|
||
get_api().reset()
|
||
except ApiError as exc:
|
||
flash(f"Сброс не выполнен: {exc}")
|
||
return
|
||
forget_demo_state()
|
||
_personas.clear() # the seed may be recreated on reset
|
||
flash("Демо сброшено: часы вернулись к реальному времени, чат и прогресс сценария очищены.", "success")
|
||
|
||
|
||
def _therapy_day(api: ConsoleApi, persona: dict[str, Any]) -> int | None:
|
||
try:
|
||
me = next((p for p in api.patients(doctor_id()) if p.get("token") == persona["token"]), None)
|
||
except ApiError:
|
||
me = None
|
||
day = (me or persona).get("therapy_day")
|
||
return day if isinstance(day, int) else None
|
||
|
||
|
||
def format_now(now: datetime) -> str:
|
||
return f"{WEEKDAYS[now.weekday()]} {now:%d.%m.%Y} · {now:%H:%M}"
|
||
|
||
|
||
def render_clock_bar(persona: dict[str, Any], *, key: str, with_reset: bool = True) -> None:
|
||
"""Virtual date/time, therapy day and the time controls (ADR-005)."""
|
||
api = get_api()
|
||
token = persona["token"]
|
||
info, *buttons = st.columns([3.2, 1, 1.2, 1.2, 1.3, 1.2] if with_reset else [3.2, 1, 1.2, 1.2, 1.3])
|
||
try:
|
||
clock = api.clock()
|
||
now = parse_clock_now(clock)
|
||
day = _therapy_day(api, persona)
|
||
mode = "виртуальные часы" if clock.get("virtual") else "реальное время"
|
||
info.markdown(f"### 🕒 {format_now(now)}")
|
||
info.caption(f"День терапии: **{day if day is not None else '—'}** · {mode}")
|
||
except (ApiError, ValueError) as exc:
|
||
info.warning(f"Часы недоступны: {exc}")
|
||
buttons[0].button("+1 час", key=f"{key}-h1", on_click=_advance, args=(token,), kwargs={"hours": 1})
|
||
buttons[1].button("К 08:00 (приём)", key=f"{key}-08", on_click=_advance_to, args=(token, "08:00"))
|
||
buttons[2].button("К 20:00 (чек-ин)", key=f"{key}-20", on_click=_advance_to, args=(token, "20:00"))
|
||
buttons[3].button(
|
||
"Следующий день", key=f"{key}-d1", type="primary", on_click=_advance, args=(token,), kwargs={"days": 1}
|
||
)
|
||
if with_reset:
|
||
with buttons[4].popover("Сброс демо"):
|
||
st.write("Вернуть часы к реальному времени и очистить чат и прогресс сценария?")
|
||
st.button("Да, сбросить", key=f"{key}-reset", type="primary", on_click=_reset)
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ chat
|
||
def _send(token: str, type_: str, *, text: str | None = None, button: str | None = None, label: str = "") -> None:
|
||
api = get_api()
|
||
chat = chat_for(token)
|
||
params: dict[str, Any] = {}
|
||
if type_ == "start":
|
||
code = (st.session_state.get(f"console_code:{token}") or "").strip()
|
||
params = {"code": code} if code else {}
|
||
chat.add_in("/start" + (f" {code}" if code else ""), pressed=True)
|
||
elif type_ == "button":
|
||
chat.add_in(label, pressed=True)
|
||
else:
|
||
chat.add_in(text or "")
|
||
try:
|
||
chat.add_replies(api.send_event(token, type_, text=text, button=button, params=params))
|
||
except ApiError as exc:
|
||
flash(f"Сообщение не доставлено: {exc}")
|
||
return
|
||
_drain_quiet(api, token)
|
||
|
||
|
||
def _submit_text(token: str, key: str) -> None:
|
||
text = (st.session_state.get(key) or "").strip()
|
||
if text:
|
||
_send(token, "text", text=text)
|
||
|
||
|
||
def _render_entry(entry: dict[str, Any]) -> None:
|
||
if entry.get("dir") == "in":
|
||
with st.chat_message("user", avatar="🧑"):
|
||
st.markdown(f"🔘 *{entry['text']}*" if entry.get("pressed") else entry["text"])
|
||
return
|
||
kind = entry.get("kind", "normal")
|
||
with st.chat_message("assistant", avatar="🤝"):
|
||
if kind == "crisis":
|
||
st.error(entry["text"], icon="🚨")
|
||
elif kind == "safety_plan":
|
||
with st.container(border=True):
|
||
st.markdown("**🛡️ План безопасности**")
|
||
st.markdown(entry["text"])
|
||
elif kind == "info":
|
||
st.caption(entry["text"])
|
||
else:
|
||
st.markdown(entry["text"])
|
||
|
||
|
||
def render_messages(token: str, *, height: int, key: str) -> None:
|
||
"""Transcript + the keyboard of the newest reply with buttons. Drains the outbox on every call."""
|
||
_drain_quiet(get_api(), token)
|
||
chat = chat_for(token)
|
||
box = st.container(height=height, border=True, autoscroll=True)
|
||
with box:
|
||
if not chat.entries:
|
||
st.caption("Сообщений пока нет. Нажмите «Начать» или сдвиньте часы к напоминанию.")
|
||
active = chat.last_buttons_index()
|
||
for i, entry in enumerate(chat.entries):
|
||
_render_entry(entry)
|
||
if i == active:
|
||
for r, row in enumerate(entry["buttons"]):
|
||
cols = st.columns(len(row))
|
||
for c, b in enumerate(row):
|
||
cols[c].button(
|
||
b["label"] or "…",
|
||
key=f"{key}-b{i}-{r}-{c}",
|
||
width="stretch",
|
||
on_click=_send,
|
||
args=(token, "button"),
|
||
kwargs={"button": b["data"], "label": b["label"]},
|
||
)
|
||
|
||
|
||
@st.fragment(run_every=POLL_SECONDS)
|
||
def _polled_messages(token: str, height: int, key: str) -> None:
|
||
render_messages(token, height=height, key=key)
|
||
|
||
|
||
def render_chat(token: str, *, height: int = 520, key: str = "chat") -> None:
|
||
"""Fallback patient web chat: polled transcript, text input, «Начать» with an optional connect code."""
|
||
_polled_messages(token, height, key)
|
||
input_key = f"{key}-input:{token}"
|
||
with st.form(f"{key}-form:{token}", clear_on_submit=True, border=False):
|
||
left, right = st.columns([5, 1])
|
||
left.text_input("Сообщение", key=input_key, label_visibility="collapsed", placeholder="Сообщение пациента…")
|
||
right.form_submit_button("➤", width="stretch", on_click=_submit_text, args=(token, input_key))
|
||
code = st.query_params.get("code") # connect links from the doctor point to the console with ?code=
|
||
if code and not st.session_state.get(f"console_code:{token}"):
|
||
st.session_state[f"console_code:{token}"] = code
|
||
with st.expander("Начать диалог / код подключения", expanded=bool(code)):
|
||
st.text_input("Код подключения (необязательно)", key=f"console_code:{token}", placeholder="DEMO-0001")
|
||
c1, c2 = st.columns(2)
|
||
c1.button("🚀 Начать", key=f"{key}-start", on_click=_send, args=(token, "start"), width="stretch")
|
||
c2.button(
|
||
"Очистить окно чата",
|
||
key=f"{key}-clear",
|
||
on_click=lambda: chat_for(token).clear(),
|
||
width="stretch",
|
||
)
|
||
|
||
|
||
def patient_web_url() -> str | None:
|
||
url = (os.environ.get("PATIENT_WEB_URL") or "").strip()
|
||
return url or None
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ doctor panel
|
||
def _dismiss_banner(alert_id: int) -> None:
|
||
st.session_state.setdefault("console_alerts_dismissed", set()).add(alert_id)
|
||
|
||
|
||
def _ack(alert_id: int) -> None:
|
||
try:
|
||
get_api().ack(alert_id, doctor_id())
|
||
except ApiError as exc:
|
||
flash(f"Алерт не подтверждён: {exc}")
|
||
_dismiss_banner(alert_id)
|
||
|
||
|
||
def _fmt_ts(raw: Any) -> str:
|
||
try:
|
||
return datetime.fromisoformat(str(raw).replace("Z", "+00:00")).strftime("%d.%m %H:%M")
|
||
except ValueError:
|
||
return ""
|
||
|
||
|
||
def render_doctor_panel(highlight_token: str | None, *, key: str = "doc") -> None:
|
||
"""Compact doctor view for the projector: patients with colours and the live alerts feed."""
|
||
api = get_api()
|
||
doctor = doctor_id()
|
||
try:
|
||
alerts = [a for a in api.alerts(doctor) if isinstance(a, dict)]
|
||
except ApiError as exc:
|
||
alerts = []
|
||
st.warning(f"Алерты недоступны: {exc}")
|
||
ids = [int(a.get("id", -1)) for a in alerts]
|
||
# Alerts already present when the panel first loads are "old": the banner is only for new acute ones.
|
||
if "console_alerts_baseline" not in st.session_state:
|
||
st.session_state["console_alerts_baseline"] = max(ids, default=-1)
|
||
baseline = st.session_state["console_alerts_baseline"]
|
||
dismissed = st.session_state.get("console_alerts_dismissed", set())
|
||
fresh_acute = [
|
||
a
|
||
for a in alerts
|
||
if a.get("level") == "acute"
|
||
and not a.get("acked_at")
|
||
and baseline < int(a.get("id", -1))
|
||
and int(a.get("id", -1)) not in dismissed
|
||
]
|
||
for a in fresh_acute[:1]:
|
||
st.markdown(
|
||
"<div class='mt-banner t-red mt-console-alert'><div>"
|
||
f"🚨 ОСТРЫЙ АЛЕРТ · {_escape(a.get('patient_name', ''))} · {_fmt_ts(a.get('ts'))}<br>"
|
||
f"<div class='mt-sub'>{_escape(a.get('reason', ''))}</div></div></div>",
|
||
unsafe_allow_html=True,
|
||
)
|
||
c1, c2 = st.columns(2)
|
||
aid = int(a.get("id", -1))
|
||
c1.button("Взять в работу (ack)", key=f"{key}-ack{aid}", type="primary", on_click=_ack, args=(aid,))
|
||
c2.button("Скрыть", key=f"{key}-hide{aid}", on_click=_dismiss_banner, args=(aid,))
|
||
|
||
st.markdown("#### Пациенты")
|
||
try:
|
||
patients = [p for p in api.patients(doctor) if isinstance(p, dict)]
|
||
except ApiError as exc:
|
||
patients = []
|
||
st.warning(f"Список пациентов недоступен: {exc}")
|
||
lines = []
|
||
for p in patients:
|
||
name = str(p.get("display_name", "?"))
|
||
if p.get("token") == highlight_token:
|
||
name = f"**{name} ◀**"
|
||
extra = f" · {p['last_event']}" if p.get("last_event") else ""
|
||
lines.append(
|
||
f"{COLOR_EMOJI.get(p.get('status_color', ''), '⚪')} {name} · день {p.get('therapy_day', '—')}{extra}"
|
||
)
|
||
st.markdown("\n".join(f"- {line}" for line in lines) if lines else "_нет пациентов_")
|
||
|
||
st.markdown("#### Алерты")
|
||
if not alerts:
|
||
st.caption("Новых алертов нет")
|
||
for a in alerts[:8]:
|
||
done = " ✔︎" if a.get("acked_at") else ""
|
||
icon = LEVEL_EMOJI.get(a.get("level", ""), "⚪")
|
||
who = a.get("patient_name", "")
|
||
st.markdown(f"{icon} `{_fmt_ts(a.get('ts'))}` **{who}** — {a.get('reason', '')}{done}")
|
||
|
||
|
||
@st.fragment(run_every=POLL_SECONDS)
|
||
def polled_doctor_panel(highlight_token: str | None, key: str = "doc") -> None:
|
||
render_doctor_panel(highlight_token, key=key)
|
||
|
||
|
||
def _escape(value: Any) -> str:
|
||
text = str(value)
|
||
for ch, rep in (("&", "&"), ("<", "<"), (">", ">")):
|
||
text = text.replace(ch, rep)
|
||
return text
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ playback
|
||
def scenario_for(persona: dict[str, Any]) -> tuple[Scenario, str]:
|
||
"""Scenario of the persona: uploaded text (if any) wins over the repo file. Returns (scenario, source)."""
|
||
token = persona["token"]
|
||
uploaded = st.session_state.get(f"console_scenario_text:{token}")
|
||
if uploaded:
|
||
return parse_scenario(uploaded), "загруженный файл"
|
||
path = st.session_state.get(f"console_scenario_path:{token}") or default_scenario_path(persona)
|
||
if not path:
|
||
return Scenario(errors=["у персоны нет файла сценария"]), "—"
|
||
return load_scenario(path), path
|
||
|
||
|
||
def make_runner(persona: dict[str, Any], *, delay: float) -> PlaybackRunner:
|
||
token = persona["token"]
|
||
return PlaybackRunner(get_api(), token, chat_for(token), playback_state(token), doctor=doctor_id(), delay=delay)
|
||
|
||
|
||
def execute(logs: Any, state: PlaybackState, *, title: str) -> bool:
|
||
"""Run a playback iterator with a live log. Returns True if it finished without a failed step."""
|
||
ok = True
|
||
state.log.clear()
|
||
with st.status(title, expanded=True) as status:
|
||
try:
|
||
for entry in logs:
|
||
line = entry.render() if isinstance(entry, StepLog) else str(entry)
|
||
state.log.append(line)
|
||
status.write(line)
|
||
ok = ok and entry.status != "fail"
|
||
except Exception as exc: # last-resort guard: the page must stay alive
|
||
ok = False
|
||
state.log.append(f"❌ сбой проигрывания: {exc}")
|
||
status.write(state.log[-1])
|
||
status.update(label=f"{title}: {'готово' if ok else 'остановлено'}", state="complete" if ok else "error")
|
||
return ok
|