Initial scaffold: ESP32-P4 firmware + voice gateway for Tatlock endpoint

DeskLock gives the Tatlock butler a face and voice on a Waveshare
ESP32-P4-WIFI6-Touch-LCD-3.4C round display in the living room.

- firmware/: ESP-IDF project targeting esp32p4 with the Waveshare XC BSP
- gateway/: FastAPI voice bridge (faster-whisper STT, Tatlock chat, Piper TTS)
- docs/architecture.md: component design and device<->gateway WS protocol

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

gateway/src/desklock_gateway/config.py

@@ -0,0 +1,17 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """Gateway configuration, overridable via DESKLOCK_* environment variables."""
+
+    tatlock_base_url: str = "http://tatlock.schweitz.internal:8000"
+    tatlock_model: str = "Tatlock"
+    sample_rate: int = 16000
+    stt_model: str = "small"
+    stt_device: str = "cuda"
+    tts_voice: str = "en_GB-alan-medium"
+
+    model_config = {"env_prefix": "DESKLOCK_"}
+
+
+settings = Settings()

gateway/src/desklock_gateway/main.py

@@ -0,0 +1,77 @@
+"""DeskLock gateway: WebSocket voice bridge between the device and Tatlock.
+
+Protocol (see docs/architecture.md — keep in sync):
+  device -> gateway : {"type": "utterance_start"} + binary PCM + {"type": "utterance_end"}
+  gateway -> device : state / transcript / reply_text events, then audio_start + PCM + audio_end
+"""
+
+import asyncio
+import logging
+
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+
+from . import stt, tts
+from .config import settings
+from .tatlock import TatlockClient
+
+logger = logging.getLogger("desklock.gateway")
+
+app = FastAPI(title="DeskLock Gateway")
+
+PCM_CHUNK_BYTES = 4096
+
+
+@app.get("/healthz")
+async def healthz() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.websocket("/ws/voice")
+async def voice(ws: WebSocket) -> None:
+    await ws.accept()
+    tatlock = TatlockClient()
+    pcm = bytearray()
+    recording = False
+    try:
+        while True:
+            message = await ws.receive()
+            if "bytes" in message and message["bytes"] is not None:
+                if recording:
+                    pcm.extend(message["bytes"])
+                continue
+            if "text" not in message or message["text"] is None:
+                continue
+
+            import json
+
+            event = json.loads(message["text"])
+            if event["type"] == "utterance_start":
+                pcm.clear()
+                recording = True
+            elif event["type"] == "utterance_end":
+                recording = False
+                await _handle_utterance(ws, tatlock, bytes(pcm))
+    except WebSocketDisconnect:
+        logger.info("device disconnected")
+    finally:
+        await tatlock.aclose()
+
+
+async def _handle_utterance(ws: WebSocket, tatlock: TatlockClient, pcm: bytes) -> None:
+    await ws.send_json({"type": "state", "value": "thinking"})
+
+    transcript = await asyncio.to_thread(stt.transcribe, pcm, settings.sample_rate)
+    await ws.send_json({"type": "transcript", "text": transcript})
+    if not transcript:
+        await ws.send_json({"type": "state", "value": "idle"})
+        return
+
+    reply = await tatlock.ask(transcript)
+    await ws.send_json({"type": "reply_text", "text": reply})
+
+    audio = await asyncio.to_thread(tts.synthesize, reply)
+    await ws.send_json({"type": "audio_start", "sample_rate": settings.sample_rate})
+    for offset in range(0, len(audio), PCM_CHUNK_BYTES):
+        await ws.send_bytes(audio[offset : offset + PCM_CHUNK_BYTES])
+    await ws.send_json({"type": "audio_end"})
+    await ws.send_json({"type": "state", "value": "idle"})

gateway/src/desklock_gateway/stt.py

@@ -0,0 +1,24 @@
+"""Speech-to-text: faster-whisper on the tower-of-joy GPU.
+
+Import of faster_whisper is deferred so the gateway can run (health checks,
+protocol tests) without the heavy speech extras installed.
+"""
+
+from .config import settings
+
+_model = None
+
+
+def transcribe(pcm: bytes, sample_rate: int | None = None) -> str:
+    """Transcribe raw s16le mono PCM to text."""
+    global _model
+    if _model is None:
+        from faster_whisper import WhisperModel
+
+        _model = WhisperModel(settings.stt_model, device=settings.stt_device)
+
+    import numpy as np
+
+    audio = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0
+    segments, _info = _model.transcribe(audio, language="en")
+    return " ".join(segment.text.strip() for segment in segments).strip()

gateway/src/desklock_gateway/tatlock.py

@@ -0,0 +1,36 @@
+"""Client for the Tatlock butler's OpenAI-compatible chat API."""
+
+import httpx
+
+from .config import settings
+
+
+class TatlockClient:
+    def __init__(self, base_url: str | None = None) -> None:
+        self._client = httpx.AsyncClient(
+            base_url=base_url or settings.tatlock_base_url,
+            timeout=httpx.Timeout(120.0, connect=5.0),
+        )
+        self._history: list[dict[str, str]] = []
+
+    async def ask(self, text: str) -> str:
+        """Send one user utterance, return Tatlock's reply, keeping conversation history."""
+        self._history.append({"role": "user", "content": text})
+        response = await self._client.post(
+            "/v1/chat/completions",
+            json={
+                "model": settings.tatlock_model,
+                "messages": self._history,
+                "stream": False,
+            },
+        )
+        response.raise_for_status()
+        reply = response.json()["choices"][0]["message"]["content"]
+        self._history.append({"role": "assistant", "content": reply})
+        return reply
+
+    def reset(self) -> None:
+        self._history.clear()
+
+    async def aclose(self) -> None:
+        await self._client.aclose()

gateway/src/desklock_gateway/tts.py

@@ -0,0 +1,23 @@
+"""Text-to-speech: Piper, resampled to the device sample rate.
+
+Import of piper is deferred so the gateway can run without the speech extras.
+"""
+
+from .config import settings
+
+_voice = None
+
+
+def synthesize(text: str) -> bytes:
+    """Synthesize text to raw s16le mono PCM at the configured sample rate."""
+    global _voice
+    if _voice is None:
+        from piper import PiperVoice
+
+        _voice = PiperVoice.load(settings.tts_voice)
+
+    chunks = bytearray()
+    for chunk in _voice.synthesize_stream_raw(text):
+        chunks.extend(chunk)
+    # TODO: resample from the Piper voice's native rate to settings.sample_rate
+    return bytes(chunks)