4eeddc4b1f
Modular async pipeline: FrameSource->Vision->context and STT/text->Brain->TTS. All stages are Protocols; mock backends run end-to-end with no deps/keys. Real backends included: mss screen capture, Claude vision+brain (guarded imports).
100 lines
3.0 KiB
Python
100 lines
3.0 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) -> None:
|
|
self.script = script or [
|
|
"지금 화면에 뭐 보여?",
|
|
"저 에러 왜 나는 거야?",
|
|
"고마워",
|
|
]
|
|
self.interval = interval
|
|
|
|
async def utterances(self) -> AsyncIterator[Utterance]:
|
|
for line in self.script:
|
|
await asyncio.sleep(self.interval)
|
|
yield Utterance(text=line, ts=time.monotonic(), source="voice")
|
|
|
|
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(),
|
|
)
|