sirmrak
5a5cac90b1
Проблема: detach_kernel_driver(0) отключает ядро от интерфейса 0, который является частью audio function (интерфейсы 0-2). Решение: Убрать detach_kernel_driver() вообще. Control transfer с CTRL_TYPE_VENDOR работает без claim interface и не требует отключения ядра. Это как в xvf_host.py — там тоже нет detach_kernel_driver().
359 lines
11 KiB
Python
359 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
XVF3800 USB LED Control Module.
|
||
|
||
Прямой доступ к ReSpeaker XVF3800 через USB control transfers.
|
||
Использует pyusb для отправки команд без subprocess.
|
||
|
||
Пример использования:
|
||
from wyoming_satellite.xvf_usb import XVF3800Controller
|
||
|
||
controller = XVF3800Controller()
|
||
controller.set_led_effect(1) # breath effect
|
||
controller.set_led_color(0x00BCD4) # бирюзовый
|
||
controller.set_led_brightness(150)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import struct
|
||
import threading
|
||
from typing import Optional, Tuple
|
||
|
||
try:
|
||
import usb.core
|
||
import usb.util
|
||
_HAS_USB = True
|
||
except ImportError:
|
||
_HAS_USB = False
|
||
logging.warning("pyusb not installed – XVF3800 control disabled")
|
||
|
||
|
||
# ===========================================================================
|
||
# Configuration
|
||
# ===========================================================================
|
||
|
||
USB_VENDOR_ID = 0x2886
|
||
USB_PRODUCT_IDS = [0x0018, 0x001A]
|
||
|
||
# GPO_SERVICER_RESID LED commands
|
||
GPO_RESID = 20
|
||
LED_EFFECT_CMD = 12
|
||
LED_BRIGHTNESS_CMD = 13
|
||
LED_SPEED_CMD = 15
|
||
LED_COLOR_CMD = 16
|
||
LED_DOA_COLOR_CMD = 17
|
||
LED_RING_COLOR_CMD = 19
|
||
|
||
# Effects
|
||
EFFECT_OFF = 0
|
||
EFFECT_BREATH = 1
|
||
EFFECT_RAINBOW = 2
|
||
EFFECT_SINGLE = 3
|
||
EFFECT_DOA = 4
|
||
EFFECT_RING = 5
|
||
|
||
# Color format: RGB tuple
|
||
RGB = Tuple[int, int, int]
|
||
|
||
_LOGGER = logging.getLogger("xvf_usb")
|
||
|
||
|
||
# ===========================================================================
|
||
# Helper functions
|
||
# ===========================================================================
|
||
|
||
def rgb_to_bgr_int(r: int, g: int, b: int) -> int:
|
||
"""Convert RGB (0-255) to 0xBBGGRR format."""
|
||
return (b << 16) | (g << 8) | r
|
||
|
||
|
||
def bgr_int_to_rgb(color_int: int) -> RGB:
|
||
"""Convert 0xBBGGRR to RGB tuple."""
|
||
b = (color_int >> 16) & 0xFF
|
||
g = (color_int >> 8) & 0xFF
|
||
r = color_int & 0xFF
|
||
return (r, g, b)
|
||
|
||
|
||
def scale_color(color: RGB, brightness: int) -> RGB:
|
||
"""
|
||
Scale RGB color by brightness (0-255).
|
||
|
||
Args:
|
||
color: RGB tuple (r, g, b)
|
||
brightness: Brightness value 0-255
|
||
|
||
Returns:
|
||
Scaled RGB tuple
|
||
"""
|
||
factor = brightness / 255.0
|
||
return (
|
||
int(color[0] * factor),
|
||
int(color[1] * factor),
|
||
int(color[2] * factor)
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# XVF3800 Controller
|
||
# ===========================================================================
|
||
|
||
class XVF3800Controller:
|
||
"""
|
||
Controller for ReSpeaker XVF3800 via USB.
|
||
|
||
Provides direct USB control transfer access to LED commands.
|
||
Thread-safe with automatic device reconnection.
|
||
|
||
Example:
|
||
controller = XVF3800Controller()
|
||
controller.set_led_effect(EFFECT_BREATH)
|
||
controller.set_led_color(rgb_to_bgr_int(0, 188, 212))
|
||
controller.set_led_brightness(150)
|
||
"""
|
||
|
||
TIMEOUT = 1000 # USB transfer timeout in ms
|
||
|
||
def __init__(
|
||
self,
|
||
vendor_id: int = USB_VENDOR_ID,
|
||
product_ids: list[int] | None = None
|
||
) -> None:
|
||
self._vendor_id = vendor_id
|
||
self._product_ids = product_ids or USB_PRODUCT_IDS
|
||
self._dev: Optional[usb.core.Device] = None
|
||
self._dev_lock = threading.Lock()
|
||
self._find_device()
|
||
|
||
def _find_device(self) -> None:
|
||
"""Find and initialize USB device."""
|
||
_LOGGER.info(
|
||
"Searching for XVF3800 USB device (VID=%04x, PIDs=%s)...",
|
||
self._vendor_id,
|
||
[f"{pid:04x}" for pid in self._product_ids]
|
||
)
|
||
|
||
# List all USB devices for debugging
|
||
if _HAS_USB:
|
||
try:
|
||
all_devs = usb.core.find(find_all=True)
|
||
_LOGGER.debug("All USB devices found:")
|
||
for dev in all_devs:
|
||
_LOGGER.debug(
|
||
" - VID=%04x, PID=%04x, bus=%d, addr=%d",
|
||
dev.idVendor, dev.idProduct, dev.bus, dev.address
|
||
)
|
||
except Exception as exc:
|
||
_LOGGER.debug("Could not list USB devices: %s", exc)
|
||
|
||
for pid in self._product_ids:
|
||
_LOGGER.debug("Looking for PID %04x...", pid)
|
||
dev = usb.core.find(idVendor=self._vendor_id, idProduct=pid)
|
||
if dev is not None:
|
||
# НЕ делаем detach_kernel_driver() — это отключает audio интерфейсы!
|
||
# Control transfer с CTRL_TYPE_VENDOR работает без claim interface
|
||
|
||
self._dev = dev
|
||
_LOGGER.info(
|
||
"✅ XVF3800 found (USB %04x:%04x, bus %d, addr %d)",
|
||
self._vendor_id, pid, dev.bus, dev.address
|
||
)
|
||
return
|
||
else:
|
||
_LOGGER.debug("Device %04x:%04x not found", self._vendor_id, pid)
|
||
|
||
_LOGGER.error(
|
||
"❌ XVF3800 not found (USB %04x). Check connection and udev rules.",
|
||
self._vendor_id
|
||
)
|
||
|
||
def _ctrl_transfer(
|
||
self,
|
||
resid: int,
|
||
cmdid: int,
|
||
payload: list | bytes
|
||
) -> bool:
|
||
"""
|
||
Send USB control transfer.
|
||
|
||
Args:
|
||
resid: Resource ID (e.g., GPO_RESID=20)
|
||
cmdid: Command ID within the resource
|
||
payload: Data to send (list of int or bytes)
|
||
|
||
Returns:
|
||
True if successful, False otherwise
|
||
"""
|
||
with self._dev_lock:
|
||
if self._dev is None:
|
||
_LOGGER.warning("USB device not available")
|
||
return False
|
||
|
||
try:
|
||
# Convert list to bytes if needed
|
||
if isinstance(payload, list):
|
||
payload = bytes(payload)
|
||
|
||
_LOGGER.debug(
|
||
"CTRL: resid=%d, cmdid=%d, payload_len=%d",
|
||
resid, cmdid, len(payload)
|
||
)
|
||
|
||
result = self._dev.ctrl_transfer(
|
||
usb.util.CTRL_OUT | usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_RECIPIENT_DEVICE,
|
||
0, # request
|
||
cmdid, # wValue
|
||
resid, # wIndex
|
||
payload,
|
||
self.TIMEOUT
|
||
)
|
||
|
||
_LOGGER.debug("✅ USB transfer successful: %d bytes sent", result)
|
||
return True
|
||
|
||
except usb.core.USBError as exc:
|
||
_LOGGER.error("❌ USB write error: %s", exc)
|
||
self._dev = None
|
||
return False
|
||
|
||
def is_connected(self) -> bool:
|
||
"""Check if USB device is available."""
|
||
with self._dev_lock:
|
||
return self._dev is not None
|
||
|
||
def close(self) -> None:
|
||
"""Release USB resources."""
|
||
self.set_led_effect(EFFECT_OFF)
|
||
with self._dev_lock:
|
||
if self._dev is not None:
|
||
usb.util.dispose_resources(self._dev)
|
||
self._dev = None
|
||
|
||
# =======================================================================
|
||
# LED Commands
|
||
# =======================================================================
|
||
|
||
def set_led_effect(self, effect: int) -> bool:
|
||
"""
|
||
Set LED effect mode.
|
||
|
||
Args:
|
||
effect: 0=off, 1=breath, 2=rainbow, 3=single, 4=doa, 5=ring
|
||
|
||
Returns:
|
||
True if successful
|
||
"""
|
||
return self._ctrl_transfer(GPO_RESID, LED_EFFECT_CMD, [effect])
|
||
|
||
def set_led_brightness(self, brightness: int) -> bool:
|
||
"""
|
||
Set LED brightness (for breath and rainbow effects only).
|
||
|
||
Args:
|
||
brightness: 0-255
|
||
|
||
Returns:
|
||
True if successful
|
||
"""
|
||
brightness = max(0, min(255, brightness))
|
||
return self._ctrl_transfer(GPO_RESID, LED_BRIGHTNESS_CMD, [brightness])
|
||
|
||
def set_led_speed(self, speed: int) -> bool:
|
||
"""
|
||
Set LED effect speed (for breath and rainbow effects).
|
||
|
||
Args:
|
||
speed: 0-255
|
||
|
||
Returns:
|
||
True if successful
|
||
"""
|
||
speed = max(0, min(255, speed))
|
||
return self._ctrl_transfer(GPO_RESID, LED_SPEED_CMD, [speed])
|
||
|
||
def set_led_color(self, color_bgr: int) -> bool:
|
||
"""
|
||
Set LED color for breath or single color mode.
|
||
|
||
Args:
|
||
color_bgr: Color in 0xBBGGRR format
|
||
|
||
Returns:
|
||
True if successful
|
||
"""
|
||
payload = list(struct.pack('<I', color_bgr))
|
||
return self._ctrl_transfer(GPO_RESID, LED_COLOR_CMD, payload)
|
||
|
||
def set_led_doa_color(self, base_color_bgr: int, doa_color_bgr: int) -> bool:
|
||
"""
|
||
Set LED colors for DoA (Direction of Arrival) mode.
|
||
|
||
Args:
|
||
base_color_bgr: Base color in 0xBBGGRR format
|
||
doa_color_bgr: DoA color in 0xBBGGRR format
|
||
|
||
Returns:
|
||
True if successful
|
||
"""
|
||
payload = list(struct.pack('<II', base_color_bgr, doa_color_bgr))
|
||
return self._ctrl_transfer(GPO_RESID, LED_DOA_COLOR_CMD, payload)
|
||
|
||
def set_led_ring_color(self, pixels: list[RGB], brightness: float = 1.0) -> bool:
|
||
"""
|
||
Set custom LED frame for ring mode (12 individual LEDs).
|
||
|
||
Args:
|
||
pixels: List of 12 RGB tuples
|
||
brightness: Brightness multiplier 0.0-1.0
|
||
|
||
Returns:
|
||
True if successful
|
||
"""
|
||
if len(pixels) != 12:
|
||
_LOGGER.error("Expected 12 pixels, got %d", len(pixels))
|
||
return False
|
||
|
||
# Build 12 uint32 values in 0xBBGGRR format
|
||
payload = b''
|
||
for r, g, b in pixels:
|
||
br = max(0, min(255, int(r * brightness)))
|
||
bg = max(0, min(255, int(g * brightness)))
|
||
bb = max(0, min(255, int(b * brightness)))
|
||
color_int = (bb << 16) | (bg << 8) | br
|
||
payload += struct.pack('<I', color_int)
|
||
|
||
return self._ctrl_transfer(GPO_RESID, LED_RING_COLOR_CMD, payload)
|
||
|
||
# =======================================================================
|
||
# Convenience Methods
|
||
# =======================================================================
|
||
|
||
def set_led_state(
|
||
self,
|
||
effect: int,
|
||
color_bgr: Optional[int] = None,
|
||
brightness: Optional[int] = None,
|
||
speed: Optional[int] = None
|
||
) -> None:
|
||
"""
|
||
Set complete LED state in one call.
|
||
|
||
Args:
|
||
effect: Effect mode (0-5)
|
||
color_bgr: Color in 0xBBGGRR format (for effects 1, 3)
|
||
brightness: Brightness 0-255 (for effects 1, 2)
|
||
speed: Speed 0-255 (for effects 1, 2)
|
||
"""
|
||
self.set_led_effect(effect)
|
||
|
||
if color_bgr is not None:
|
||
self.set_led_color(color_bgr)
|
||
|
||
if brightness is not None:
|
||
self.set_led_brightness(brightness)
|
||
|
||
if speed is not None:
|
||
self.set_led_speed(speed)
|