6b0755e1ff
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>
109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
"""Mock backends. These let the full pipeline run with no GPU, no mic, no API
|
|
key — so the skeleton is verifiable and gives every real backend a reference
|
|
implementation to match.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import itertools
|
|
import time
|
|
from typing import AsyncIterator
|
|
|
|
from ..interfaces import (
|
|
Frame,
|
|
Reply,
|
|
ScreenObservation,
|
|
Utterance,
|
|
)
|
|
|
|
|
|
class MockFrameSource:
|
|
"""Emits tiny synthetic frames on a fixed interval."""
|
|
|
|
def __init__(self, interval: float = 1.0, limit: int | None = None) -> None:
|
|
self.interval = interval
|
|
self.limit = limit
|
|
|
|
async def frames(self) -> AsyncIterator[Frame]:
|
|
for i in itertools.count():
|
|
if self.limit is not None and i >= self.limit:
|
|
return
|
|
yield Frame(
|
|
data=b"\x89PNG\r\n\x1a\n", # PNG magic; enough for a stub
|
|
width=1280,
|
|
height=720,
|
|
ts=time.monotonic(),
|
|
mime="image/png",
|
|
)
|
|
await asyncio.sleep(self.interval)
|
|
|
|
async def aclose(self) -> None: # nothing to release
|
|
return
|
|
|
|
|
|
class MockVision:
|
|
"""Pretends to read the screen. Cycles through a few canned scenes."""
|
|
|
|
SCENES = [
|
|
"VS Code is open with a Python file; a traceback is visible in the terminal.",
|
|
"A browser shows a GitHub pull request diff.",
|
|
"A game is running; the player is in a menu screen.",
|
|
]
|
|
|
|
def __init__(self) -> None:
|
|
self._i = 0
|
|
|
|
async def describe(self, frame: Frame, hint: str | None = None) -> ScreenObservation:
|
|
scene = self.SCENES[self._i % len(self.SCENES)]
|
|
self._i += 1
|
|
return ScreenObservation(text=scene, ts=frame.ts)
|
|
|
|
|
|
class MockSTT:
|
|
"""Feeds a scripted set of user utterances, then goes quiet."""
|
|
|
|
def __init__(
|
|
self,
|
|
script: list[str] | None = None,
|
|
interval: float = 2.0,
|
|
loop: bool = False,
|
|
) -> None:
|
|
self.script = script or [
|
|
"지금 화면에 뭐 보여?",
|
|
"저 에러 왜 나는 거야?",
|
|
"고마워",
|
|
]
|
|
self.interval = interval
|
|
self.loop = loop
|
|
|
|
async def utterances(self) -> AsyncIterator[Utterance]:
|
|
while True:
|
|
for line in self.script:
|
|
await asyncio.sleep(self.interval)
|
|
yield Utterance(text=line, ts=time.monotonic(), source="voice")
|
|
if not self.loop:
|
|
return
|
|
|
|
async def aclose(self) -> None:
|
|
return
|
|
|
|
|
|
class MockTTS:
|
|
"""'Speaks' by printing. Real TTS swaps in here."""
|
|
|
|
async def speak(self, reply: Reply) -> None:
|
|
print(f"[TTS] {reply.text}")
|
|
|
|
|
|
class MockBrain:
|
|
"""Echo-style brain that references the current screen, so you can see the
|
|
screen context actually reaching the conversation loop."""
|
|
|
|
async def respond(self, user_text, screen, history) -> Reply:
|
|
seen = screen.text if screen else "아직 화면을 못 읽었어요"
|
|
return Reply(
|
|
text=f'(화면: "{seen}") 라고 봤어요. 말씀하신 "{user_text}"에 대해 답하자면… [mock]',
|
|
ts=time.monotonic(),
|
|
)
|