sirmrak/linux-voice-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Delete linux_voice_assistant/xvf3800_controller.py

Browse Source 
Duplicate file...
This commit is contained in:
imonlinux
2025-12-08 11:34:22 -08:00
committed by GitHub
parent e845c5c811
commit 59d8071384
1 changed files with 0 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux_voice_assistant/xvf3800_controller.py
-320
View File
@@ -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)