"""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 _speech_text(reply: str) -> str:
"""Return reply text exactly as authored for the TTS backend.
The TTS backend itself understands bracketed emotion tags: known emotion
tags steer delivery and are not spoken; non-emotion brackets are spoken.
Do not rewrite a leading tag here, or the first emotion would be read aloud
and lost before ``MeloTTS.synth`` can parse it."""
return reply
def _thought_summary(reply: str) -> str:
"""A short '생각내용' line: the emotion/tone plan the bot chose for delivery,
derived from the [감정] tags in its reply. This is the AI's decision about
*how* to say the answer, shown between 들음 and 답변."""
import re
from .backends.emotion import match_emotion
tags = re.findall(r"\[([^\[\]]*)\]", reply or "")
emotions = [t.strip() for t in tags if match_emotion(t)]
if emotions:
return "감정 톤: " + " → ".join(emotions)
return "감정 태그 없음 · 기본 톤으로 답변"
def _make_handler(dash: "Dashboard"):
monitor = dash.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 == "/api/prompt":
self._handle_prompt_get()
elif path == "/api/bot/state":
self._send_json(dash.bot.state())
elif path == "/api/bot/commands":
self._send_json({"commands": dash.bot.drain()})
elif path == "/api/bot/lists":
import urllib.parse as _up
q = _up.parse_qs(self.path.split("?", 1)[1] if "?" in self.path else "")
gid = (q.get("guildId", [""])[0])
self._send_json({"ok": True, "guildId": gid, "lists": dash.bot.get_lists(gid)})
elif path == "/events":
self._stream_events()
else:
self._send(404, b"not found", "text/plain; charset=utf-8")
def do_POST(self) -> None: # noqa: N802
path = self.path.split("?", 1)[0]
if path == "/api/stt":
self._handle_stt()
elif path == "/api/voice-turn":
self._handle_voice_turn()
elif path == "/api/prompt":
self._handle_prompt_post()
elif path == "/api/logs/clear":
monitor.clear_events()
self._send(200, json.dumps({"ok": True}).encode(), "application/json; charset=utf-8")
elif path == "/api/logs/delete":
self._handle_log_mutate("delete")
elif path == "/api/logs/edit":
self._handle_log_mutate("edit")
elif path == "/api/bot/report":
self._handle_bot_report()
elif path == "/api/bot/select":
self._handle_bot_select()
elif path == "/api/bot/lists":
self._handle_bot_lists()
else:
self._send(404, b"not found", "text/plain; charset=utf-8")
def _read_body(self) -> bytes:
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
length = 0
return self.rfile.read(length) if length > 0 else b""
def _handle_voice_turn(self) -> None:
"""Discord voice bridge: utterance wav in -> reply wav out. The
recognised/reply text ride along as URL-encoded response headers so
the bot can log them; the body is the reply audio to play back."""
import urllib.parse
if dash.stt is None or dash.tts is None:
self._send(503, json.dumps({"ok": False, "error": "voice loop not enabled"}).encode(),
"application/json; charset=utf-8")
return
raw = self._read_body()
if not raw:
self._send(400, json.dumps({"ok": False, "error": "empty upload"}).encode(),
"application/json; charset=utf-8")
return
speaker = urllib.parse.unquote(self.headers.get("X-User-Name", "") or "")
guild = urllib.parse.unquote(self.headers.get("X-Guild-Name", "") or "")
channel = urllib.parse.unquote(self.headers.get("X-Channel-Name", "") or "")
try:
res = dash.voice_turn(raw, speaker=speaker, guild=guild, channel=channel)
except Exception as exc: # noqa: BLE001
log.exception("voice-turn failed")
self._send(500, json.dumps({"ok": False, "error": f"{type(exc).__name__}: {exc}"},
ensure_ascii=False).encode(), "application/json; charset=utf-8")
return
body = res["wav"]
self.send_response(200 if body else 204)
self.send_header("Content-Type", "audio/wav")
self.send_header("Content-Length", str(len(body)))
self.send_header("X-Heard", urllib.parse.quote(res.get("heard", "")))
self.send_header("X-Reply", urllib.parse.quote(res.get("reply", "")))
self.send_header("Cache-Control", "no-store")
self.end_headers()
if body:
self.wfile.write(body)
def _handle_stt(self) -> None:
"""Accept an uploaded audio blob (mic recording or file), run it
through the real GPU STT, and return the recognised text."""
if dash.stt is None:
self._send(503, json.dumps({"ok": False, "error": "STT not enabled"}).encode(),
"application/json; charset=utf-8")
return
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
length = 0
if length <= 0:
self._send(400, json.dumps({"ok": False, "error": "empty upload"}).encode(),
"application/json; charset=utf-8")
return
raw = self.rfile.read(length)
try:
result = dash.transcribe_upload(raw)
body = json.dumps({"ok": True, **result}, ensure_ascii=False).encode("utf-8")
self._send(200, body, "application/json; charset=utf-8")
except Exception as exc: # noqa: BLE001 — surface the reason to the page
log.exception("STT upload failed")
body = json.dumps({"ok": False, "error": f"{type(exc).__name__}: {exc}"},
ensure_ascii=False).encode("utf-8")
self._send(500, body, "application/json; charset=utf-8")
def _default_persona(self) -> str:
# The built-in seed prompt, used when no override is saved. Imported
# lazily so the dashboard has no hard dependency on the Claude backend.
try:
from .backends.claude import ClaudeBrain
return ClaudeBrain.PERSONA
except Exception:
return ""
def _handle_prompt_get(self) -> None:
"""Return the live bot system prompt so the page can show/edit it."""
from . import prompt_store
default = self._default_persona()
body = json.dumps({
"ok": True,
"prompt": prompt_store.get_persona(default),
"default": default,
"overridden": prompt_store.is_overridden(),
}, ensure_ascii=False).encode("utf-8")
self._send(200, body, "application/json; charset=utf-8")
def _handle_prompt_post(self) -> None:
"""Save an edited system prompt; takes effect on the next reply. An
empty prompt clears the override and reverts to the built-in default."""
from . import prompt_store
raw = self._read_body()
try:
data = json.loads(raw.decode("utf-8")) if raw else {}
prompt = data.get("prompt", "")
except (ValueError, AttributeError):
self._send(400, json.dumps({"ok": False, "error": "invalid JSON"}).encode(),
"application/json; charset=utf-8")
return
prompt_store.set_persona(prompt)
monitor.log("info", "봇 프롬프트가 수정되었습니다" if prompt.strip()
else "봇 프롬프트가 기본값으로 초기화되었습니다")
body = json.dumps({
"ok": True,
"prompt": prompt_store.get_persona(self._default_persona()),
"overridden": prompt_store.is_overridden(),
}, ensure_ascii=False).encode("utf-8")
self._send(200, body, "application/json; charset=utf-8")
def _send_json(self, obj, code: int = 200) -> None:
self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"),
"application/json; charset=utf-8")
def _handle_bot_report(self) -> None:
"""The Discord bot pushes its live state (identity, guilds, voice
channels, current channel + members, list settings)."""
raw = self._read_body()
try:
data = json.loads(raw.decode("utf-8")) if raw else {}
except (ValueError, AttributeError):
self._send_json({"ok": False, "error": "invalid JSON"}, 400)
return
dash.bot.report(data)
# Hand the bot any queued commands + the current listen filters in the
# same round trip so it does not have to poll extra endpoints.
self._send_json({"ok": True, "commands": dash.bot.drain(),
"lists": dash.bot.all_lists()})
def _handle_bot_select(self) -> None:
"""UI picked a server/voice channel → queue a join (or leave) command."""
raw = self._read_body()
try:
data = json.loads(raw.decode("utf-8")) if raw else {}
except (ValueError, AttributeError):
self._send_json({"ok": False, "error": "invalid JSON"}, 400)
return
guild_id = (data.get("guildId") or "").strip()
channel_id = (data.get("channelId") or "").strip()
if channel_id and guild_id:
cid = dash.bot.enqueue({"type": "join", "guildId": guild_id, "channelId": channel_id})
monitor.log("info", f"음성채널 참여 요청 (guild={guild_id} channel={channel_id})")
else:
cid = dash.bot.enqueue({"type": "leave"})
monitor.log("info", "음성채널 나가기 요청")
self._send_json({"ok": True, "commandId": cid})
def _handle_bot_lists(self) -> None:
"""Save the whitelist/blacklist (users + roles) for a guild."""
raw = self._read_body()
try:
data = json.loads(raw.decode("utf-8")) if raw else {}
guild_id = (data.get("guildId") or "").strip()
if not guild_id:
raise ValueError("guildId required")
except (ValueError, AttributeError) as exc:
self._send_json({"ok": False, "error": str(exc)}, 400)
return
saved = dash.bot.set_lists(guild_id, data.get("lists") or {})
monitor.log("info", f"청취 화이트/블랙리스트 업데이트 (guild={guild_id})")
self._send_json({"ok": True, "guildId": guild_id, "lists": saved})
def _handle_log_mutate(self, action: str) -> None:
"""Per-line log delete/edit by event id."""
raw = self._read_body()
try:
data = json.loads(raw.decode("utf-8")) if raw else {}
event_id = int(data.get("id"))
except (ValueError, TypeError, AttributeError):
self._send(400, json.dumps({"ok": False, "error": "id required"}).encode(),
"application/json; charset=utf-8")
return
if action == "delete":
ok = monitor.delete_event(event_id)
else:
ok = monitor.edit_event(event_id, str(data.get("message", "")))
self._send(200 if ok else 404,
json.dumps({"ok": ok}).encode(), "application/json; 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.
Optionally holds a real STT backend so the page can offer a live
recognition test (upload/record audio -> GPU whisper -> text). The STT
backend is async, so the dashboard runs its own asyncio loop in a
background thread and bridges the synchronous HTTP handlers onto it.
"""
def __init__(self, monitor: Monitor, host: str = "0.0.0.0", port: int = 8787,
stt=None, tts=None, brain=None, history_turns: int = 12) -> None:
from .bot_control import BotControl
self.monitor = monitor
self.host = host
self.port = port
self.stt = stt
self.tts = tts
self.brain = brain
self.bot = BotControl() # dashboard <-> Discord bot control plane
self._history: list[tuple[str, str]] = []
self._history_turns = history_turns
self._server: ThreadingHTTPServer | None = None
self._thread: threading.Thread | None = None
self._loop = None
self._loop_thread: threading.Thread | None = None
def start(self) -> None:
if self.stt is not None or self.tts is not None:
self._start_loop()
handler = _make_handler(self)
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
if self._loop is not None:
self._loop.call_soon_threadsafe(self._loop.stop)
self._loop = None
# -- async bridge (STT test) ----------------------------------------- #
def _start_loop(self) -> None:
import asyncio
self._loop = asyncio.new_event_loop()
self._loop_thread = threading.Thread(
target=self._loop.run_forever, name="wsai-dashboard-loop", daemon=True
)
self._loop_thread.start()
def _submit(self, coro, timeout: float = 120.0):
import asyncio
fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
return fut.result(timeout=timeout)
def warm(self) -> None:
"""Pre-start the STT/TTS workers (loads + warms the GPU) so the first
recognition/synth is instant instead of paying model-load + CUDA autotune."""
if self.stt is not None:
self._submit(self.stt._ensure())
if self.tts is not None:
self._submit(self.tts._ensure())
def voice_turn(self, audio_bytes: bytes, speaker: str = "",
guild: str = "", channel: str = "") -> dict:
"""One Discord voice turn: decode the uploaded utterance, recognise it
on the GPU, think of a reply (Claude brain if wired, else echo),
synthesise it on the GPU, and return {heard, reply, wav} where wav is the
reply audio bytes for the bot to play back into the channel."""
import os
import subprocess
import time
import uuid
if self.stt is None or self.tts is None:
raise RuntimeError("voice_turn needs both STT and TTS")
updir = os.path.expanduser("~/.cache/wsai/uploads")
os.makedirs(updir, exist_ok=True)
stem = os.path.join(updir, uuid.uuid4().hex)
src, wav = stem + ".bin", stem + ".wav"
with open(src, "wb") as f:
f.write(audio_bytes)
turn = self.monitor.turn(source="discord")
if speaker:
turn.speaker = speaker # who spoke (for the "누가 말했는지" log)
turn.guild, turn.channel = guild, channel # for 서버별/채널별 필터
t0 = time.monotonic()
try:
subprocess.run(
["ffmpeg", "-y", "-i", src, "-ar", "16000", "-ac", "1", wav],
check=True, capture_output=True,
)
heard = (self._submit(self.stt.transcribe(wav)) or "").strip()
turn.heard(heard or "(빈 결과)")
if not heard:
# Nothing recognised (silence/noise): mark it as [잡음] and skip
# the brain/TTS so the bot plays nothing back.
turn.thought("목소리가 아닌 잡음으로 판단 → 응답하지 않음")
turn.replied("[잡음]")
turn.finish()
return {"heard": heard, "reply": "[잡음]", "wav": b""}
reply_text = self._think(heard)
turn.thought(_thought_summary(reply_text))
turn.replied(reply_text)
out_path = self._submit(self.tts.synth(_speech_text(reply_text)))
with open(out_path, "rb") as f:
reply_wav = f.read()
ms = int((time.monotonic() - t0) * 1000)
step = turn.step("STT+두뇌+TTS" if self.brain else "STT+TTS(GPU)")
step.ok, step.ms = True, float(ms)
turn._steps.append(step)
turn.finish()
try:
os.remove(out_path)
except OSError:
pass
return {"heard": heard, "reply": reply_text, "wav": reply_wav}
except subprocess.CalledProcessError as exc:
turn.finish(error="ffmpeg decode failed")
err = exc.stderr.decode("utf-8", "replace")[-300:] if exc.stderr else str(exc)
raise RuntimeError(f"ffmpeg: {err}") from exc
except Exception as exc:
turn.finish(error=str(exc))
raise
finally:
for p in (src, wav):
try:
os.remove(p)
except OSError:
pass
def _think(self, heard: str) -> str:
"""Turn what was heard into a reply. Uses the Claude brain when wired
(with rolling conversation history); falls back to echo if there is no
brain, and to a spoken apology if the brain call fails — so one API hiccup
never kills the voice loop."""
if self.brain is None:
return heard # echo mode
try:
reply = self._submit(self.brain.respond(heard, None, list(self._history)))
text = (reply.text or "").strip()
u = getattr(reply, "usage", None)
if u:
self.monitor.add_claude_usage(u.get("input", 0), u.get("output", 0))
except Exception as exc: # noqa: BLE001
log.exception("brain failed")
self.monitor.log("error", f"두뇌 응답 실패: {exc}")
blob = f"{getattr(exc, 'status_code', '')} {exc}".lower()
if "529" in blob or "overload" in blob:
# Transient server overload survived the SDK retries.
return "지금 서버가 잠깐 붐벼서 생각이 늦네. 잠시 뒤에 다시 말해줄래?"
return "미안, 지금 잠깐 생각이 안 났어. 다시 말해줄래?"
if not text:
return "음, 뭐라고 해야 할지 모르겠어. 다시 말해줄래?"
self._history.append((heard, text))
if len(self._history) > self._history_turns:
self._history = self._history[-self._history_turns:]
return text
def transcribe_upload(self, audio_bytes: bytes) -> dict:
"""ffmpeg-normalise an uploaded blob to 16 kHz mono wav, transcribe it
on the GPU, and record the result as a monitor turn so it also shows in
the live feed. Returns {text, ms, device}."""
import os
import subprocess
import tempfile
import time
import uuid
updir = os.path.expanduser("~/.cache/wsai/uploads")
os.makedirs(updir, exist_ok=True)
stem = os.path.join(updir, uuid.uuid4().hex)
src, wav = stem + ".bin", stem + ".wav"
with open(src, "wb") as f:
f.write(audio_bytes)
turn = self.monitor.turn(source="web")
t0 = time.monotonic()
try:
# Decode whatever the browser sent (webm/opus, ogg, mp4, wav) to the
# 16 kHz mono wav faster-whisper expects.
subprocess.run(
["ffmpeg", "-y", "-i", src, "-ar", "16000", "-ac", "1", wav],
check=True, capture_output=True,
)
text = self._submit(self.stt.transcribe(wav))
ms = int((time.monotonic() - t0) * 1000)
turn.heard(text or "(빈 결과)")
step = turn.step("STT(GPU)")
step.ok, step.ms = True, float(ms)
turn._steps.append(step)
turn.finish()
return {"text": text, "ms": ms,
"device": getattr(self.stt, "resolved_device", None) or "?"}
except subprocess.CalledProcessError as exc:
turn.finish(error="ffmpeg decode failed")
err = exc.stderr.decode("utf-8", "replace")[-300:] if exc.stderr else str(exc)
raise RuntimeError(f"ffmpeg: {err}") from exc
except Exception as exc:
turn.finish(error=str(exc))
raise
finally:
for p in (src, wav):
try:
os.remove(p)
except OSError:
pass
# --------------------------------------------------------------------------- #
# The page. One file, no external assets, so it works offline / behind a LAN.
# --------------------------------------------------------------------------- #
PAGE = r"""
watch_sceen_ai · 실시간 상태