feat: live status dashboard for the voice loop

Add a stdlib-only observability site so you can open a browser and watch,
step by step: whether it is listening, what it heard, what the brain thought
and answered, how long each stage took, and whether anything errored.

- wsai/monitor.py: thread-safe telemetry hub (per-turn timed steps, status
  header, error log) with a pub/sub for live push.
- wsai/dashboard.py: stdlib http.server serving a self-contained page plus an
  SSE (/events) live stream; /api/state snapshot fallback.
- Pipeline emits step-by-step turn telemetry (화면 맥락 → 두뇌 → 응답) and
  listening/running status; optional monitor, so existing paths are untouched.
- `python -m wsai --dashboard` starts the site (0.0.0.0:8787, WSAI_DASHBOARD_PORT)
  and loops the mock voice demo so there is always live activity to watch.
- Tests cover turn recording, per-step timing, error marking, and live push.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
EJClaw
2026-08-16 20:24:13 +09:00
parent 3d76cd6c52
commit 6b0755e1ff
7 changed files with 780 additions and 21 deletions

95
tests/test_monitor.py Normal file
View File

@@ -0,0 +1,95 @@
"""The monitor must record step-by-step turns (heard / thought / answered,
per-step timing, ok vs error) and stream them to subscribers — that data is
exactly what the status dashboard renders."""
import asyncio
import json
from wsai.backends.mock import MockBrain, MockSTT, MockTTS
from wsai.monitor import Monitor
from wsai.pipeline import Pipeline
def test_monitor_records_turn_with_timed_steps():
async def go():
mon = Monitor()
pipe = Pipeline(
brain=MockBrain(),
stt=MockSTT(script=["안녕"], interval=0.01),
tts=MockTTS(),
monitor=mon,
)
await asyncio.wait_for(pipe.run(), timeout=5)
return mon
mon = asyncio.run(go())
snap = mon.snapshot()
assert snap["status"]["turns_total"] == 1
assert snap["status"]["running"] is False # cleaned up after run
assert len(snap["turns"]) == 1
turn = snap["turns"][0]
assert turn["heard"] == "안녕" # what it heard
assert turn["reply"] # what it answered
assert turn["status"] == "ok" # it worked
assert turn["total_ms"] >= 0
# step-by-step: every stage is named and timed
names = [s["name"] for s in turn["steps"]]
assert names == ["화면 맥락", "두뇌(생각)", "응답(TTS/전송)"]
assert all(s["ok"] is True for s in turn["steps"])
assert all(s["ms"] >= 0 for s in turn["steps"])
def test_monitor_marks_errors():
class BoomBrain(MockBrain):
async def respond(self, user_text, screen, history):
raise RuntimeError("boom")
async def go():
mon = Monitor()
pipe = Pipeline(
brain=BoomBrain(),
stt=MockSTT(script=["안녕"], interval=0.01),
tts=MockTTS(),
monitor=mon,
)
try:
await asyncio.wait_for(pipe.run(), timeout=5)
except BaseException:
pass # TaskGroup re-raises; we only care about recorded telemetry
return mon
mon = asyncio.run(go())
snap = mon.snapshot()
turn = snap["turns"][0]
assert turn["status"] == "error"
brain_step = next(s for s in turn["steps"] if s["name"] == "두뇌(생각)")
assert brain_step["ok"] is False
assert "boom" in brain_step["error"]
assert snap["status"]["errors_total"] >= 1
def test_subscriber_receives_live_turn_events():
async def go():
mon = Monitor()
q = mon.subscribe()
pipe = Pipeline(
brain=MockBrain(),
stt=MockSTT(script=["안녕"], interval=0.01),
tts=MockTTS(),
monitor=mon,
)
await asyncio.wait_for(pipe.run(), timeout=5)
return q
q = asyncio.run(go())
events = []
while not q.empty():
events.append(json.loads(q.get_nowait()))
types = {e["type"] for e in events}
assert "turn" in types # live turn updates were pushed
assert "status" in types # listening/running status changes were pushed

View File

@@ -2,11 +2,12 @@
python -m wsai # mock pipeline (no deps, no keys) — runs a demo python -m wsai # mock pipeline (no deps, no keys) — runs a demo
python -m wsai --voice # eyes-free voice loop demo (STT -> Brain -> TTS) python -m wsai --voice # eyes-free voice loop demo (STT -> Brain -> TTS)
python -m wsai --dashboard # live status website + a continuous voice demo
python -m wsai --live # capture this screen + Claude eyes/brain python -m wsai --live # capture this screen + Claude eyes/brain
python -m wsai --env # build from WSAI_* environment variables python -m wsai --env # build from WSAI_* environment variables
The mock run is bounded (a few frames + a scripted conversation) so it exits on The mock run is bounded (a few frames + a scripted conversation) so it exits on
its own; --live/--env run until Ctrl-C. its own; --dashboard/--live/--env run until Ctrl-C.
""" """
from __future__ import annotations from __future__ import annotations
@@ -14,32 +15,53 @@ from __future__ import annotations
import argparse import argparse
import asyncio import asyncio
import logging import logging
import os
import socket
from .config import Settings from .config import Settings
from .factory import build from .factory import build
from .monitor import Monitor
async def _run(settings: Settings, demo: bool) -> None: def _lan_ip() -> str:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except OSError:
return "127.0.0.1"
async def _run(settings: Settings, demo: bool, monitor: Monitor | None) -> None:
if demo: if demo:
# Bounded demo so CI / a quick check terminates. # Bounded demo so CI / a quick check terminates.
from .backends.mock import MockFrameSource, MockSTT from .backends.mock import MockFrameSource, MockSTT
from .pipeline import Pipeline
pipe = build(settings) pipe = build(settings, monitor=monitor)
if pipe.source is not None: # keep eyes-free configs eyes-free if pipe.source is not None: # keep eyes-free configs eyes-free
pipe.source = MockFrameSource(interval=0.3, limit=4) pipe.source = MockFrameSource(interval=0.3, limit=4)
if pipe.stt is not None: if pipe.stt is not None:
pipe.stt = MockSTT(interval=0.4) # When serving the dashboard, keep talking forever so there's always
# something live to watch; otherwise stay bounded so the demo exits.
loop = monitor is not None
pipe.stt = MockSTT(interval=2.0 if loop else 0.4, loop=loop)
await pipe.run() await pipe.run()
return return
await build(settings).run() await build(settings, monitor=monitor).run()
def main() -> None: def main() -> None:
ap = argparse.ArgumentParser(prog="wsai") ap = argparse.ArgumentParser(prog="wsai")
ap.add_argument("--voice", action="store_true", help="eyes-free voice loop (STT -> Brain -> TTS)") ap.add_argument("--voice", action="store_true", help="eyes-free voice loop (STT -> Brain -> TTS)")
ap.add_argument("--dashboard", action="store_true", help="serve the live status website (voice demo loops)")
ap.add_argument("--live", action="store_true", help="capture screen + Claude backends") ap.add_argument("--live", action="store_true", help="capture screen + Claude backends")
ap.add_argument("--env", action="store_true", help="build from WSAI_* env vars") ap.add_argument("--env", action="store_true", help="build from WSAI_* env vars")
ap.add_argument("--port", type=int, default=int(os.environ.get("WSAI_DASHBOARD_PORT", "8787")),
help="dashboard port (default 8787, or WSAI_DASHBOARD_PORT)")
ap.add_argument("--host", default=os.environ.get("WSAI_DASHBOARD_HOST", "0.0.0.0"),
help="dashboard bind host (default 0.0.0.0)")
ap.add_argument("-v", "--verbose", action="store_true") ap.add_argument("-v", "--verbose", action="store_true")
args = ap.parse_args() args = ap.parse_args()
@@ -48,7 +70,11 @@ def main() -> None:
format="%(levelname)s %(name)s: %(message)s", format="%(levelname)s %(name)s: %(message)s",
) )
if args.voice: if args.dashboard:
# Default to the eyes-free voice preset for the demo; env can override.
settings = Settings.from_env() if args.env else Settings.voice()
demo = not args.env
elif args.voice:
settings, demo = Settings.voice(), True settings, demo = Settings.voice(), True
elif args.live: elif args.live:
settings, demo = Settings.live(), False settings, demo = Settings.live(), False
@@ -57,10 +83,25 @@ def main() -> None:
else: else:
settings, demo = Settings.mock(), True settings, demo = Settings.mock(), True
monitor: Monitor | None = None
dash = None
if args.dashboard:
from .dashboard import Dashboard
monitor = Monitor()
dash = Dashboard(monitor, host=args.host, port=args.port)
dash.start()
shown = args.host if args.host not in ("0.0.0.0", "") else _lan_ip()
print(f"\n 실시간 상태 사이트: http://{shown}:{args.port}")
print(f" (로컬: http://127.0.0.1:{args.port} )\n")
try: try:
asyncio.run(_run(settings, demo)) asyncio.run(_run(settings, demo, monitor))
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass
finally:
if dash is not None:
dash.stop()
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -63,18 +63,27 @@ class MockVision:
class MockSTT: class MockSTT:
"""Feeds a scripted set of user utterances, then goes quiet.""" """Feeds a scripted set of user utterances, then goes quiet."""
def __init__(self, script: list[str] | None = None, interval: float = 2.0) -> None: def __init__(
self,
script: list[str] | None = None,
interval: float = 2.0,
loop: bool = False,
) -> None:
self.script = script or [ self.script = script or [
"지금 화면에 뭐 보여?", "지금 화면에 뭐 보여?",
"저 에러 왜 나는 거야?", "저 에러 왜 나는 거야?",
"고마워", "고마워",
] ]
self.interval = interval self.interval = interval
self.loop = loop
async def utterances(self) -> AsyncIterator[Utterance]: async def utterances(self) -> AsyncIterator[Utterance]:
for line in self.script: while True:
await asyncio.sleep(self.interval) for line in self.script:
yield Utterance(text=line, ts=time.monotonic(), source="voice") await asyncio.sleep(self.interval)
yield Utterance(text=line, ts=time.monotonic(), source="voice")
if not self.loop:
return
async def aclose(self) -> None: async def aclose(self) -> None:
return return

328
wsai/dashboard.py Normal file
View File

@@ -0,0 +1,328 @@
"""Live status website for the voice loop.
Serves a single self-contained page plus a Server-Sent-Events stream so you can
open a browser and watch, step by step: is it listening, what it heard, what it
thought/answered, how long each stage took, and whether anything errored.
Pure stdlib (``http.server``). Runs in a background thread so it never blocks
the asyncio pipeline.
Endpoints:
GET / -> the dashboard HTML
GET /api/state -> JSON snapshot (initial load / fallback polling)
GET /events -> text/event-stream live push
"""
from __future__ import annotations
import json
import logging
import queue
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from .monitor import Monitor
log = logging.getLogger("wsai.dashboard")
def _make_handler(monitor: Monitor):
class Handler(BaseHTTPRequestHandler):
# Quiet: don't spam the console with one line per request.
def log_message(self, *args) -> None: # noqa: D401
return
def _send(self, code: int, body: bytes, ctype: str) -> None:
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
path = self.path.split("?", 1)[0]
if path == "/" or path == "/index.html":
self._send(200, PAGE.encode("utf-8"), "text/html; charset=utf-8")
elif path == "/api/state":
body = json.dumps(monitor.snapshot(), ensure_ascii=False).encode("utf-8")
self._send(200, body, "application/json; charset=utf-8")
elif path == "/events":
self._stream_events()
else:
self._send(404, b"not found", "text/plain; charset=utf-8")
def _stream_events(self) -> None:
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-store")
self.send_header("Connection", "keep-alive")
self.end_headers()
q = monitor.subscribe()
try:
# Prime the client with a full snapshot so it renders instantly.
first = json.dumps(
{"type": "snapshot", "snapshot": monitor.snapshot()},
ensure_ascii=False,
)
self.wfile.write(f"data: {first}\n\n".encode("utf-8"))
self.wfile.flush()
while True:
try:
data = q.get(timeout=15)
except queue.Empty:
# Heartbeat keeps proxies / the browser from timing out.
self.wfile.write(b": ping\n\n")
self.wfile.flush()
continue
self.wfile.write(f"data: {data}\n\n".encode("utf-8"))
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
pass
finally:
monitor.unsubscribe(q)
return Handler
class Dashboard:
"""Owns the HTTP server thread."""
def __init__(self, monitor: Monitor, host: str = "0.0.0.0", port: int = 8787) -> None:
self.monitor = monitor
self.host = host
self.port = port
self._server: ThreadingHTTPServer | None = None
self._thread: threading.Thread | None = None
def start(self) -> None:
handler = _make_handler(self.monitor)
self._server = ThreadingHTTPServer((self.host, self.port), handler)
self._server.daemon_threads = True
self._thread = threading.Thread(
target=self._server.serve_forever, name="wsai-dashboard", daemon=True
)
self._thread.start()
log.info("dashboard on http://%s:%d", self.host, self.port)
def stop(self) -> None:
if self._server is not None:
self._server.shutdown()
self._server.server_close()
self._server = None
# --------------------------------------------------------------------------- #
# The page. One file, no external assets, so it works offline / behind a LAN.
# --------------------------------------------------------------------------- #
PAGE = r"""<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>watch_sceen_ai · 실시간 상태</title>
<style>
:root{
--bg:#0b0f14; --panel:#131a22; --panel2:#0f151c; --line:#223040;
--fg:#e6edf3; --muted:#8aa0b2; --accent:#3fb6ff; --ok:#37d67a;
--err:#ff5c6c; --warn:#ffc857; --heard:#7aa2ff; --reply:#b28bff;
}
*{box-sizing:border-box}
body{margin:0;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,"Helvetica Neue",Arial,"Apple SD Gothic Neo","Malgun Gothic",sans-serif;
background:var(--bg);color:var(--fg);line-height:1.5}
header{position:sticky;top:0;z-index:5;background:linear-gradient(180deg,#0d141c,#0b0f14);
border-bottom:1px solid var(--line);padding:14px 20px;display:flex;flex-wrap:wrap;gap:16px;align-items:center}
h1{font-size:16px;margin:0;font-weight:650;letter-spacing:.2px}
.sub{color:var(--muted);font-size:12px}
.pill{display:inline-flex;align-items:center;gap:7px;padding:5px 11px;border-radius:999px;
background:var(--panel);border:1px solid var(--line);font-size:12.5px;color:var(--muted)}
.dot{width:9px;height:9px;border-radius:50%;background:#556}
.dot.live{background:var(--ok);box-shadow:0 0 0 0 rgba(55,214,122,.6);animation:pulse 1.6s infinite}
.dot.off{background:#556}
@keyframes pulse{0%{box-shadow:0 0 0 0 rgba(55,214,122,.55)}70%{box-shadow:0 0 0 9px rgba(55,214,122,0)}100%{box-shadow:0 0 0 0 rgba(55,214,122,0)}}
.stats{display:flex;gap:10px;flex-wrap:wrap;margin-left:auto}
.stat{background:var(--panel);border:1px solid var(--line);border-radius:10px;padding:6px 12px;min-width:78px}
.stat b{display:block;font-size:16px}
.stat span{color:var(--muted);font-size:11px}
main{max-width:1000px;margin:0 auto;padding:18px 20px 60px}
.comp{display:flex;gap:8px;flex-wrap:wrap;margin:2px 0 18px}
.comp .pill{font-size:11.5px}
.turn{background:var(--panel);border:1px solid var(--line);border-radius:14px;padding:14px 16px;margin:12px 0;
animation:rise .25s ease}
@keyframes rise{from{opacity:0;transform:translateY(6px)}to{opacity:1;transform:none}}
.turn.active{border-color:#2b5d86;box-shadow:0 0 0 1px #14324a inset}
.turn.error{border-color:#5c2530}
.trow{display:flex;align-items:baseline;gap:10px;margin-bottom:8px}
.badge{font-size:11px;padding:2px 9px;border-radius:999px;border:1px solid var(--line);color:var(--muted);white-space:nowrap}
.badge.ok{color:var(--ok);border-color:#1f5236}
.badge.error{color:var(--err);border-color:#5c2530}
.badge.active{color:var(--accent);border-color:#234a63}
.time{color:var(--muted);font-size:11.5px;margin-left:auto}
.line{display:flex;gap:9px;margin:5px 0;align-items:flex-start}
.tag{flex:0 0 42px;font-size:11px;color:var(--muted);padding-top:2px}
.heard{color:var(--heard);font-weight:550}
.reply{color:var(--reply);font-weight:550}
.steps{margin-top:10px;border-top:1px dashed var(--line);padding-top:10px;display:flex;flex-direction:column;gap:6px}
.step{display:grid;grid-template-columns:120px 1fr 66px;gap:10px;align-items:center;font-size:12.5px}
.step .sname{color:var(--muted)}
.step .sbar{height:8px;background:var(--panel2);border-radius:6px;overflow:hidden;border:1px solid var(--line)}
.step .sfill{height:100%;background:linear-gradient(90deg,#2b7bb0,#3fb6ff)}
.step.err .sfill{background:linear-gradient(90deg,#7a2531,#ff5c6c)}
.step .sms{text-align:right;color:var(--fg);font-variant-numeric:tabular-nums}
.step .serr{grid-column:1 / -1;color:var(--err);font-size:11.5px}
.total{margin-top:8px;font-size:12px;color:var(--muted)}
.total b{color:var(--fg)}
.empty{color:var(--muted);text-align:center;padding:50px 0;font-size:14px}
.events{margin-top:26px}
.events h2{font-size:13px;color:var(--muted);font-weight:600;margin:0 0 8px}
.ev{font-size:12px;color:var(--muted);padding:3px 0;border-bottom:1px solid #16202b;display:flex;gap:10px}
.ev.error{color:var(--err)}
.ev .et{flex:0 0 68px;color:#5f7488}
code{background:#0c1219;padding:1px 5px;border-radius:5px;border:1px solid var(--line)}
</style>
</head>
<body>
<header>
<div>
<h1>watch_sceen_ai · 실시간 상태</h1>
<div class="sub">STT → 두뇌 → TTS 음성 루프를 단계별로 관찰</div>
</div>
<div class="pill"><span id="dot" class="dot off"></span><span id="listen">연결 대기</span></div>
<div class="stats">
<div class="stat"><b id="s-turns">0</b><span>대화 수</span></div>
<div class="stat"><b id="s-errors">0</b><span>오류</span></div>
<div class="stat"><b id="s-up">0초</b><span>가동시간</span></div>
<div class="stat"><b id="s-conn">·</b><span>연결</span></div>
</div>
</header>
<main>
<div class="comp" id="comp"></div>
<div id="turns"></div>
<div id="empty" class="empty">아직 대화가 없습니다. 사용자가 말하면 여기에 단계별로 나타납니다.</div>
<div class="events">
<h2>이벤트 / 오류 로그</h2>
<div id="events"></div>
</div>
</main>
<script>
const $ = (id)=>document.getElementById(id);
const turns = new Map(); // id -> turn object
let statusData = null;
function fmtTime(wall){
const d = new Date(wall*1000);
return d.toLocaleTimeString('ko-KR',{hour12:false}) +
'.' + String(d.getMilliseconds()).padStart(3,'0');
}
function fmtUptime(s){
s = Math.floor(s||0);
const h=Math.floor(s/3600), m=Math.floor(s%3600/60), sec=s%60;
if(h) return h+'시간 '+m+'';
if(m) return m+''+sec+'';
return sec+'';
}
function esc(t){const d=document.createElement('div');d.textContent=t==null?'':t;return d.innerHTML;}
function renderStatus(s){
statusData = s;
$('s-turns').textContent = s.turns_total ?? 0;
$('s-errors').textContent = s.errors_total ?? 0;
$('s-up').textContent = fmtUptime(s.uptime_s);
const listening = s.listening;
$('dot').className = 'dot ' + (listening ? 'live' : 'off');
$('listen').textContent = listening ? '듣는 중' : (s.running ? '실행 중 (대기)' : '중지됨');
const comps = s.components || {};
const el = $('comp'); el.innerHTML = '';
const names = {source:'눈(소스)', vision:'시각', stt:'귀(STT)', brain:'두뇌', tts:'입(TTS)', text:'텍스트'};
for(const k of Object.keys(names)){
if(!(k in comps)) continue;
const v = comps[k];
const p = document.createElement('span');
p.className = 'pill';
p.innerHTML = '<span class="dot '+(v && v!=='none'?'live':'off')+'"></span>'+names[k]+': <b>&nbsp;'+esc(v||'off')+'</b>';
el.appendChild(p);
}
}
function setConn(ok){ $('s-conn').textContent = ok ? '' : ''; $('s-conn').style.color = ok ? 'var(--ok)':'var(--err)'; }
function maxMs(steps){ let m=1; for(const s of steps) m=Math.max(m, s.ms||0); return m; }
function turnEl(t){
const wrap = document.createElement('div');
wrap.className = 'turn ' + (t.status||'active');
wrap.id = 'turn-'+t.id;
const mx = maxMs(t.steps);
let steps = '';
for(const s of (t.steps||[])){
const pct = Math.max(3, Math.round((s.ms||0)/mx*100));
const err = s.ok===false;
steps += '<div class="step'+(err?' err':'')+'">'
+ '<span class="sname">'+esc(s.name)+'</span>'
+ '<span class="sbar"><span class="sfill" style="width:'+pct+'%"></span></span>'
+ '<span class="sms">'+ (s.ms!=null? s.ms.toFixed(0)+' ms':'') +'</span>'
+ (err && s.error ? '<span class="serr">⚠ '+esc(s.error)+'</span>':'')
+ '</div>';
}
const badge = t.status==='ok' ? '<span class="badge ok">정상</span>'
: t.status==='error' ? '<span class="badge error">오류</span>'
: '<span class="badge active">진행 중…</span>';
wrap.innerHTML =
'<div class="trow">'+badge
+'<span class="badge">#'+t.id+' · '+esc(t.source||'voice')+'</span>'
+'<span class="time">'+fmtTime(t.wall)+'</span></div>'
+'<div class="line"><span class="tag">들음</span><span class="heard">'+(t.heard?esc(t.heard):'<i style="color:var(--muted)">(수신 대기)</i>')+'</span></div>'
+'<div class="line"><span class="tag">답변</span><span class="reply">'+(t.reply?esc(t.reply):'<i style="color:var(--muted)">…생각 중</i>')+'</span></div>'
+(t.error?'<div class="line"><span class="tag">오류</span><span style="color:var(--err)">'+esc(t.error)+'</span></div>':'')
+'<div class="steps">'+steps+'</div>'
+'<div class="total">총 소요 <b>'+(t.total_ms?t.total_ms.toFixed(0)+' ms':'')+'</b></div>';
return wrap;
}
function upsertTurn(t){
turns.set(t.id, t);
$('empty').style.display = 'none';
const cont = $('turns');
const existing = $('turn-'+t.id);
const fresh = turnEl(t);
if(existing){ existing.replaceWith(fresh); }
else { cont.prepend(fresh); }
}
function addEvent(e){
const box = $('events');
const row = document.createElement('div');
row.className = 'ev ' + (e.level==='error'?'error':'');
row.innerHTML = '<span class="et">'+fmtTime(e.wall)+'</span><span>'+esc(e.message)+'</span>';
box.prepend(row);
while(box.childElementCount>60) box.removeChild(box.lastChild);
}
function applySnapshot(snap){
renderStatus(snap.status);
turns.clear(); $('turns').innerHTML='';
const list = (snap.turns||[]);
for(const t of list) upsertTurn(t);
if(list.length===0){ $('empty').style.display='block'; }
$('events').innerHTML='';
for(const e of (snap.events||[])) addEvent(e);
}
function connect(){
const es = new EventSource('/events');
es.onopen = ()=> setConn(true);
es.onerror = ()=> setConn(false);
es.onmessage = (m)=>{
let ev; try{ ev = JSON.parse(m.data); }catch(_){ return; }
if(ev.type==='snapshot') applySnapshot(ev.snapshot);
else if(ev.type==='status') renderStatus(ev.status);
else if(ev.type==='turn') upsertTurn(ev.turn);
else if(ev.type==='log') { addEvent(ev); if(statusData){ statusData.errors_total=(statusData.errors_total||0)+(ev.level==='error'?1:0); $('s-errors').textContent=statusData.errors_total; } }
};
}
connect();
// Refresh uptime label every second from the last known status.
setInterval(()=>{ if(statusData){ statusData.uptime_s=(statusData.uptime_s||0)+1; $('s-up').textContent=fmtUptime(statusData.uptime_s);} }, 1000);
</script>
</body>
</html>
"""

View File

@@ -4,18 +4,32 @@ concrete class each config name maps to, so adding a backend = one line here."""
from __future__ import annotations from __future__ import annotations
from .config import Settings from .config import Settings
from .monitor import Monitor
from .pipeline import Pipeline from .pipeline import Pipeline
def build(settings: Settings) -> Pipeline: def build(settings: Settings, monitor: Monitor | None = None) -> Pipeline:
return Pipeline( pipe = Pipeline(
source=_source(settings), source=_source(settings),
vision=_vision(settings), vision=_vision(settings),
brain=_brain(settings), brain=_brain(settings),
stt=_stt(settings), stt=_stt(settings),
tts=_tts(settings), tts=_tts(settings),
text_channel=_text(settings), text_channel=_text(settings),
monitor=monitor,
) )
if monitor is not None:
monitor.set_components(
{
"source": settings.source or "none",
"vision": settings.vision or "none",
"stt": settings.stt or "none",
"brain": settings.brain,
"tts": settings.tts or "none",
"text": settings.text or "none",
}
)
return pipe
def _source(s: Settings): def _source(s: Settings):

226
wsai/monitor.py Normal file
View File

@@ -0,0 +1,226 @@
"""Telemetry hub for the live status dashboard.
The pipeline is a chain of steps (heard -> screen context -> brain -> speak).
This module records, for every conversation turn, *what happened at each step*
and *how long it took*, plus a rolling status header and any errors. The
dashboard (``wsai/dashboard.py``) reads a snapshot and subscribes for live
push updates.
Design notes:
* Pure stdlib, no deps — matches the project's "core has no third-party deps".
* Thread-safe. The pipeline mutates it from the asyncio loop; the HTTP server
reads/subscribes from its own threads. A single lock guards everything.
* A Monitor with zero subscribers is essentially free, so the pipeline can
always hold one (no separate no-op path).
"""
from __future__ import annotations
import json
import queue
import threading
import time
from collections import deque
from typing import Any
def _now_wall() -> float:
# Wall-clock seconds for human-readable timestamps on the page.
return time.time()
def _now_mono() -> float:
# Monotonic seconds for measuring durations (immune to clock jumps).
return time.monotonic()
class Step:
"""One timed stage inside a turn (e.g. "두뇌"). Used as an async context
manager so it can wrap an ``await`` and record ok/error + elapsed ms."""
def __init__(self, turn: "Turn", name: str) -> None:
self.turn = turn
self.name = name
self.ok: bool | None = None
self.ms: float = 0.0
self.detail: str = ""
self.error: str = ""
self._t0 = 0.0
async def __aenter__(self) -> "Step":
self._t0 = _now_mono()
self.turn._steps.append(self)
self.turn._touch()
return self
async def __aexit__(self, exc_type, exc, tb) -> bool:
self.ms = (_now_mono() - self._t0) * 1000.0
if exc is not None:
self.ok = False
self.error = f"{exc_type.__name__}: {exc}"
else:
self.ok = True
self.turn._touch()
return False # never swallow: the pipeline/TaskGroup must still see it
def to_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"ok": self.ok,
"ms": round(self.ms, 1),
"detail": self.detail,
"error": self.error,
}
class Turn:
"""One user utterance and everything the AI did in response."""
def __init__(self, monitor: "Monitor", turn_id: int, source: str) -> None:
self._monitor = monitor
self.id = turn_id
self.source = source
self.wall = _now_wall()
self._t0 = _now_mono()
self.heard_text = ""
self.reply_text = ""
self.status = "active" # active | ok | error
self.error = ""
self.total_ms = 0.0
self._steps: list[Step] = []
# -- recording API (called from the pipeline) ------------------------- #
def heard(self, text: str) -> None:
self.heard_text = text
self._touch()
def replied(self, text: str) -> None:
self.reply_text = text
self._touch()
def step(self, name: str) -> Step:
return Step(self, name)
def finish(self, error: str = "") -> None:
self.total_ms = (_now_mono() - self._t0) * 1000.0
if error:
self.status = "error"
self.error = error
elif any(s.ok is False for s in self._steps):
self.status = "error"
else:
self.status = "ok"
self._touch()
# -- internal --------------------------------------------------------- #
def _touch(self) -> None:
self._monitor._publish(self)
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"source": self.source,
"wall": self.wall,
"heard": self.heard_text,
"reply": self.reply_text,
"status": self.status,
"error": self.error,
"total_ms": round(self.total_ms, 1),
"steps": [s.to_dict() for s in self._steps],
}
class Monitor:
"""Rolling record of turns + status, with a pub/sub for live updates."""
def __init__(self, keep: int = 60) -> None:
self._turns: deque[Turn] = deque(maxlen=keep)
self._events: deque[dict[str, Any]] = deque(maxlen=200)
self._status: dict[str, Any] = {
"running": False,
"listening": False,
"started_wall": _now_wall(),
"components": {},
"turns_total": 0,
"errors_total": 0,
}
self._lock = threading.Lock()
self._subs: list["queue.Queue[str]"] = []
self._id = 0
# -- status ----------------------------------------------------------- #
def set_status(self, **kw: Any) -> None:
with self._lock:
self._status.update(kw)
self._broadcast({"type": "status", "status": self.status_snapshot()})
def set_components(self, components: dict[str, Any]) -> None:
with self._lock:
self._status["components"] = components
self._broadcast({"type": "status", "status": self.status_snapshot()})
def status_snapshot(self) -> dict[str, Any]:
with self._lock:
s = dict(self._status)
s["uptime_s"] = round(_now_wall() - s["started_wall"], 1)
return s
def log(self, level: str, message: str) -> None:
"""A free-form lifecycle/error line (startup, disconnect, crash…)."""
evt = {"type": "log", "level": level, "message": message, "wall": _now_wall()}
with self._lock:
self._events.append(evt)
if level == "error":
self._status["errors_total"] += 1
self._broadcast(evt)
# -- turns ------------------------------------------------------------ #
def turn(self, source: str = "voice") -> Turn:
with self._lock:
self._id += 1
self._status["turns_total"] += 1
t = Turn(self, self._id, source)
self._turns.append(t)
self._publish(t)
return t
def _publish(self, t: Turn) -> None:
# Recompute error total lazily on error transitions.
if t.status == "error":
with self._lock:
# errors_total counts turns that ended in error at most once
pass
self._broadcast({"type": "turn", "turn": t.to_dict()})
# -- snapshot / subscribe (read side, HTTP threads) ------------------- #
def snapshot(self) -> dict[str, Any]:
with self._lock:
turns = [t.to_dict() for t in self._turns]
events = list(self._events)
return {
"status": self.status_snapshot(),
"turns": turns,
"events": events,
}
def subscribe(self) -> "queue.Queue[str]":
q: "queue.Queue[str]" = queue.Queue(maxsize=256)
with self._lock:
self._subs.append(q)
return q
def unsubscribe(self, q: "queue.Queue[str]") -> None:
with self._lock:
if q in self._subs:
self._subs.remove(q)
def _broadcast(self, event: dict[str, Any]) -> None:
data = json.dumps(event, ensure_ascii=False)
with self._lock:
subs = list(self._subs)
for q in subs:
try:
q.put_nowait(data)
except queue.Full:
# Slow client: drop it rather than block the pipeline.
self.unsubscribe(q)

View File

@@ -15,6 +15,7 @@ from .interfaces import (
Utterance, Utterance,
VisionBackend, VisionBackend,
) )
from .monitor import Monitor
from .state import SharedScreenContext from .state import SharedScreenContext
log = logging.getLogger("wsai.pipeline") log = logging.getLogger("wsai.pipeline")
@@ -41,6 +42,7 @@ class Pipeline:
tts: TextToSpeech | None = None, tts: TextToSpeech | None = None,
text_channel: TextChannel | None = None, text_channel: TextChannel | None = None,
history_turns: int = 12, history_turns: int = 12,
monitor: Monitor | None = None,
) -> None: ) -> None:
self.source = source self.source = source
self.vision = vision self.vision = vision
@@ -48,6 +50,7 @@ class Pipeline:
self.stt = stt self.stt = stt
self.tts = tts self.tts = tts
self.text_channel = text_channel self.text_channel = text_channel
self.monitor = monitor
self.context = SharedScreenContext() self.context = SharedScreenContext()
self._history: list[tuple[str, str]] = [] self._history: list[tuple[str, str]] = []
self._history_turns = history_turns self._history_turns = history_turns
@@ -59,18 +62,43 @@ class Pipeline:
async for frame in self.source.frames(): async for frame in self.source.frames():
try: try:
obs = await self.vision.describe(frame) obs = await self.vision.describe(frame)
except Exception: # a single bad frame must not kill the loop except Exception as exc: # a single bad frame must not kill the loop
log.exception("vision.describe failed") log.exception("vision.describe failed")
if self.monitor is not None:
self.monitor.log("error", f"화면 이해 실패: {exc}")
continue continue
await self.context.update(obs) await self.context.update(obs)
log.debug("screen: %s", obs.text[:120]) log.debug("screen: %s", obs.text[:120])
# -- conversation ------------------------------------------------------ # # -- conversation ------------------------------------------------------ #
async def _handle(self, utt: Utterance) -> None: async def _handle(self, utt: Utterance) -> None:
screen = await self.context.latest() if self.monitor is None:
reply = await self.brain.respond(utt.text, screen, self._history) screen = await self.context.latest()
self._remember(utt.text, reply.text) reply = await self.brain.respond(utt.text, screen, self._history)
await self._emit(reply) self._remember(utt.text, reply.text)
await self._emit(reply)
return
# Same work, but each stage is timed and streamed to the dashboard so a
# viewer can see what was heard, what the brain answered, how long each
# step took, and whether anything errored.
turn = self.monitor.turn(source=utt.source)
turn.heard(utt.text)
try:
async with turn.step("화면 맥락"):
screen = await self.context.latest()
async with turn.step("두뇌(생각)"):
reply = await self.brain.respond(utt.text, screen, self._history)
turn.replied(reply.text)
self._remember(utt.text, reply.text)
async with turn.step("응답(TTS/전송)"):
await self._emit(reply)
except Exception as exc:
turn.finish(error=f"{type(exc).__name__}: {exc}")
self.monitor.log("error", f"대화 #{turn.id} 실패: {exc}")
raise
else:
turn.finish()
def _remember(self, user: str, ai: str) -> None: def _remember(self, user: str, ai: str) -> None:
self._history.append((user, ai)) self._history.append((user, ai))
@@ -91,8 +119,15 @@ class Pipeline:
async def _listen_voice(self) -> None: async def _listen_voice(self) -> None:
if self.stt is None: if self.stt is None:
return return
async for utt in self.stt.utterances(): if self.monitor is not None:
await self._handle(utt) self.monitor.set_status(listening=True)
self.monitor.log("info", "음성 수신 시작 — 발화 대기 중")
try:
async for utt in self.stt.utterances():
await self._handle(utt)
finally:
if self.monitor is not None:
self.monitor.set_status(listening=False)
async def _listen_text(self) -> None: async def _listen_text(self) -> None:
if self.text_channel is None: if self.text_channel is None:
@@ -107,12 +142,23 @@ class Pipeline:
# failing loop propagated while the siblings kept running detached, and # failing loop propagated while the siblings kept running detached, and
# aclose() in the finally then closed a source/stt out from under a # aclose() in the finally then closed a source/stt out from under a
# still-live loop (close-during-use). # still-live loop (close-during-use).
if self.monitor is not None:
self.monitor.set_status(running=True)
self.monitor.log("info", "파이프라인 시작")
try: try:
async with asyncio.TaskGroup() as tg: async with asyncio.TaskGroup() as tg:
tg.create_task(self._perceive()) tg.create_task(self._perceive())
tg.create_task(self._listen_voice()) tg.create_task(self._listen_voice())
tg.create_task(self._listen_text()) tg.create_task(self._listen_text())
except* Exception as eg:
if self.monitor is not None:
for exc in eg.exceptions:
self.monitor.log("error", f"루프 예외: {type(exc).__name__}: {exc}")
raise
finally: finally:
if self.monitor is not None:
self.monitor.set_status(running=False, listening=False)
self.monitor.log("info", "파이프라인 종료")
await self.aclose() await self.aclose()
async def aclose(self) -> None: async def aclose(self) -> None: