From 59d8071384777fcfb709f02d4e7435af4bec4504 Mon Sep 17 00:00:00 2001 From: imonlinux <39863321+imonlinux@users.noreply.github.com> Date: Mon, 8 Dec 2025 11:34:22 -0800 Subject: [PATCH] Delete linux_voice_assistant/xvf3800_controller.py Duplicate file... --- linux_voice_assistant/xvf3800_controller.py | 320 -------------------- 1 file changed, 320 deletions(-) delete mode 100644 linux_voice_assistant/xvf3800_controller.py diff --git a/linux_voice_assistant/xvf3800_controller.py b/linux_voice_assistant/xvf3800_controller.py deleted file mode 100644 index da87b9e..0000000 --- a/linux_voice_assistant/xvf3800_controller.py +++ /dev/null @@ -1,320 +0,0 @@ -"""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)