b8950b38a4
Xiaozhi WebSocket endpoint with handshake/auth, mock STT/TTS/Hermes backends, Thai chunker, barge-in queue, latency logger, GPU planner, voice-profile guard (reasoning=none, session_search only). CON-002: no GPU/audio libs loaded at import. MUST-NOT-001: all model names/tokens from config.
54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
"""Opus encode/decode — lazy (CON-002).
|
|
|
|
Only the codec the device uses is needed; we default to a passthrough so the
|
|
pipeline runs on PCM in tests. The real opus (``opuslib`` / ``PyNaCl``-free
|
|
``opus``) is imported at construction time, never at module import.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
|
|
class OpusCodec(ABC):
|
|
@abstractmethod
|
|
def encode(self, pcm: bytes) -> bytes: ...
|
|
|
|
@abstractmethod
|
|
def decode(self, data: bytes) -> bytes: ...
|
|
|
|
|
|
class PassthroughOpus(OpusCodec):
|
|
"""PCM in, PCM out. Lets the whole pipeline run without a native codec."""
|
|
|
|
def encode(self, pcm: bytes) -> bytes:
|
|
return pcm
|
|
|
|
def decode(self, data: bytes) -> bytes:
|
|
return data
|
|
|
|
|
|
class RealOpus(OpusCodec):
|
|
"""Lazy real codec. Fails to construct only if opus is not installed —
|
|
acceptable on the Mac; the voice-server has it."""
|
|
|
|
def __init__(self, sample_rate: int = 16000, channels: int = 1, frame_ms: int = 20):
|
|
import opus # type: ignore # noqa: F401
|
|
|
|
self._enc = opus.Encoder(sample_rate, channels, opus.APPLICATION_VOIP)
|
|
self._dec = opus.Decoder(sample_rate, channels)
|
|
self._frame_ms = frame_ms
|
|
|
|
def encode(self, pcm: bytes) -> bytes:
|
|
return self._enc.encode(pcm, self._frame_ms)
|
|
|
|
def decode(self, data: bytes) -> bytes:
|
|
return self._dec.decode(data, 20000)
|
|
|
|
|
|
def make_opus(kind: str = "passthrough", **kw) -> OpusCodec:
|
|
if kind == "passthrough":
|
|
return PassthroughOpus()
|
|
if kind == "real":
|
|
return RealOpus(**kw)
|
|
raise ValueError(f"unknown opus kind: {kind}")
|