Update xvf3800_led_backend.py
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
This commit is contained in:
@@ -54,6 +54,10 @@ PARAMETERS = {
|
|||||||
|
|
||||||
# Newer firmware: per-LED ring control (12 WS2812 LEDs)
|
# Newer firmware: per-LED ring control (12 WS2812 LEDs)
|
||||||
"LED_RING_COLOR": (20, 19, 12, "rw", "uint32"),
|
"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]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -67,13 +71,25 @@ class _ReSpeaker:
|
|||||||
def __init__(self, dev: "usb.core.Device") -> None: # type: ignore[name-defined]
|
def __init__(self, dev: "usb.core.Device") -> None: # type: ignore[name-defined]
|
||||||
self.dev = dev
|
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:
|
def close(self) -> None:
|
||||||
"""Release any underlying libusb resources (best-effort)."""
|
"""Release any underlying libusb resources (best-effort)."""
|
||||||
try:
|
if hasattr(self, 'dev') and self.dev is not None:
|
||||||
usb.util.dispose_resources(self.dev)
|
try:
|
||||||
except Exception:
|
usb.util.dispose_resources(self.dev)
|
||||||
pass
|
except Exception as e:
|
||||||
|
_LOGGER.debug("Error disposing USB resources: %s", e)
|
||||||
|
finally:
|
||||||
|
self.dev = None
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
# Encoding / decoding helpers
|
# Encoding / decoding helpers
|
||||||
@@ -181,9 +197,6 @@ class _ReSpeaker:
|
|||||||
|
|
||||||
raise RuntimeError(f"XVF3800 control read failed (status={status}, name={name})")
|
raise RuntimeError(f"XVF3800 control read failed (status={status}, name={name})")
|
||||||
|
|
||||||
def close(self) -> None:
|
|
||||||
usb.util.dispose_resources(self.dev)
|
|
||||||
|
|
||||||
|
|
||||||
def _find_device(vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> Optional[_ReSpeaker]:
|
def _find_device(vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> Optional[_ReSpeaker]:
|
||||||
dev = usb.core.find(idVendor=vid, idProduct=pid)
|
dev = usb.core.find(idVendor=vid, idProduct=pid)
|
||||||
@@ -192,7 +205,6 @@ def _find_device(vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> Option
|
|||||||
return _ReSpeaker(dev)
|
return _ReSpeaker(dev)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class XVF3800USBDevice:
|
class XVF3800USBDevice:
|
||||||
"""Small helper for issuing non-LED XVF3800 control commands via PyUSB.
|
"""Small helper for issuing non-LED XVF3800 control commands via PyUSB.
|
||||||
|
|
||||||
@@ -201,12 +213,17 @@ class XVF3800USBDevice:
|
|||||||
|
|
||||||
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID):
|
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID):
|
||||||
self._rsp = _find_device(vid=vid, pid=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:
|
def close(self) -> None:
|
||||||
try:
|
if self._rsp is not None:
|
||||||
self._rsp.close()
|
try:
|
||||||
except Exception:
|
self._rsp.close()
|
||||||
pass
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
# --- Device control -----------------------------------------------------
|
# --- Device control -----------------------------------------------------
|
||||||
|
|
||||||
@@ -255,11 +272,15 @@ class XVF3800USBDevice:
|
|||||||
if settle_s > 0:
|
if settle_s > 0:
|
||||||
time.sleep(settle_s)
|
time.sleep(settle_s)
|
||||||
|
|
||||||
|
|
||||||
class XVF3800LedBackend:
|
class XVF3800LedBackend:
|
||||||
"""High-level LED backend for the XVF3800."""
|
"""High-level LED backend for the XVF3800."""
|
||||||
|
|
||||||
ring_led_count: int = 12
|
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:
|
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> None:
|
||||||
wrapper = _find_device(vid, pid)
|
wrapper = _find_device(vid, pid)
|
||||||
if wrapper is None:
|
if wrapper is None:
|
||||||
@@ -270,6 +291,18 @@ class XVF3800LedBackend:
|
|||||||
self._dev = wrapper
|
self._dev = wrapper
|
||||||
self.supports_per_led: bool = False
|
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
|
# Best-effort feature detection: if we can read LED_RING_COLOR, we assume
|
||||||
# per-LED control is supported by the current firmware.
|
# per-LED control is supported by the current firmware.
|
||||||
try:
|
try:
|
||||||
@@ -296,12 +329,43 @@ class XVF3800LedBackend:
|
|||||||
self.supports_per_led,
|
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
|
# Legacy/global controls
|
||||||
# ---------------------------------------------------------------------
|
# ---------------------------------------------------------------------
|
||||||
|
|
||||||
def set_effect(self, effect_id: int) -> None:
|
def set_effect(self, effect_id: int) -> None:
|
||||||
"""Set LED effect mode (0=off, 1=breath, 2=rainbow, 3=single color, 4=doa)."""
|
"""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])
|
self._dev.write("LED_EFFECT", [int(effect_id) & 0xFF])
|
||||||
|
|
||||||
def set_brightness(self, brightness_0_255: int) -> None:
|
def set_brightness(self, brightness_0_255: int) -> None:
|
||||||
@@ -333,6 +397,9 @@ class XVF3800LedBackend:
|
|||||||
raise ValueError(
|
raise ValueError(
|
||||||
f"LED_RING_COLOR expects {self.ring_led_count} values, got {len(color_values)}"
|
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])
|
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:
|
def set_ring_rgb(self, colors: Sequence[Tuple[int, int, int]]) -> None:
|
||||||
|
|||||||
Reference in New Issue
Block a user