sirmrak/satellite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

feat: Multi-event server support + USB LED controller

Browse Source 
New features:
- wyoming_satellite/xvf_usb.py — прямой USB доступ к XVF3800 без subprocess
- docs/development/multi-event-server-plan.md — план разработки event server
- examples/respeaker_usb_mic_array.py — исходный файл с LED анимациями

Documentation updates:
- DEPLOY.md — добавлена секция про LED управление и event server план
- docs/integration/mqtt-led.md — обновлена документация по USB и event server
- docs/deployment/systemd-services.md — добавлен event server раздел

Refactoring:
- examples/wyoming-led.py — замена subprocess на XVF3800Controller
- install_service.sh — удалён (устарел, используется install_systemd_services.sh)

Bug fixes:
- wyoming-led.py — яркость DoA эффекта через масштабирование цвета
- systemd сервис — добавлена поддержка wyoming_satellite.xvf_usb модуля
This commit is contained in:
sirmrak
2026-05-20 13:50:16 +03:00
parent 75ba5481b6
commit c62203b220
9 changed files with 2267 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wyoming_satellite/xvf_usb.py
+362
View File
@@ -0,0 +1,362 @@
#!/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:
try:
if dev.is_kernel_driver_active(0):
_LOGGER.info("Detaching kernel driver from device...")
dev.detach_kernel_driver(0)
except Exception as exc:
_LOGGER.warning("Could not detach kernel driver: %s", exc)
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)