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>
313 lines
14 KiB
Python
313 lines
14 KiB
Python
"""Scenario playback over the HTTP API (#17): runs `data/synthetic/<persona>.jsonl` step by step.
|
||
|
||
Time handling. A scenario day is a calendar date on the demo clock. When day N is played, its date is
|
||
derived from the last played day M: date(N) = date(M) + (N − M); with nothing played yet (or if that date
|
||
is already behind the clock, e.g. after manual «Следующий день») `advance_to HH:MM` goes to the next
|
||
occurrence of HH:MM. The clock only moves forward (`/demo/clock/advance` replays the scheduler), so a target
|
||
already in the past is skipped with a warning instead of failing.
|
||
|
||
Failure handling. Every step is wrapped: an error stops playback, the position is kept in `PlaybackState`
|
||
and the page offers «повторить шаг» / «пропустить шаг» / «продолжить вручную». The demo never crashes.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time as _time
|
||
from collections.abc import Callable, Iterator, Sequence
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime, time, timedelta
|
||
from typing import Any, Literal
|
||
|
||
from .api import DEFAULT_DOCTOR, ApiError, DemoApi
|
||
from .chatlog import ChatLog
|
||
from .scenario import Action, DayScript, Scenario, parse_hhmm
|
||
|
||
# Virtual time keeps ticking between steps: a target a few minutes behind "now" counts as reached.
|
||
TOLERANCE = timedelta(minutes=5)
|
||
PRESS_WINDOW = 10
|
||
|
||
StepStatus = Literal["ok", "warn", "fail", "info"]
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ pure helpers
|
||
def _norm(s: str) -> str:
|
||
return " ".join(s.casefold().replace("ё", "е").split())
|
||
|
||
|
||
def resolve_press(replies: Sequence[dict[str, Any]], label: str, window: int = PRESS_WINDOW) -> tuple[str, str] | None:
|
||
"""Find the button to press for a scenario `press` step.
|
||
|
||
Searches the last `window` replies newest first; inside a reply an exact (case/ё-insensitive) label match
|
||
wins over a substring match. Returns `(data, label)` or None. Button `data` is dynamic (dose ids), so
|
||
scenarios address buttons by label.
|
||
"""
|
||
needle = _norm(label)
|
||
if not needle:
|
||
return None
|
||
for reply in reversed(list(replies)[-window:] if window > 0 else []):
|
||
buttons = [b for row in reply.get("buttons") or [] for b in row if isinstance(b, dict)]
|
||
for exact in (True, False):
|
||
for b in buttons:
|
||
text = _norm(str(b.get("label", "")))
|
||
if (text == needle) if exact else (needle in text):
|
||
return str(b.get("data", "")), str(b.get("label", ""))
|
||
return None
|
||
|
||
|
||
def parse_clock_now(clock: dict[str, Any]) -> datetime:
|
||
"""`ClockOut.now` (ISO, tz-aware) → datetime; raises ValueError on garbage."""
|
||
raw = clock.get("now")
|
||
if not isinstance(raw, str):
|
||
raise ValueError("часы API не вернули поле now")
|
||
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
||
|
||
|
||
def target_datetime(now: datetime, hhmm: str, day_date: date | None) -> datetime:
|
||
"""Moment to move the clock to for `advance_to HH:MM` (see module docstring for the rules)."""
|
||
h, m = parse_hhmm(hhmm)
|
||
if day_date is None:
|
||
cand = now.replace(hour=h, minute=m, second=0, microsecond=0)
|
||
if cand < now - TOLERANCE:
|
||
cand += timedelta(days=1)
|
||
return cand
|
||
return datetime.combine(day_date, time(h, m), tzinfo=now.tzinfo)
|
||
|
||
|
||
def hours_until(now: datetime, target: datetime) -> float | None:
|
||
"""Hours to advance (≥ 0), 0 within the tolerance window, None if the target is clearly in the past."""
|
||
delta = target - now
|
||
if delta >= timedelta(0):
|
||
return delta.total_seconds() / 3600
|
||
if delta >= -TOLERANCE:
|
||
return 0.0
|
||
return None
|
||
|
||
|
||
def planned_date(dates: dict[int, date], day: int, today: date) -> date | None:
|
||
"""Calendar date for scenario `day` from already played days, or None to use "next occurrence" rules."""
|
||
if day in dates:
|
||
cand = dates[day]
|
||
else:
|
||
earlier = [d for d in dates if d < day]
|
||
if not earlier:
|
||
return None
|
||
prev = max(earlier)
|
||
cand = dates[prev] + timedelta(days=day - prev)
|
||
return cand if cand >= today else None
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ runner
|
||
@dataclass
|
||
class StepLog:
|
||
day: int
|
||
index: int # 0-based step index inside the day; -1 for day-level messages
|
||
status: StepStatus
|
||
message: str
|
||
action: Action | None = None
|
||
|
||
def render(self) -> str:
|
||
icon = {"ok": "✅", "warn": "⚠️", "fail": "❌", "info": "ℹ️"}[self.status]
|
||
step = f"шаг {self.index + 1}: " if self.index >= 0 else ""
|
||
return f"{icon} день {self.day}, {step}{self.message}"
|
||
|
||
|
||
@dataclass
|
||
class Resume:
|
||
day: int
|
||
step: int
|
||
until: int
|
||
|
||
|
||
@dataclass
|
||
class PlaybackState:
|
||
"""Per-persona playback memory kept in session state; survives an interrupted Streamlit run."""
|
||
|
||
dates: dict[int, date] = field(default_factory=dict) # scenario day → calendar date it was played on
|
||
done: list[int] = field(default_factory=list) # fully played days
|
||
failed: Resume | None = None # where playback stopped on an error
|
||
position: Resume | None = None # step currently being executed (left set if the run was interrupted)
|
||
log: list[str] = field(default_factory=list)
|
||
|
||
@property
|
||
def last_done(self) -> int | None:
|
||
return max(self.done) if self.done else None
|
||
|
||
def reset(self) -> None:
|
||
self.dates.clear()
|
||
self.done.clear()
|
||
self.failed = None
|
||
self.position = None
|
||
self.log.clear()
|
||
|
||
|
||
class StepFailed(Exception):
|
||
pass
|
||
|
||
|
||
class PlaybackRunner:
|
||
def __init__(
|
||
self,
|
||
api: DemoApi,
|
||
token: str,
|
||
chat: ChatLog,
|
||
state: PlaybackState,
|
||
*,
|
||
doctor: str = DEFAULT_DOCTOR,
|
||
delay: float = 0.0,
|
||
press_retries: int = 3,
|
||
retry_wait: float = 0.7,
|
||
sleep: Callable[[float], None] = _time.sleep,
|
||
):
|
||
self.api = api
|
||
self.token = token
|
||
self.chat = chat
|
||
self.state = state
|
||
self.doctor = doctor
|
||
self.delay = delay
|
||
self.press_retries = press_retries
|
||
self.retry_wait = retry_wait
|
||
self.sleep = sleep
|
||
|
||
# ---------------------------------------------------------------- helpers
|
||
def drain(self) -> list[dict[str, Any]]:
|
||
"""Pull system-initiated replies (reminders, check-ins) into the chat."""
|
||
return self.chat.add_replies(self.api.outbox(self.token))
|
||
|
||
def _now(self) -> datetime:
|
||
return parse_clock_now(self.api.clock())
|
||
|
||
@staticmethod
|
||
def _preview(replies: list[dict[str, Any]]) -> str:
|
||
if not replies:
|
||
return ""
|
||
text = replies[-1].get("text", "").replace("\n", " ")
|
||
more = f" (+{len(replies) - 1})" if len(replies) > 1 else ""
|
||
return f" → «{text[:70]}{'…' if len(text) > 70 else ''}»{more}"
|
||
|
||
def next_day(self, scenario: Scenario) -> DayScript | None:
|
||
last = self.state.last_done
|
||
return next((d for d in scenario.days if last is None or d.day > last), None)
|
||
|
||
# ---------------------------------------------------------------- one step
|
||
def _step(self, day: DayScript, action: Action, day_date: list[date | None]) -> tuple[StepStatus, str]:
|
||
if action.kind == "advance_to":
|
||
now = self._now()
|
||
target = target_datetime(now, action.value, day_date[0])
|
||
if day_date[0] is None:
|
||
day_date[0] = target.date()
|
||
self.state.dates[day.day] = target.date()
|
||
hours = hours_until(now, target)
|
||
if hours is None:
|
||
return "warn", f"{action.describe()}: время уже прошло ({now:%d.%m %H:%M}), пропускаю"
|
||
if hours > 0:
|
||
self.api.advance(hours=hours)
|
||
got = self.drain()
|
||
return "ok", f"{action.describe()} ({target:%d.%m %H:%M}){self._preview(got)}"
|
||
|
||
if action.kind == "press":
|
||
found = resolve_press(self.chat.recent_replies(PRESS_WINDOW), action.value)
|
||
for _ in range(self.press_retries):
|
||
if found is not None:
|
||
break
|
||
# The reply may still be on its way (scheduler tick): wait and drain the outbox again.
|
||
self.sleep(self.retry_wait)
|
||
self.drain()
|
||
found = resolve_press(self.chat.recent_replies(PRESS_WINDOW), action.value)
|
||
if found is None:
|
||
raise StepFailed(f"кнопка «{action.value}» не найдена в последних {PRESS_WINDOW} сообщениях")
|
||
data, label = found
|
||
self.chat.add_in(label, pressed=True)
|
||
got = self.chat.add_replies(self.api.send_event(self.token, "button", button=data))
|
||
got += self.drain()
|
||
return "ok", f"{action.describe()}{self._preview(got)}"
|
||
|
||
if action.kind == "text":
|
||
self.chat.add_in(action.value)
|
||
got = self.chat.add_replies(self.api.send_event(self.token, "text", text=action.value))
|
||
got += self.drain()
|
||
return "ok", f"{action.describe()}{self._preview(got)}"
|
||
|
||
if action.kind == "start":
|
||
params = {"code": action.value} if action.value else {}
|
||
self.chat.add_in("/start", pressed=True)
|
||
got = self.chat.add_replies(self.api.send_event(self.token, "start", params=params))
|
||
got += self.drain()
|
||
return "ok", f"{action.describe()}{self._preview(got)}"
|
||
|
||
# expect_color is a soft check: a mismatch is shown to the presenter but never stops the demo.
|
||
try:
|
||
patients = self.api.patients(self.doctor)
|
||
except ApiError as exc:
|
||
return "warn", f"{action.describe()}: не удалось проверить ({exc})"
|
||
me = next((p for p in patients if isinstance(p, dict) and p.get("token") == self.token), None)
|
||
if me is None:
|
||
return "warn", f"{action.describe()}: пациента нет в списке врача"
|
||
actual = me.get("status_color")
|
||
if actual == action.value:
|
||
return "ok", f"{action.describe()}: совпало"
|
||
return "warn", f"{action.describe()}: у врача сейчас {actual}"
|
||
|
||
# ---------------------------------------------------------------- days
|
||
def run(self, scenario: Scenario, from_day: int, from_step: int, until_day: int) -> Iterator[StepLog]:
|
||
"""Play days `from_day..until_day` (those present in the scenario), the first one from `from_step`."""
|
||
days = [d for d in scenario.days if from_day <= d.day <= until_day]
|
||
if not days:
|
||
yield StepLog(from_day, -1, "info", "в сценарии нет дней для проигрывания")
|
||
return
|
||
self.state.failed = None
|
||
for n, day in enumerate(days):
|
||
start = from_step if n == 0 else 0
|
||
try:
|
||
today = self._now().date()
|
||
except (ApiError, ValueError) as exc:
|
||
self._fail(day.day, max(start, 0), until_day)
|
||
yield StepLog(day.day, -1, "fail", f"часы API недоступны: {exc}")
|
||
return
|
||
day_date: list[date | None] = [planned_date(self.state.dates, day.day, today)]
|
||
if day_date[0] is not None:
|
||
self.state.dates[day.day] = day_date[0]
|
||
yield StepLog(day.day, -1, "info", f"начинаю день {day.day}" + (f": {day.note}" if day.note else ""))
|
||
for i in range(start, len(day.actions)):
|
||
action = day.actions[i]
|
||
self.state.position = Resume(day.day, i, until_day)
|
||
try:
|
||
status, message = self._step(day, action, day_date)
|
||
except Exception as exc: # robustness first: any error pauses playback, never crashes the page
|
||
self._fail(day.day, i, until_day)
|
||
reason = str(exc) or type(exc).__name__
|
||
yield StepLog(day.day, i, "fail", f"{action.describe()}: {reason}", action)
|
||
return
|
||
yield StepLog(day.day, i, status, message, action)
|
||
if self.delay > 0 and i < len(day.actions) - 1:
|
||
self.sleep(self.delay)
|
||
if day.day not in self.state.dates:
|
||
self.state.dates[day.day] = today
|
||
if day.day not in self.state.done:
|
||
self.state.done.append(day.day)
|
||
self.state.position = None
|
||
yield StepLog(day.day, -1, "ok", f"день {day.day} проигран")
|
||
if self.delay > 0 and n < len(days) - 1:
|
||
self.sleep(self.delay)
|
||
|
||
def _fail(self, day: int, step: int, until: int) -> None:
|
||
self.state.failed = Resume(day, step, until)
|
||
self.state.position = None
|
||
|
||
def play_day(self, scenario: Scenario, day: int) -> Iterator[StepLog]:
|
||
return self.run(scenario, day, 0, day)
|
||
|
||
def scroll_to(self, scenario: Scenario, day: int) -> Iterator[StepLog]:
|
||
"""Play every not yet played day up to and including `day`."""
|
||
nxt = self.next_day(scenario)
|
||
if nxt is None or nxt.day > day:
|
||
return iter([StepLog(day, -1, "info", f"дни до {day} уже проиграны")])
|
||
return self.run(scenario, nxt.day, 0, day)
|
||
|
||
def resume(self, scenario: Scenario, *, skip: bool) -> Iterator[StepLog]:
|
||
"""Continue after a failure (or an interrupted run): retry the failed step or skip it."""
|
||
point = self.state.failed or self.state.position
|
||
if point is None:
|
||
return iter([StepLog(0, -1, "info", "нечего продолжать")])
|
||
step = point.step + (1 if skip else 0)
|
||
return self.run(scenario, point.day, step, point.until)
|