#!/usr/bin/env python3 """ ReSpeaker XVF3800 USB 4-MIC Array – LED Controller with Custom Animations. Implements custom LED animations mirroring the ESPHome home-assistant-voice.yaml effects. Uses direct USB control transfers to drive the 12 APA102-style LEDs. LED animations mirror the ESPHome home-assistant-voice.yaml effects exactly. Muted indicator: LEDs at the 4 mic positions light red. The XVF3800 has 4 mics at DOA 0°, 90°, 180°, 270° which on a 12-LED ring (30° per step) maps to LEDs 0, 3, 6, 9 — the four cardinal points. Ввод цифр 0-10 выбирает анимацию, q - выход. """ from __future__ import annotations import argparse import logging import math import random import struct import signal import sys import threading import time try: import usb.core import usb.util except ImportError: print("❌ pyusb not installed. Install with: pip install pyusb") sys.exit(1) # =========================================================================== # Configuration # =========================================================================== USB_VENDOR_ID = 0x2886 USB_PRODUCT_IDS = [0x0018, 0x001A] LED_COUNT = 12 LED_BRIGHTNESS = 31 # 0–31 (APA102 5-bit global brightness register) # GPO_SERVICER_RESID LED commands GPO_RESID = 20 LED_EFFECT = 12 LED_BRIGHTNESS_CMD = 13 LED_SPEED = 15 LED_COLOR = 16 LED_DOA_COLOR = 17 # Effects EFFECT_OFF = 0 EFFECT_BREATH = 1 EFFECT_RAINBOW = 2 EFFECT_SINGLE = 3 EFFECT_DOA = 4 EFFECT_RING = 5 # Colors (RGB tuples for custom animations) RGB = tuple[int, int, int] BLACK: RGB = (0, 0, 0) RED: RGB = (255, 0, 0) BLUE: RGB = (0, 0, 255) CYAN: RGB = (0, 200, 200) GREEN: RGB = (0, 200, 50) YELLOW: RGB = (220, 180, 0) MAGENTA: RGB = (255, 0, 255) WHITE: RGB = (255, 255, 255) # Default ring color matching HA Voice PE DEFAULT_RING_COLOR: RGB = (24, 187, 242) # Mic LED positions — 4 mics at DOA 0°, 90°, 180°, 270° # On a 12-LED ring (30° per step): DOA 0° → LED 0, 90° → LED 3, # 180° → LED 6, 270° → LED 9 MIC_LEDS = (0, 3, 6, 9) _LOGGER = logging.getLogger("xvf3800-led") # =========================================================================== # Helper functions # =========================================================================== def _scale(color: RGB, factor: float) -> RGB: """Scale RGB color by factor (0.0-1.0).""" f = max(0.0, min(1.0, factor)) return (int(color[0] * f), int(color[1] * f), int(color[2] * f)) def _rgb_to_bgr_int(color: RGB) -> int: """Convert RGB tuple to BGR integer (0xBBGGRR format).""" r, g, b = color return (b << 16) | (g << 8) | r # =========================================================================== # USB LED driver # =========================================================================== class XVF3800LED: """ Controls the 12 LEDs on ReSpeaker XVF3800 via USB control transfers. Supports both built-in effects and custom frame-by-frame control. """ TIMEOUT = 1000 def __init__(self) -> None: self._dev = None self._dev_lock = threading.Lock() self._find_device() def _find_device(self) -> None: _LOGGER.info("Searching for XVF3800 USB device (VID=%04x, PIDs=%s)...", USB_VENDOR_ID, [f"{pid:04x}" for pid in USB_PRODUCT_IDS]) # List all USB devices for debugging try: import usb.core all_devs = usb.core.find(find_all=True) _LOGGER.info("All USB devices found:") for dev in all_devs: _LOGGER.info(" - VID=%04x, PID=%04x, bus=%d, addr=%d", dev.idVendor, dev.idProduct, dev.bus, dev.address) except Exception as exc: _LOGGER.warning("Could not list USB devices: %s", exc) for pid in USB_PRODUCT_IDS: _LOGGER.debug("Looking for PID %04x...", pid) dev = usb.core.find(idVendor=USB_VENDOR_ID, idProduct=pid) if dev is not None: # НЕ делаем detach_kernel_driver() — это отключает audio интерфейсы! # Control transfer с CTRL_TYPE_VENDOR работает без claim interface # Так работает xvf_host.py — там тоже нет detach_kernel_driver() self._dev = dev _LOGGER.info( "✅ XVF3800 found (USB %04x:%04x, bus %d, addr %d)", USB_VENDOR_ID, pid, dev.bus, dev.address, ) return else: _LOGGER.debug("Device %04x:%04x not found", USB_VENDOR_ID, pid) _LOGGER.error( "❌ XVF3800 not found (USB %04x). Check connection and udev rules.", USB_VENDOR_ID, ) def _ctrl_transfer(self, resid: int, cmdid: int, payload: list) -> bool: with self._dev_lock: if self._dev is None: _LOGGER.warning("USB device not available") return False try: _LOGGER.debug("CTRL: resid=%d, cmdid=%d, payload=%s", resid, cmdid, 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, ) return True except usb.core.USBError as exc: _LOGGER.warning("USB write error: %s", exc) self._dev = None return False def set_effect(self, effect: int) -> bool: return self._ctrl_transfer(GPO_RESID, LED_EFFECT, [effect]) def set_brightness(self, brightness: int) -> bool: brightness = max(0, min(255, brightness)) return self._ctrl_transfer(GPO_RESID, LED_BRIGHTNESS_CMD, [brightness]) def set_speed(self, speed: int) -> bool: speed = max(0, min(255, speed)) return self._ctrl_transfer(GPO_RESID, LED_SPEED, [speed]) def set_color(self, color: int) -> bool: payload = list(struct.pack(b' bool: payload = list(struct.pack(b' bool: """ Set custom LED frame using LED_RING_COLOR command. Uses the XVF3800's LED_RING_COLOR command (resid=20, cmdid=19) which expects 12 × uint32 values (one color per LED in 0xBBGGRR format). """ # Build 12 uint32 values in 0xBBGGRR format colors_uint32 = [] 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))) # XVF3800 uses 0xBBGGRR format color_int = (bb << 16) | (bg << 8) | br colors_uint32.append(color_int) # Pack as 12 × uint32 little-endian payload = b'' for color_int in colors_uint32: payload += struct.pack(' None: self.set_effect(EFFECT_OFF) def close(self) -> None: self.off() with self._dev_lock: if self._dev is not None: usb.util.dispose_resources(self._dev) self._dev = None # =========================================================================== # LED ring animator - custom animations mirroring ESPHome effects # =========================================================================== class LEDRing: """ Manages the 12-LED ring animations for ReSpeaker XVF3800. All animations mirror the ESPHome home-assistant-voice.yaml effects. """ ANIM_OFF = "off" ANIM_TWINKLE = "twinkle" ANIM_TWINKLE_BLUE = "twinkle_blue" ANIM_WAIT_CMD = "waiting" ANIM_LISTENING = "listening" ANIM_THINKING = "thinking" ANIM_REPLYING = "replying" ANIM_MUTED = "muted" ANIM_ERROR = "error" ANIM_TIMER_TICK = "timer_tick" ANIM_TIMER_RING = "timer_ring" def __init__(self) -> None: self._leds = XVF3800LED() self._animation = self.ANIM_OFF self._anim_lock = threading.Lock() self._stop_event = threading.Event() self._thread = threading.Thread( target=self._loop, daemon=True, name="led-ring" ) self._index: int = 0 self._brightness_step: int = 0 self._brightness_dec: bool = True self._twinkle_state: list[float] = [0.0] * LED_COUNT # Timer state for timer_tick animation self._timer_total: int = 60 self._timer_left: int = 60 self._timer_tick_count: int = 0 def start(self) -> None: self._thread.start() def stop(self) -> None: self._stop_event.set() self._thread.join(timeout=2) self._leds.off() self._leds.close() def set_animation(self, name: str) -> None: with self._anim_lock: if self._animation != name: _LOGGER.info("LED animation → %s", name) self._animation = name self._index = 0 self._brightness_step = 0 self._brightness_dec = True self._twinkle_state = [random.random() for _ in range(LED_COUNT)] # Reset timer state when entering timer_tick if name == self.ANIM_TIMER_TICK: self._timer_total = 60 self._timer_left = 60 self._timer_tick_count = 0 # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _color(self) -> RGB: """Default ring colour matching HA Voice PE.""" return DEFAULT_RING_COLOR def _pulse_step(self, steps: int = 10) -> float: """Generate pulsing factor 0.0-1.0.""" factor = (steps - self._brightness_step) / steps if self._brightness_dec: self._brightness_step += 1 else: self._brightness_step -= 1 if self._brightness_step <= 0 or self._brightness_step >= steps: self._brightness_dec = not self._brightness_dec return factor def _apply_mic_indicators(self, pixels: list[RGB]) -> None: """ Mark all 4 mic positions red on the rendered frame. ReSpeaker XVF3800 has 4 mics at DOA 0°, 90°, 180°, 270°. On the 12-LED ring (30° per step): DOA 0° → LED 0 (right) DOA 90° → LED 3 (top) DOA 180° → LED 6 (left) DOA 270° → LED 9 (bottom) Neighbours are blanked so each indicator reads as a distinct dot. """ pixels[11] = BLACK; pixels[0] = RED; pixels[1] = BLACK pixels[2] = BLACK; pixels[3] = RED; pixels[4] = BLACK pixels[5] = BLACK; pixels[6] = RED; pixels[7] = BLACK pixels[8] = BLACK; pixels[9] = RED; pixels[10] = BLACK def _apply_mic_indicators_pulsed(self, pixels: list[RGB], factor: float) -> None: """Pulsed version for timer_ring — keeps indicators in sync with ring brightness.""" pixels[0] = _scale(RED, factor) pixels[3] = _scale(RED, factor) pixels[6] = _scale(RED, factor) pixels[9] = _scale(RED, factor) # ------------------------------------------------------------------ # Animations — each returns sleep time in seconds # ------------------------------------------------------------------ def _anim_off(self) -> float: pixels = [BLACK] * LED_COUNT self._leds.set_led_frame(pixels, brightness=1.0) return 0.1 def _anim_twinkle(self, color: RGB) -> float: FADE = 0.85 SPARK_PROB = 0.15 pixels = [BLACK] * LED_COUNT for i in range(LED_COUNT): if random.random() < SPARK_PROB: self._twinkle_state[i] = 1.0 else: self._twinkle_state[i] *= FADE pixels[i] = _scale(color, self._twinkle_state[i]) self._leds.set_led_frame(pixels) return 0.05 def _anim_spin(self, color: RGB, interval: float, reverse: bool = False) -> float: """ Shared spin for Waiting / Listening / Replying. Mirrors ESPHome "Waiting for Command", "Listening For Command", and "Replying" effects. """ pixels = [BLACK] * LED_COUNT if reverse: self._index = (LED_COUNT + self._index - 1) % LED_COUNT offsets = [(0, 1.0), (1, 192/255), (2, 128/255), (6, 1.0), (7, 192/255), (8, 128/255)] else: offsets = [(0, 1.0), (11, 192/255), (10, 128/255), (6, 1.0), (5, 192/255), (4, 128/255)] for offset, brightness in offsets: idx = (self._index + offset) % LED_COUNT pixels[idx] = _scale(color, brightness) if not reverse: self._index = (self._index + 1) % LED_COUNT self._leds.set_led_frame(pixels) return interval def _anim_thinking(self, color: RGB) -> float: """Two opposing LEDs pulsing. Mirrors ESPHome "Thinking" effect.""" factor = self._pulse_step(10) pixels = [BLACK] * LED_COUNT for i in range(LED_COUNT): if i == self._index % LED_COUNT or i == (self._index + 6) % LED_COUNT: pixels[i] = _scale(color, factor) self._leds.set_led_frame(pixels) return 0.01 def _anim_muted(self, color: RGB) -> float: """ Solid ring with red at all 4 mic positions (cardinal points). DOA 0°/90°/180°/270° → LEDs 0, 3, 6, 9. Neighbours are blanked so each indicator stands out clearly. Mirrors ESPHome "Muted or Silent" effect. """ pixels = [color] * LED_COUNT self._apply_mic_indicators(pixels) self._leds.set_led_frame(pixels) return 0.016 def _anim_error(self) -> float: """All LEDs red, pulsing. Mirrors ESPHome "Error" effect.""" factor = self._pulse_step(10) pixels = [_scale(RED, factor)] * LED_COUNT self._leds.set_led_frame(pixels) return 0.01 def _anim_timer_ring(self, color: RGB) -> float: """ Full ring pulsing; red at all 4 mic positions if muted. Mirrors ESPHome "Timer Ring" effect. """ factor = self._pulse_step(10) pixels = [_scale(color, factor)] * LED_COUNT self._apply_mic_indicators_pulsed(pixels, factor) self._leds.set_led_frame(pixels) return 0.01 def _anim_timer_tick(self, color: RGB, seconds_left: int, total_seconds: int) -> float: """ Arc proportional to remaining time with sweep marker dip. Mirrors ESPHome "Timer Tick" effect exactly. Visual: A bar graph showing remaining time as an arc of LEDs. - Full timer = all LEDs lit - Half timer = 6 LEDs lit - Each LED represents ~5 seconds (for 60s timer) - A small brightness dip moves around the arc as a "sweep marker" """ total = max(total_seconds, 1) timer_ratio = LED_COUNT * seconds_left / total last_led_on = max(0, math.ceil(timer_ratio) - 1) pixels = [BLACK] * LED_COUNT for i in range(LED_COUNT): brightness_dip = ( 0.9 if (i == self._index % LED_COUNT and i != last_led_on) else 1.0 ) if i <= timer_ratio: brightness = min(brightness_dip * (timer_ratio - i), brightness_dip) pixels[i] = _scale(color, brightness) self._index = (LED_COUNT + self._index - 1) % LED_COUNT self._leds.set_led_frame(pixels) return 0.1 # ------------------------------------------------------------------ # Animation loop # ------------------------------------------------------------------ def _loop(self) -> None: _LOGGER.info("LED animation loop started") frame_count = 0 while not self._stop_event.is_set(): with self._anim_lock: anim = self._animation color = self._color() if frame_count % 30 == 0: _LOGGER.debug("Animation loop: anim=%s, frame=%d", anim, frame_count) if anim == self.ANIM_OFF: sleep = self._anim_off() elif anim == self.ANIM_TWINKLE: sleep = self._anim_twinkle(RED) elif anim == self.ANIM_TWINKLE_BLUE: sleep = self._anim_twinkle(BLUE) elif anim == self.ANIM_WAIT_CMD: sleep = self._anim_spin(color, interval=0.1, reverse=False) elif anim == self.ANIM_LISTENING: sleep = self._anim_spin(color, interval=0.05, reverse=False) elif anim == self.ANIM_THINKING: sleep = self._anim_thinking(color) elif anim == self.ANIM_REPLYING: sleep = self._anim_spin(color, interval=0.05, reverse=True) elif anim == self.ANIM_MUTED: sleep = self._anim_muted(color) elif anim == self.ANIM_ERROR: sleep = self._anim_error() elif anim == self.ANIM_TIMER_RING: sleep = self._anim_timer_ring(color) elif anim == self.ANIM_TIMER_TICK: # Simulate timer counting down for demo with self._anim_lock: self._timer_tick_count += 1 # Update timer every second (10 frames at 0.1s each) if self._timer_tick_count % 10 == 0: self._timer_left = max(0, self._timer_left - 1) if self._timer_left == 0: self._timer_left = self._timer_total # Reset for demo sleep = self._anim_timer_tick(color, self._timer_left, self._timer_total) else: sleep = self._anim_off() time.sleep(sleep) frame_count += 1 # =========================================================================== # Main # =========================================================================== ANIM_MAP = { 0: LEDRing.ANIM_OFF, 1: LEDRing.ANIM_TWINKLE, 2: LEDRing.ANIM_TWINKLE_BLUE, 3: LEDRing.ANIM_WAIT_CMD, 4: LEDRing.ANIM_LISTENING, 5: LEDRing.ANIM_THINKING, 6: LEDRing.ANIM_REPLYING, 7: LEDRing.ANIM_MUTED, 8: LEDRing.ANIM_ERROR, 9: LEDRing.ANIM_TIMER_TICK, 10: LEDRing.ANIM_TIMER_RING, } ANIM_NAMES = { 0: "off - выключить", 1: "twinkle - красные искры", 2: "twinkle_blue - синие искры", 3: "waiting - ожидание (медленное вращение)", 4: "listening - слушание (быстрое вращение)", 5: "thinking - мышление (пульсация)", 6: "replying - ответ (обратное вращение)", 7: "muted --muted (красные точки на позициях микрофонов)", 8: "error - ошибка (красная пульсация)", 9: "timer_tick - таймер (дуга времени)", 10: "timer_ring - таймер звонит (пульсация)", } def print_menu() -> None: print("\n" + "=" * 60) print(" ReSpeaker XVF3800 LED Controller (Custom Animations)") print(" Анимации как в ESPHome home-assistant-voice.yaml") print("=" * 60) for idx, name in ANIM_NAMES.items(): print(f" {idx:2d} - {name}") print(" q - Quit") print("=" * 60) def run_terminal_controller() -> None: leds = LEDRing() leds.start() def _shutdown(signum, frame) -> None: print("\n\nShutting down...") leds.stop() sys.exit(0) signal.signal(signal.SIGINT, _shutdown) signal.signal(signal.SIGTERM, _shutdown) # Wait for device to initialize time.sleep(0.5) # Check if device was found if leds._leds._dev is None: print("\n⚠️ WARNING: XVF3800 device not found!") print(" Check USB connection and udev rules.") print(" Run: lsusb | grep -i 2886") else: print(f"\n✅ Device connected: bus={leds._leds._dev.bus}, addr={leds._leds._dev.address}") print_menu() _LOGGER.info("Starting with default animation: twinkle (red sparks)") leds.set_animation(LEDRing.ANIM_TWINKLE) while True: try: user_input = input("\nSelect animation (0-10, q to quit): ").strip().lower() except EOFError: break if user_input == "q": print("Exiting...") break try: anim_idx = int(user_input) if anim_idx not in ANIM_MAP: print(f"❌ Invalid selection: {anim_idx}. Choose 0-10 or q.") continue anim_name = ANIM_MAP[anim_idx] leds.set_animation(anim_name) print(f"✅ Selected: {anim_idx} - {ANIM_NAMES[anim_idx]}") _LOGGER.info("Animation changed to: %s", anim_name) except ValueError: print(f"❌ Invalid input: '{user_input}'. Choose 0-10 or q.") leds.stop() def run_test_pattern() -> None: """Run test pattern to verify LED hardware.""" print("\n" + "=" * 60) print(" XVF3800 LED Hardware Test") print("=" * 60) leds = LEDRing() leds.start() def _shutdown(signum, frame) -> None: print("\n\nTest interrupted") leds.stop() sys.exit(0) signal.signal(signal.SIGINT, _shutdown) signal.signal(signal.SIGTERM, _shutdown) time.sleep(0.5) if leds._leds._dev is None: print("\n❌ ERROR: XVF3800 device not found!") print(" Check USB connection:") print(" - Run: lsusb | grep -i 2886") print(" - Check udev rules: cat /etc/udev/rules.d/99-respeaker.rules") print(" - Try: sudo chmod 666 /dev/bus/usb/*/`lsusb | grep 2886 | awk '{print $4}' | cut -d/ -f2`") leds.stop() return print(f"\n✅ Device found: bus={leds._leds._dev.bus}, addr={leds._leds._dev.address}") print("\nRunning test patterns (5 seconds each):") # Test 1: All red print(" 1. All RED...") leds.set_animation(LEDRing.ANIM_OFF) time.sleep(0.1) # Direct frame test pixels = [RED] * LED_COUNT leds._leds.set_led_frame(pixels, brightness=1.0) time.sleep(5) # Test 2: All green print(" 2. All GREEN...") pixels = [GREEN] * LED_COUNT leds._leds.set_led_frame(pixels, brightness=1.0) time.sleep(5) # Test 3: All blue print(" 3. All BLUE...") pixels = [BLUE] * LED_COUNT leds._leds.set_led_frame(pixels, brightness=1.0) time.sleep(5) # Test 4: Twinkle animation print(" 4. Twinkle animation (10s)...") leds.set_animation(LEDRing.ANIM_TWINKLE) time.sleep(10) print("\n✅ Test complete!") leds.stop() def main() -> None: parser = argparse.ArgumentParser( description="ReSpeaker XVF3800 USB 4-MIC Array LED Controller" ) parser.add_argument( "--debug", action="store_true", help="Enable DEBUG logging", ) parser.add_argument( "--test", action="store_true", help="Run LED hardware test pattern", ) parser.add_argument( "--verbose", action="store_true", help="Enable INFO logging to console", ) args = parser.parse_args() log_level = logging.DEBUG if args.debug else (logging.INFO if args.verbose else logging.WARNING) logging.basicConfig( level=log_level, format="%(asctime)s %(levelname)-8s %(name)s %(message)s", ) if args.test: run_test_pattern() else: run_terminal_controller() if __name__ == "__main__": main()