feat: scaffold watch-screen AI pipeline (mock-runnable skeleton)
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).
This commit is contained in:
99
wsai/backends/mock.py
Normal file
99
wsai/backends/mock.py
Normal file
@@ -0,0 +1,99 @@
|
||||
"""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(),
|
||||
)
|
||||
Reference in New Issue
Block a user