imonlinux
d10cf87eef
Problem XVF3800 LED ring worked inconsistently (~50% success rate at startup) due to: 1. Timing race: LED commands sent before WS2812 power (X0D33) enabled 2. Missing context manager: USB resources could leak on exceptions Changes Context Manager Support (Critical Fix B) Added __enter__ and __exit__ methods to _ReSpeaker class Enhanced close() with proper null checks and error handling Enables safe with statement usage LED Power Management (NEW Critical Fix) Proactive initialization: Enable X0D33 immediately in __init__ before any LED operations Defensive runtime checks: Added _ensure_led_power() helper method Power verification: Check X0D33 state before set_effect() and set_ring_colors() Added 50ms settle time after enabling power Documentation Added GPO_WRITE_VALUE and GPO_READ_VALUES to PARAMETERS dict Added GPO_WS2812_POWER_INDEX constant for clarity Enhanced inline comments explaining power management
445 lines
16 KiB
Python
445 lines
16 KiB
Python
"""XVF3800 USB LED ring backend.
|
|
|
|
This backend supports two generations of LED control:
|
|
|
|
1) Legacy/global LED control (older firmware):
|
|
- LED_EFFECT, LED_BRIGHTNESS, LED_SPEED, LED_COLOR
|
|
|
|
2) Per-LED ring control (newer firmware):
|
|
- LED_RING_COLOR: 12x uint32 values, one per LED (0xRRGGBB)
|
|
|
|
We intentionally keep the implementation minimal and self-contained so it can be
|
|
used by LVA without shelling out to xvf_host.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import struct
|
|
import time
|
|
from typing import List, Optional, Sequence, Tuple
|
|
|
|
import usb.core # type: ignore[import]
|
|
import usb.util # type: ignore[import]
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
CONTROL_SUCCESS = 0
|
|
SERVICER_COMMAND_RETRY = 64
|
|
|
|
# name -> (resid, cmdid, count, access, type)
|
|
PARAMETERS = {
|
|
# ---------------------------------------------------------------------
|
|
# APPLICATION_SERVICER_RESID (core info)
|
|
# ---------------------------------------------------------------------
|
|
"VERSION": (48, 0, 3, "ro", "uint8"),
|
|
|
|
# Firmware / device control
|
|
"REBOOT": (48, 7, 1, "wo", "uint8"),
|
|
"SAVE_CONFIGURATION": (48, 9, 1, "wo", "uint8"),
|
|
|
|
# Audio manager output routing (category, source) per channel
|
|
"AUDIO_MGR_OP_L": (35, 15, 2, "rw", "uint8"),
|
|
"AUDIO_MGR_OP_R": (35, 19, 2, "rw", "uint8"),
|
|
|
|
# ---------------------------------------------------------------------
|
|
# GPO_SERVICER_RESID (LED controls)
|
|
# ---------------------------------------------------------------------
|
|
# Legacy/global controls
|
|
"LED_EFFECT": (20, 12, 1, "rw", "uint8"),
|
|
"LED_BRIGHTNESS": (20, 13, 1, "rw", "uint8"),
|
|
"LED_GAMMIFY": (20, 14, 1, "rw", "uint8"),
|
|
"LED_SPEED": (20, 15, 1, "rw", "uint8"),
|
|
"LED_COLOR": (20, 16, 1, "rw", "uint32"),
|
|
|
|
# Newer firmware: per-LED ring control (12 WS2812 LEDs)
|
|
"LED_RING_COLOR": (20, 19, 12, "rw", "uint32"),
|
|
|
|
# GPO control for button monitoring and LED power
|
|
"GPO_READ_VALUES": (20, 0, 5, "ro", "uint8"), # [X0D11, X0D30, X0D31, X0D33, X0D39]
|
|
"GPO_WRITE_VALUE": (20, 1, 2, "wo", "uint8"), # [pin_index, value]
|
|
}
|
|
|
|
|
|
class _ReSpeaker:
|
|
"""Low-level USB control wrapper for XVF3800 parameters."""
|
|
|
|
TIMEOUT_MS = 100_000
|
|
VID = 0x2886
|
|
PID = 0x001A
|
|
|
|
def __init__(self, dev: "usb.core.Device") -> None: # type: ignore[name-defined]
|
|
self.dev = dev
|
|
|
|
# CRITICAL FIX: Add context manager support
|
|
def __enter__(self):
|
|
"""Context manager entry."""
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
"""Context manager exit."""
|
|
self.close()
|
|
return False
|
|
|
|
def close(self) -> None:
|
|
"""Release any underlying libusb resources (best-effort)."""
|
|
if hasattr(self, 'dev') and self.dev is not None:
|
|
try:
|
|
usb.util.dispose_resources(self.dev)
|
|
except Exception as e:
|
|
_LOGGER.debug("Error disposing USB resources: %s", e)
|
|
finally:
|
|
self.dev = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Encoding / decoding helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _pack_values(self, data_type: str, values: Sequence[int]) -> bytes:
|
|
if data_type == "uint8":
|
|
return bytes([int(v) & 0xFF for v in values])
|
|
if data_type == "uint32":
|
|
return b"".join(struct.pack("<I", int(v) & 0xFFFFFFFF) for v in values)
|
|
if data_type == "int32":
|
|
return b"".join(struct.pack("<i", int(v)) for v in values)
|
|
raise ValueError(f"Unsupported data type '{data_type}'")
|
|
|
|
def _unpack_values(self, data_type: str, raw: bytes, count: int) -> List[int]:
|
|
if data_type == "uint8":
|
|
return list(raw[:count])
|
|
if data_type == "uint32":
|
|
return list(struct.unpack("<" + "I" * count, raw[: count * 4]))
|
|
if data_type == "int32":
|
|
return list(struct.unpack("<" + "i" * count, raw[: count * 4]))
|
|
raise ValueError(f"Unsupported data type '{data_type}'")
|
|
|
|
def _read_length(self, data_type: str, count: int) -> int:
|
|
# +1 for status byte returned by XMOS vendor control read
|
|
if data_type == "uint8":
|
|
return count + 1
|
|
if data_type in ("uint32", "int32"):
|
|
return (count * 4) + 1
|
|
raise ValueError(f"Unsupported data type '{data_type}'")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public parameter IO
|
|
# ------------------------------------------------------------------
|
|
|
|
def write(self, name: str, data_list: Sequence[int]) -> None:
|
|
try:
|
|
resid, cmdid, count, access, data_type = PARAMETERS[name]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Unknown XVF3800 parameter '{name}'") from exc
|
|
|
|
if access == "ro":
|
|
raise ValueError(f"{name} is read-only")
|
|
|
|
if len(data_list) != count:
|
|
raise ValueError(f"{name} expects {count} values, got {len(data_list)}")
|
|
|
|
payload = self._pack_values(data_type, data_list)
|
|
|
|
_LOGGER.debug(
|
|
"XVF3800 write: name=%s resid=%s cmdid=%s payload_len=%d",
|
|
name,
|
|
resid,
|
|
cmdid,
|
|
len(payload),
|
|
)
|
|
|
|
self.dev.ctrl_transfer(
|
|
usb.util.CTRL_OUT
|
|
| usb.util.CTRL_TYPE_VENDOR
|
|
| usb.util.CTRL_RECIPIENT_DEVICE,
|
|
0,
|
|
cmdid,
|
|
resid,
|
|
payload,
|
|
self.TIMEOUT_MS,
|
|
)
|
|
|
|
def read(self, name: str, max_retries: int = 10) -> List[int]:
|
|
try:
|
|
resid, cmdid, count, access, data_type = PARAMETERS[name]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Unknown XVF3800 parameter '{name}'") from exc
|
|
|
|
if access == "wo":
|
|
raise ValueError(f"{name} is write-only")
|
|
|
|
length = self._read_length(data_type, count)
|
|
wValue = 0x80 | cmdid # per XMOS protocol: read is (0x80 | cmdid)
|
|
|
|
attempt = 0
|
|
while True:
|
|
attempt += 1
|
|
resp = self.dev.ctrl_transfer(
|
|
usb.util.CTRL_IN
|
|
| usb.util.CTRL_TYPE_VENDOR
|
|
| usb.util.CTRL_RECIPIENT_DEVICE,
|
|
0,
|
|
wValue,
|
|
resid,
|
|
length,
|
|
self.TIMEOUT_MS,
|
|
)
|
|
|
|
if not resp:
|
|
raise RuntimeError("Empty response from XVF3800 control read")
|
|
|
|
status = int(resp[0])
|
|
if status == CONTROL_SUCCESS:
|
|
raw = bytes(resp[1:])
|
|
return self._unpack_values(data_type, raw, count)
|
|
|
|
if status == SERVICER_COMMAND_RETRY and attempt < max_retries:
|
|
continue
|
|
|
|
raise RuntimeError(f"XVF3800 control read failed (status={status}, name={name})")
|
|
|
|
|
|
def _find_device(vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> Optional[_ReSpeaker]:
|
|
dev = usb.core.find(idVendor=vid, idProduct=pid)
|
|
if not dev:
|
|
return None
|
|
return _ReSpeaker(dev)
|
|
|
|
|
|
class XVF3800USBDevice:
|
|
"""Small helper for issuing non-LED XVF3800 control commands via PyUSB.
|
|
|
|
This is intentionally minimal and does *not* depend on xvf_host.py.
|
|
"""
|
|
|
|
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID):
|
|
self._rsp = _find_device(vid=vid, pid=pid)
|
|
if self._rsp is None:
|
|
raise RuntimeError(
|
|
f"XVF3800 USB device not found (vid=0x{vid:04x}, pid=0x{pid:04x})"
|
|
)
|
|
|
|
def close(self) -> None:
|
|
if self._rsp is not None:
|
|
try:
|
|
self._rsp.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# --- Device control -----------------------------------------------------
|
|
|
|
def reboot(self) -> None:
|
|
"""Reboot the XVF3800 firmware (resets parameters to defaults)."""
|
|
self._rsp.write("REBOOT", [1])
|
|
|
|
def save_configuration(self) -> None:
|
|
"""Persist current runtime settings to flash."""
|
|
self._rsp.write("SAVE_CONFIGURATION", [1])
|
|
|
|
# --- Audio routing ------------------------------------------------------
|
|
|
|
def set_audio_mgr_op_l(self, category: int, source: int) -> None:
|
|
"""Set the L output channel source selection (category, source)."""
|
|
self._rsp.write("AUDIO_MGR_OP_L", [int(category) & 0xFF, int(source) & 0xFF])
|
|
|
|
def set_audio_mgr_op_r(self, category: int, source: int) -> None:
|
|
"""Set the R output channel source selection (category, source)."""
|
|
self._rsp.write("AUDIO_MGR_OP_R", [int(category) & 0xFF, int(source) & 0xFF])
|
|
|
|
# --- Wait helpers -------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def wait_for_reenumeration(
|
|
vid: int = _ReSpeaker.VID,
|
|
pid: int = _ReSpeaker.PID,
|
|
timeout_s: float = 12.0,
|
|
settle_s: float = 1.0,
|
|
) -> None:
|
|
"""Wait for the XVF3800 to disappear and then reappear on USB."""
|
|
start = time.time()
|
|
# Wait for it to disappear (best effort)
|
|
while time.time() - start < timeout_s:
|
|
if usb.core.find(idVendor=vid, idProduct=pid) is None:
|
|
break
|
|
time.sleep(0.1)
|
|
|
|
# Wait for it to reappear
|
|
while time.time() - start < timeout_s:
|
|
if usb.core.find(idVendor=vid, idProduct=pid) is not None:
|
|
break
|
|
time.sleep(0.1)
|
|
|
|
# Give kernel/userspace audio stack a moment to settle
|
|
if settle_s > 0:
|
|
time.sleep(settle_s)
|
|
|
|
|
|
class XVF3800LedBackend:
|
|
"""High-level LED backend for the XVF3800."""
|
|
|
|
ring_led_count: int = 12
|
|
|
|
# GPO indices (from XVF3800 documentation)
|
|
GPO_WS2812_POWER_INDEX = 3 # X0D33 in GPO_READ_VALUES response
|
|
|
|
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> None:
|
|
wrapper = _find_device(vid, pid)
|
|
if wrapper is None:
|
|
raise RuntimeError(
|
|
f"XVF3800 USB device not found (vid=0x{vid:04x}, pid=0x{pid:04x})"
|
|
)
|
|
|
|
self._dev = wrapper
|
|
self.supports_per_led: bool = False
|
|
|
|
# CRITICAL FIX: Ensure WS2812 LED power is enabled BEFORE any LED operations
|
|
# This prevents intermittent LED failures caused by X0D33 being low at startup
|
|
try:
|
|
_LOGGER.debug("Ensuring XVF3800 WS2812 LED power (X0D33) is enabled")
|
|
self._dev.write("GPO_WRITE_VALUE", [33, 1]) # X0D33 = high
|
|
time.sleep(0.05) # Give firmware time to settle
|
|
_LOGGER.info("XVF3800 WS2812 LED power enabled")
|
|
except Exception as e:
|
|
_LOGGER.warning(
|
|
"Could not enable WS2812 LED power, LEDs may not work reliably: %s", e
|
|
)
|
|
|
|
# Best-effort feature detection: if we can read LED_RING_COLOR, we assume
|
|
# per-LED control is supported by the current firmware.
|
|
try:
|
|
_ = self._dev.read("LED_RING_COLOR")
|
|
self.supports_per_led = True
|
|
except Exception:
|
|
self.supports_per_led = False
|
|
|
|
# Best-effort version read (not fatal)
|
|
try:
|
|
ver = self._dev.read("VERSION")
|
|
_LOGGER.info(
|
|
"Connected to XVF3800 USB device (vid=0x%04x, pid=0x%04x, version=%s, per_led=%s)",
|
|
vid,
|
|
pid,
|
|
".".join(str(int(x)) for x in ver[:3]),
|
|
self.supports_per_led,
|
|
)
|
|
except Exception:
|
|
_LOGGER.info(
|
|
"Connected to XVF3800 USB device (vid=0x%04x, pid=0x%04x, per_led=%s)",
|
|
vid,
|
|
pid,
|
|
self.supports_per_led,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Helper: Ensure LED power before critical operations
|
|
# ---------------------------------------------------------------------
|
|
|
|
def _ensure_led_power(self) -> bool:
|
|
"""Ensure WS2812 LED power is enabled before operations.
|
|
|
|
This provides belt-and-suspenders protection against the LED power
|
|
being disabled by firmware or button interactions.
|
|
|
|
Returns:
|
|
bool: True if power is confirmed on, False if check failed
|
|
"""
|
|
try:
|
|
# Read GPO values
|
|
values = self._dev.read("GPO_READ_VALUES")
|
|
if len(values) > self.GPO_WS2812_POWER_INDEX:
|
|
ws2812_power = bool(values[self.GPO_WS2812_POWER_INDEX])
|
|
if not ws2812_power:
|
|
_LOGGER.warning("WS2812 LED power was off, re-enabling")
|
|
self._dev.write("GPO_WRITE_VALUE", [33, 1])
|
|
time.sleep(0.01) # Brief settle time
|
|
return True
|
|
return True
|
|
except Exception as e:
|
|
_LOGGER.debug("Could not verify WS2812 LED power state: %s", e)
|
|
return False
|
|
return False
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Legacy/global controls
|
|
# ---------------------------------------------------------------------
|
|
|
|
def set_effect(self, effect_id: int) -> None:
|
|
"""Set LED effect mode (0=off, 1=breath, 2=rainbow, 3=single color, 4=doa)."""
|
|
# Ensure power before effect change
|
|
self._ensure_led_power()
|
|
self._dev.write("LED_EFFECT", [int(effect_id) & 0xFF])
|
|
|
|
def set_brightness(self, brightness_0_255: int) -> None:
|
|
"""Set LED brightness (0-255)."""
|
|
value = max(0, min(255, int(brightness_0_255)))
|
|
self._dev.write("LED_BRIGHTNESS", [value])
|
|
|
|
def set_speed(self, speed_id: int) -> None:
|
|
"""Set LED effect speed (0=slow, 1=medium, 2=fast)."""
|
|
self._dev.write("LED_SPEED", [int(speed_id) & 0xFF])
|
|
|
|
def set_color(self, r: int, g: int, b: int) -> None:
|
|
"""Set LED color for breath / single color modes (0xRRGGBB)."""
|
|
r = max(0, min(255, int(r)))
|
|
g = max(0, min(255, int(g)))
|
|
b = max(0, min(255, int(b)))
|
|
color_value = (r << 16) | (g << 8) | b
|
|
self._dev.write("LED_COLOR", [color_value])
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Per-LED ring control (newer firmware)
|
|
# ---------------------------------------------------------------------
|
|
|
|
def set_ring_colors(self, color_values: Sequence[int]) -> None:
|
|
"""Set all 12 ring LEDs with 0xRRGGBB values (length must be 12)."""
|
|
if not self.supports_per_led:
|
|
raise RuntimeError("Per-LED ring control is not supported by this firmware")
|
|
if len(color_values) != self.ring_led_count:
|
|
raise ValueError(
|
|
f"LED_RING_COLOR expects {self.ring_led_count} values, got {len(color_values)}"
|
|
)
|
|
|
|
# Ensure power before writing ring colors
|
|
self._ensure_led_power()
|
|
self._dev.write("LED_RING_COLOR", [int(v) & 0xFFFFFFFF for v in color_values])
|
|
|
|
def set_ring_rgb(self, colors: Sequence[Tuple[int, int, int]]) -> None:
|
|
"""Set all 12 ring LEDs with (r,g,b) tuples (length must be 12)."""
|
|
if len(colors) != self.ring_led_count:
|
|
raise ValueError(
|
|
f"Ring expects {self.ring_led_count} colors, got {len(colors)}"
|
|
)
|
|
vals: List[int] = []
|
|
for r, g, b in colors:
|
|
r = max(0, min(255, int(r)))
|
|
g = max(0, min(255, int(g)))
|
|
b = max(0, min(255, int(b)))
|
|
vals.append((r << 16) | (g << 8) | b)
|
|
self.set_ring_colors(vals)
|
|
|
|
def set_ring_solid(self, r: int, g: int, b: int) -> None:
|
|
"""Convenience: set all ring LEDs to the same RGB color."""
|
|
self.set_ring_rgb([(r, g, b)] * self.ring_led_count)
|
|
|
|
def clear_ring(self) -> None:
|
|
"""Convenience: turn all ring LEDs off (per-LED mode)."""
|
|
if not self.supports_per_led:
|
|
# Legacy fallback: just set effect off
|
|
self.set_effect(0)
|
|
self.set_brightness(0)
|
|
return
|
|
self.set_ring_colors([0] * self.ring_led_count)
|
|
|
|
def get_version(self) -> Optional[Tuple[int, int, int]]:
|
|
"""Return (major, minor, patch) if readable, else None."""
|
|
try:
|
|
ver = self._dev.read("VERSION")
|
|
if len(ver) >= 3:
|
|
return (int(ver[0]), int(ver[1]), int(ver[2]))
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
def close(self) -> None:
|
|
if self._dev is not None:
|
|
self._dev.close()
|