diff --git a/linux_voice_assistant/xvf3800_controller.py b/linux_voice_assistant/xvf3800_controller.py new file mode 100644 index 0000000..da87b9e --- /dev/null +++ b/linux_voice_assistant/xvf3800_controller.py @@ -0,0 +1,320 @@ +"""XVF3800 USB/HID controller scaffold. + +This module is intended to integrate the ReSpeaker XVF3800 mute button and +mute LED with the Linux Voice Assistant (LVA) mute state. + +Design goals: + - HW mute button -> LVA: + * When the user presses the XVF3800 mute button, the board mutes the + mic and lights the red LED. + * This controller detects the change over USB HID and publishes the + standard "set_mic_mute" event so the existing MicMuteHandler can: + - update ServerState.mic_muted + - update ServerState.mic_muted_event + - publish MQTT mute state + - notify the tray, etc. + + - LVA -> HW: + * When LVA toggles mute (via HA MQTT, ReSpeaker 2-Mic GPIO button, etc.), + this controller receives the same "set_mic_mute" event and sends a + USB HID command to synchronize the XVF3800's internal mute state and + red LED. + +This file is a scaffold: + - The USB/HID read/write paths are structured, but the actual report + format is not yet known. + - TODOs are marked where the mute bit must be decoded or set. + - Once the test script (xvf3800_hid_mute_probe.py) has revealed the + HID report layout, implement _parse_hw_mute_from_report() and + _send_hw_mute_state(). +""" + +from __future__ import annotations + +import logging +import threading +import time +from dataclasses import dataclass +from typing import Optional + +from .event_bus import EventBus, EventHandler, subscribe +from .models import ServerState + +_LOGGER = logging.getLogger(__name__) + +try: # pragma: no cover - depends on platform / install + import hid # type: ignore[import] +except Exception: # ImportError, RuntimeError, etc. + hid = None # type: ignore[assignment] + + +@dataclass +class Xvf3800RuntimeConfig: + """Runtime configuration for the XVF3800 controller. + + This is intentionally simple; if we later add more XVF-specific options, + they can be added here and wired from config.json. + """ # noqa: D401 + + enabled: bool = True + vendor_id: int = 0x2886 + product_id: int = 0x001A + read_timeout_ms: int = 200 + poll_interval_seconds: float = 0.02 + + +class Xvf3800Controller(EventHandler): + """Controller for the ReSpeaker XVF3800 USB/HID mute integration. + + Responsibilities: + - Open the XVF3800 HID interface (if available and enabled). + - Run a background thread that reads HID reports and detects mute + state changes, publishing "set_mic_mute" events back into the LVA. + - Listen for "set_mic_mute" events and push desired mute state down + to the XVF3800 (muting the mic and controlling the red LED). + + Safe failure: + - If the 'hid' module is not available, or no matching XVF3800 device + is found, the controller logs an info message and stays inactive. + - All errors in the polling thread are logged but do not crash LVA. + """ + + def __init__( + self, + loop, + event_bus: EventBus, + state: ServerState, + config: Optional[Xvf3800RuntimeConfig] = None, + ) -> None: + super().__init__(event_bus) + + self.loop = loop + self.state = state + self._cfg = config or Xvf3800RuntimeConfig() + + self._thread: Optional[threading.Thread] = None + self._hid_dev: Optional["hid.device"] = None # type: ignore[name-defined] + self._last_hw_mute_state: Optional[bool] = None + self._running = False + + if not self._cfg.enabled: + _LOGGER.info("Xvf3800Controller disabled in config; not initializing HID") + return + + if hid is None: + _LOGGER.info( + "hidapi not available; XVF3800 mute integration disabled. " + "Install it with 'pip install hidapi' if you want this feature.", + ) + return + + # Attempt to open the XVF3800 HID interface. + try: + self._hid_dev = self._open_device( + self._cfg.vendor_id, + self._cfg.product_id, + ) + except Exception: + _LOGGER.exception( + "Failed to open XVF3800 HID device (VID=0x%04x PID=0x%04x); " + "mute integration disabled", + self._cfg.vendor_id, + self._cfg.product_id, + ) + self._hid_dev = None + + if self._hid_dev is None: + return + + self._running = True + self._thread = threading.Thread( + target=self._poll_loop, + name="Xvf3800ControllerThread", + daemon=True, + ) + self._thread.start() + + # Subscribe to events AFTER init + self._subscribe_all_methods() + + _LOGGER.info( + "Xvf3800Controller initialized (VID=0x%04x PID=0x%04x)", + self._cfg.vendor_id, + self._cfg.product_id, + ) + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def stop(self) -> None: + """Stop the polling thread and close HID device (if any).""" + self._running = False + if self._thread is not None and self._thread.is_alive(): + self._thread.join(timeout=1.0) + if self._hid_dev is not None: + try: + self._hid_dev.close() + except Exception: + _LOGGER.exception("Error closing XVF3800 HID device") + _LOGGER.debug("Xvf3800Controller stopped") + + # ------------------------------------------------------------------ + # HID setup + # ------------------------------------------------------------------ + + def _open_device(self, vendor_id: int, product_id: int): + """Open the first HID interface that matches the XVF3800 VID/PID. + + If multiple HID interfaces are present, we currently pick the first + one returned by hid.enumerate(). If this proves unreliable, we can + refine selection using interface_number/product_string. + """ + devices = [ + dev + for dev in hid.enumerate() # type: ignore[attr-defined] + if dev.get("vendor_id") == vendor_id + and dev.get("product_id") == product_id + ] + + if not devices: + _LOGGER.warning( + "No HID interfaces found for XVF3800 (VID=0x%04x PID=0x%04x)", + vendor_id, + product_id, + ) + return None + + dev_info = devices[0] + path = dev_info.get("path") + _LOGGER.info( + "Opening XVF3800 HID device: path=%r serial=%r interface=%r", + path, + dev_info.get("serial_number"), + dev_info.get("interface_number"), + ) + + dev = hid.device() # type: ignore[call-arg] + if path is not None: + dev.open_path(path) + else: + dev.open(vendor_id, product_id) + + dev.set_nonblocking(True) + return dev + + # ------------------------------------------------------------------ + # Polling loop + # ------------------------------------------------------------------ + + def _poll_loop(self) -> None: + """Background loop reading HID reports and detecting mute changes.""" + _LOGGER.debug("Xvf3800Controller polling thread started") + + assert self._hid_dev is not None + + while self._running and not getattr(self.state, "shutdown", False): + try: + data = self._hid_dev.read( # type: ignore[union-attr] + 64, + timeout_ms=self._cfg.read_timeout_ms, + ) + except Exception: + _LOGGER.exception("Error reading from XVF3800 HID device") + break + + if data: + # 'data' is typically a list of ints; convert to bytes for parsing. + report = bytes(data) + new_state = self._parse_hw_mute_from_report(report) + if new_state is not None and new_state != self._last_hw_mute_state: + _LOGGER.debug( + "XVF3800 HW mute state changed: %s -> %s", + self._last_hw_mute_state, + new_state, + ) + self._last_hw_mute_state = new_state + + # Publish standard event; MicMuteHandler will take care + # of updating ServerState and emitting mic_muted/mic_unmuted. + self.loop.call_soon_threadsafe( + self.event_bus.publish, + "set_mic_mute", + {"state": new_state}, + ) + + time.sleep(self._cfg.poll_interval_seconds) + + _LOGGER.debug("Xvf3800Controller polling thread exiting") + + # ------------------------------------------------------------------ + # Parsing & writing HID reports (placeholders) + # ------------------------------------------------------------------ + + def _parse_hw_mute_from_report(self, report: bytes) -> Optional[bool]: + """Extract the current mute state from a raw HID report. + + Returns: + True -> HW reports 'muted' + False -> HW reports 'unmuted' + None -> This report does not contain mute information + + TODO: Implement this once we know the HID report layout. For now, + it logs the report at DEBUG level and returns None. + """ + _LOGGER.debug( + "XVF3800 HID report (len=%d): %s", + len(report), + report.hex(), + ) + # Once we have sample outputs from xvf3800_hid_mute_probe.py, + # implement the bit extraction here. + return None + + def _send_hw_mute_state(self, muted: bool) -> None: + """Send a HID command to set the XVF3800 mute state + LED. + + TODO: Once we understand the HID output/report format (from docs or + reverse engineering), construct and write the correct report. + For now, this method just logs what it *would* do. + """ + if self._hid_dev is None: + return + + _LOGGER.debug( + "(stub) Would send HID command to set XVF3800 mute=%s", + muted, + ) + # Example once we know the correct output report: + # report_id = 0x01 + # payload = bytes([report_id, 0x01 if muted else 0x00, 0x00, 0x00, ...]) + # try: + # self._hid_dev.write(payload) + # except Exception: + # _LOGGER.exception("Error writing XVF3800 HID mute command") + + # ------------------------------------------------------------------ + # EventBus subscriptions + # ------------------------------------------------------------------ + + @subscribe + def set_mic_mute(self, data: dict) -> None: + """Handle global 'set_mic_mute' events. + + This is the *same* event that MQTT, ButtonController, etc. use. By + listening here, we can sync the XVF3800 hardware state with LVA's + logical mute state. + + The central MicMuteHandler is still the source of truth; this + controller should not modify ServerState directly, only attempt to + mirror that state to the hardware. + """ + desired = bool(data.get("state", False)) + _LOGGER.debug( + "Xvf3800Controller received set_mic_mute event: desired=%s", + desired, + ) + + # Remember the last state so we don't treat an echo from hardware as new. + self._last_hw_mute_state = desired + self._send_hw_mute_state(desired)