Files
esp32-server/app/config.py
kunthawat 261b0f3e91 feat: wire Opus decode + VAD turn detection to firmware-compatible reply
Device streams continuous 20ms Opus frames with no turn markers. The loop
now decodes Opus, feeds the VAD, fires one turn on end-of-speech, and
replies with the firmware JSON contract (tts.state / stt.text) + Opus audio.

- app/audio/opus.py: RealOpus/Opus16kEncoder/PassthroughOpus; fix add_dll_directory
  to use absolute paths (WinError 87)
- app/xiaozhi/websocket.py: DeviceLoop.run() = decode -> VAD -> run_turn -> reply
- app/config.py: AudioConfig (opus kind + VAD thresholds)
- config.yaml: audio.opus=real, VAD 5/8/3000 frames
- tests/test_app.py: utterance = loud burst + silence; asserts stt + audio + tts stop
- requirements.txt: opuslib>=3

21/21 tests; live handshake verified via zhi.moreminimore.com on venv python.
2026-10-03 20:35:38 +07:00

129 lines
4.6 KiB
Python

"""Configuration for the Hermes-Xiaozhi bridge.
Everything is data. The bridge code never hard-codes a model name, a device
token, or a firmware constant (MUST-NOT-001, CON-002, CON-004).
``Config.from_file`` / ``Config.from_dict`` produce a pydantic model so bad
config fails loudly at startup, not mid-conversation.
"""
from __future__ import annotations
from typing import Literal, Optional
from pydantic import BaseModel, Field
class STTConfig(BaseModel):
engine: Literal["mock", "typhoon", "faster-whisper"] = "mock"
language: str = "th"
sample_rate: int = 16000
# Optional per-engine knobs (endpoint, model path) — kept as free-form so a
# backend can be swapped without a schema change (MUST-NOT-001).
typhoon_endpoint: Optional[str] = None
whisper_model: str = "large-v3"
class TTSConfig(BaseModel):
engine: Literal["mock", "jaitts"] = "mock"
voice: str = "default"
device: str = "cuda:0"
sample_rate: int = 24000
jaitts_endpoint: Optional[str] = None
class HermesConfig(BaseModel):
# Transport for the local Qwen/Hermes voice profile. "mock" for tests.
transport: Literal["mock", "openai_http"] = "mock"
base_url: Optional[str] = None
api_key: Optional[str] = None
# REQ-004 — voice profile is reasoning-none, read/recall tools only.
reasoning: str = "none"
tools: list[str] = Field(default_factory=lambda: ["session_search"])
max_context_tokens: int = 16000 # REQ-005 / Phase 17 working context
max_history_turns: int = 12
class DeviceConfig(BaseModel):
device_id: str
token_hash: str # SHA-256 hex of the shared token (REQ-009)
class ProtocolConfig(BaseModel):
# FIRMWARE: verify against the real Xiaozhi firmware before locking (CON-004).
hello_version: int = 1
opus_sample_rate: int = 16000
opus_channels: int = 1
frame_ms: int = 20 # typical Xiaozhi Opus frame size
class AudioConfig(BaseModel):
"""Device audio path: the device streams continuous 20 ms Opus frames
(no turn markers), so the bridge decodes Opus, accumulates through a
VAD, and fires a turn on end-of-speech. ``passthrough`` treats frames as
raw PCM (synthetic clients / tests without a native codec)."""
opus: Literal["passthrough", "real"] = "passthrough"
vad_min_speech_frames: int = 5 # 100 ms of speech before an utterance counts
vad_min_silence_frames: int = 8 # 160 ms of silence ends the utterance
vad_max_utterance_frames: int = 3000 # 60 s safety flush
class OtaConfig(BaseModel):
"""OTA endpoint data (GET/POST /ota) — xiaozhi device onboarding.
The device bakes in CONFIG_OTA_URL at build time; on boot it POSTs there
and reads ``websocket`` {url, token, version} + ``server_time`` from the
response. We deliver OUR server URL here — this is how the device learns
it must connect to this bridge (firmware detail, verified in
xiaozhi-esp32 v2.5.0 main/ota.cc).
"""
server_ws_url: str # absolute wss:// URL the device should connect to
server_token: str # token the device presents on the WS handshake
protocol_version: int = 1
timezone_minutes: int = 420 # UTC+07:00 (Bangkok)
firmware_version: str = "2.5.0" # report our own build version
class SecurityConfig(BaseModel):
enabled: bool = True
devices: list[DeviceConfig] = Field(default_factory=list)
class GpuConfig(BaseModel):
monitor_enabled: bool = True
stt_vram_budget_gb: float = 2.0
tts_vram_budget_gb: float = 4.0
class Config(BaseModel):
host: str = "127.0.0.1" # REQ-012 — LAN/localhost only; tunnel is external
port: int = 8765
ws_path: str = "/ws/xiaozhi"
stt: STTConfig = Field(default_factory=STTConfig)
tts: TTSConfig = Field(default_factory=TTSConfig)
hermes: HermesConfig = Field(default_factory=HermesConfig)
protocol: ProtocolConfig = Field(default_factory=ProtocolConfig)
security: SecurityConfig = Field(default_factory=SecurityConfig)
ota: Optional[OtaConfig] = None
gpu: GpuConfig = Field(default_factory=GpuConfig)
audio: AudioConfig = Field(default_factory=AudioConfig)
@classmethod
def from_dict(cls, data: dict) -> "Config":
return cls.model_validate(data)
@classmethod
def from_file(cls, path: str) -> "Config":
import yaml
with open(path, "r", encoding="utf-8") as fh:
data = yaml.safe_load(fh) or {}
return cls.from_dict(data)
# A tiny, importable default used by tests and as a fallback when no file is
# present. It is intentionally all-mock so the bridge is usable anywhere
# (CON-002) without a GPU, device, or native libs.
DEFAULT = Config()