linux-voice-assistant/linux_voice_assistant/openwakeword_client.py

import json
import logging
import socket
import threading
import time
from typing import Callable, Optional
from urllib.parse import urlparse

_LOGGER = logging.getLogger(__name__)

def _jsonl(obj: dict) -> bytes:
    return (json.dumps(obj, separators=(",", ":")) + "\n").encode("utf-8")

class WyomingWakeClient:
    """Minimal client for Wyoming wake detection (openWakeWord) with extra debug."""

    def __init__(self, uri: str, name: str, rate=16000, width=2, channels=1):
        self.uri, self.name = uri, name
        self.rate, self.width, self.channels = rate, width, channels
        self._sock: Optional[socket.socket] = None
        self._reader: Optional[threading.Thread] = None
        self._on_detect: Optional[Callable[[str, Optional[int]], None]] = None
        self._closed = False
        self._chunks = 0
        self._last_log_time = 0.0

    def connect(self, on_detect: Callable[[str, Optional[int]], None]) -> None:
        self._on_detect = on_detect
        parsed = urlparse(self.uri)
        if parsed.scheme != "tcp":
            raise ValueError(f"Only tcp:// URIs are supported (got {self.uri})")
        host = parsed.hostname or "127.0.0.1"
        port = parsed.port or 10400
        _LOGGER.debug("[OWW] Connecting to %s:%s", host, port)
        self._sock = socket.create_connection((host, port))
        self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self._send({ "type": "audio-start", "data": { "rate": self.rate, "width": self.width, "channels": self.channels }})
        self._arm_detect()
        self._reader = threading.Thread(target=self._read_loop, daemon=True)
        self._reader.start()
        _LOGGER.info("[OWW] Connected to Wyoming wake server at %s", self.uri)

    def close(self) -> None:
        self._closed = True
        try:
            self._send({"type": "audio-stop"})
        except Exception:
            pass
        try:
            if self._sock:
                self._sock.close()
        finally:
            self._sock = None

    def _arm_detect(self) -> None:
        _LOGGER.debug("[OWW] Arming detect for name=%s", self.name)
        self._send({"type": "detect", "data": {"names": [self.name]}})

    def send_audio_chunk(self, pcm: bytes, timestamp_ms: Optional[int] = None) -> None:
        if self._sock is None or self._closed:
            return
        hdr = {
            "type": "audio-chunk",
            "data": {"rate": self.rate, "width": self.width, "channels": self.channels},
            "payload_length": len(pcm),
        }
        if timestamp_ms is not None:
            hdr["data"]["timestamp"] = timestamp_ms
        try:
            self._sock.sendall(_jsonl(hdr))
            self._sock.sendall(pcm)
        except Exception:
            _LOGGER.exception("[OWW] Failed to send audio chunk")
            return

        self._chunks += 1
        if (self._chunks % 1000) == 0:
            # ~64ms * 1000 = ~64s of audio
            _LOGGER.debug("[OWW] Sent %d chunks (~%.1fs)", self._chunks, self._chunks * 0.064)

    # internals
    def _send(self, obj: dict) -> None:
        assert self._sock is not None
        try:
            self._sock.sendall(_jsonl(obj))
        except Exception:
            _LOGGER.exception("[OWW] send failed (%s)", obj.get("type"))

    def _cycle_stream(self) -> None:
        """Some servers require restarting the audio stream after a detection."""
        _LOGGER.debug("[OWW] Cycling audio stream")
        try:
            self._send({"type": "audio-stop"})
        except Exception:
            _LOGGER.exception("[OWW] audio-stop failed")
        try:
            self._send({"type": "audio-start", "data": {"rate": self.rate, "width": self.width, "channels": self.channels}})
        except Exception:
            _LOGGER.exception("[OWW] audio-start failed")

    def _read_loop(self) -> None:
        assert self._sock is not None
        f = self._sock.makefile("rb")
        _LOGGER.debug("[OWW] Reader loop started")
        buf = b""
        decoder = json.JSONDecoder()
        while not self._closed:
            chunk = f.readline()
            if not chunk:
                _LOGGER.debug("[OWW] Reader EOF")
                break
            buf += chunk
            try:
                # There might be multiple JSON objects in one line; parse until exhausted.
                while buf:
                    s = buf.decode("utf-8")
                    obj, idx = decoder.raw_decode(s)
                    buf = s[idx:].lstrip().encode("utf-8")
                    mtype = obj.get("type")
                    if mtype and mtype != "detection":
                        _LOGGER.debug("[OWW] recv %s", mtype)
                    if mtype == "detection":
                        data = obj.get("data", {}) or {}
                        name = data.get("name") or self.name
                        ts = data.get("timestamp")
                        _LOGGER.info("[OWW] Detection: name=%s ts=%s (chunks=%d)", name, ts, self._chunks)
                        if self._on_detect:
                            try:
                                self._on_detect(name, ts)
                            except Exception:
                                _LOGGER.exception("[OWW] Error in detection callback")
                        # Some servers need an audio stream restart before re-arming
                        try:
                            self._cycle_stream()
                        except Exception:
                            _LOGGER.exception("[OWW] Failed to cycle stream")
                        # Re-arm detect so subsequent detections work
                        try:
                            self._arm_detect()
                        except Exception:
                            _LOGGER.exception("[OWW] Failed to re-arm detect")
            except json.JSONDecodeError:
                # Need more bytes; continue accumulating
                pass