ElebobaandClaude Opus 5.5
260422138a
- pages/2_Демо-консоль.py: persona selector, virtual clock bar (+1 h, to 08:00/20:00, next day, reset with confirm), fallback patient web chat (buttons from Reply.buttons, kind styling, outbox drained after every action and on a 2 s fragment poll), scenario panel - pages/3_Сплит-экран.py: patient chat or PATIENT_WEB_URL iframe on the left, compact doctor panel (patients, live alerts, red banner on a new acute alert) on the right - console/playback.py: data/synthetic/<persona>.jsonl runner (advance_to/press/text/ expect_color/start), date chaining between days, pause on failure with retry/skip/manual - console/api.py: own HTTP client over the frozen API contract, ApiError only Refs #17 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
"""In-memory transcript of the patient chat shown by the console (kept in Streamlit session state).
|
|
|
|
Entries are plain dicts: `{"dir": "out", "text", "kind", "buttons"}` for bot replies (the `Reply` contract)
|
|
and `{"dir": "in", "text", "pressed": bool}` for what the patient sent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
from typing import Any
|
|
|
|
MAX_ENTRIES = 400
|
|
|
|
|
|
def normalize_reply(raw: dict[str, Any]) -> dict[str, Any]:
|
|
"""Coerce an API reply into a safe shape: missing fields get defaults, malformed buttons are dropped."""
|
|
rows: list[list[dict[str, str]]] = []
|
|
for row in raw.get("buttons") or []:
|
|
if not isinstance(row, list):
|
|
continue
|
|
clean = [
|
|
{"label": str(b.get("label", "")), "data": str(b.get("data", ""))}
|
|
for b in row
|
|
if isinstance(b, dict) and b.get("data") is not None
|
|
]
|
|
if clean:
|
|
rows.append(clean)
|
|
return {
|
|
"dir": "out",
|
|
"text": str(raw.get("text") or ""),
|
|
"kind": str(raw.get("kind") or "normal"),
|
|
"buttons": rows,
|
|
}
|
|
|
|
|
|
class ChatLog:
|
|
"""Wraps a list owned by the caller (session state), so every page sees the same transcript."""
|
|
|
|
def __init__(self, entries: list[dict[str, Any]] | None = None):
|
|
self.entries: list[dict[str, Any]] = entries if entries is not None else []
|
|
|
|
def _trim(self) -> None:
|
|
if len(self.entries) > MAX_ENTRIES:
|
|
del self.entries[: len(self.entries) - MAX_ENTRIES]
|
|
|
|
def add_in(self, text: str, *, pressed: bool = False) -> None:
|
|
self.entries.append({"dir": "in", "text": text, "pressed": pressed})
|
|
self._trim()
|
|
|
|
def add_replies(self, replies: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
added = [normalize_reply(r) for r in replies if isinstance(r, dict)]
|
|
self.entries.extend(added)
|
|
self._trim()
|
|
return added
|
|
|
|
def add_history(self, messages: Iterable[dict[str, Any]]) -> None:
|
|
"""Prefill from `GET /dialog/{token}/history` (MessageOut: no buttons)."""
|
|
for m in messages:
|
|
if not isinstance(m, dict):
|
|
continue
|
|
if m.get("direction") == "in":
|
|
self.entries.append({"dir": "in", "text": str(m.get("text", "")), "pressed": False})
|
|
else:
|
|
self.entries.append(normalize_reply({"text": m.get("text"), "kind": m.get("kind")}))
|
|
self._trim()
|
|
|
|
def recent_replies(self, n: int = 10) -> list[dict[str, Any]]:
|
|
"""Last `n` bot replies, oldest first."""
|
|
outs = [e for e in self.entries if e.get("dir") == "out"]
|
|
return outs[-n:] if n > 0 else []
|
|
|
|
def last_buttons_index(self) -> int | None:
|
|
"""Index of the newest bot reply that carries buttons — only its keyboard is clickable in the web chat."""
|
|
for i in range(len(self.entries) - 1, -1, -1):
|
|
e = self.entries[i]
|
|
if e.get("dir") == "out" and e.get("buttons"):
|
|
return i
|
|
return None
|
|
|
|
def clear(self) -> None:
|
|
self.entries.clear()
|