#!/usr/bin/env python3 """ ReSpeaker XVF3800 USB 4-MIC Array – Terminal LED Controller. Использует встроенные эффекты XVF3800: 0 = off 1 = breath (дыхание) 2 = rainbow (радуга) 3 = single color (один цвет) 4 = doa (направление звука) 5 = ring (кольцо, может не поддерживаться) Ввод цифр 0-10 выбирает анимацию, q - выход. """ from __future__ import annotations import argparse import logging 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] # GPO_SERVICER_RESID LED commands GPO_RESID = 20 LED_EFFECT = 12 LED_BRIGHTNESS = 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 (0xBBGGRR format - BGR byte order!) COLOR_CYAN = 0xC0BB00 COLOR_RED = 0x0000FF COLOR_GREEN = 0x00FF00 COLOR_BLUE = 0xFF0000 COLOR_YELLOW = 0x00FFFF COLOR_MAGENTA = 0xFF00FF COLOR_WHITE = 0xFFFFFF COLOR_OFF = 0x000000 _LOGGER = logging.getLogger("xvf3800-led") class XVF3800LED: """Controls LEDs on XVF3800 via USB control transfers.""" TIMEOUT = 1000 def __init__(self) -> None: self._dev = None self._dev_lock = threading.Lock() self._find_device() def _find_device(self) -> None: for pid in USB_PRODUCT_IDS: dev = usb.core.find(idVendor=USB_VENDOR_ID, idProduct=pid) if dev is not None: try: if dev.is_kernel_driver_active(0): dev.detach_kernel_driver(0) except Exception: pass self._dev = dev _LOGGER.info( "XVF3800 found (USB %04x:%04x, bus %d, addr %d)", USB_VENDOR_ID, pid, dev.bus, dev.address, ) return _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, [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' 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 - uses built-in XVF3800 effects # =========================================================================== class LEDRing: """ Manages 12-LED ring animations using built-in XVF3800 effects. """ ANIM_OFF = "off" ANIM_TWINKLE = "twinkle" # breath red ANIM_TWINKLE_BLUE = "twinkle_blue" # breath blue ANIM_WAIT_CMD = "waiting" # rainbow slow ANIM_LISTENING = "listening" # rainbow fast ANIM_THINKING = "thinking" # breath cyan ANIM_REPLYING = "replying" # rainbow fast reverse (same as listening) ANIM_MUTED = "muted" # single red ANIM_ERROR = "error" # breath magenta fast ANIM_TIMER_TICK = "timer_tick" # doa mode ANIM_TIMER_RING = "timer_ring" # breath fast 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 = 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 def _apply_animation(self) -> float: anim = self._animation if anim == self.ANIM_OFF: self._leds.set_effect(EFFECT_OFF) return 0.1 elif anim == self.ANIM_TWINKLE: # Breath effect with red color self._leds.set_effect(EFFECT_BREATH) self._leds.set_color(COLOR_RED) self._leds.set_brightness(150) self._leds.set_speed(2) return 0.5 elif anim == self.ANIM_TWINKLE_BLUE: # Breath effect with blue color self._leds.set_effect(EFFECT_BREATH) self._leds.set_color(COLOR_BLUE) self._leds.set_brightness(150) self._leds.set_speed(2) return 0.5 elif anim == self.ANIM_WAIT_CMD: # Rainbow slow self._leds.set_effect(EFFECT_RAINBOW) self._leds.set_brightness(150) self._leds.set_speed(1) return 0.5 elif anim == self.ANIM_LISTENING: # Rainbow fast self._leds.set_effect(EFFECT_RAINBOW) self._leds.set_brightness(150) self._leds.set_speed(4) return 0.5 elif anim == self.ANIM_THINKING: # Breath cyan self._leds.set_effect(EFFECT_BREATH) self._leds.set_color(COLOR_CYAN) self._leds.set_brightness(150) self._leds.set_speed(3) return 0.5 elif anim == self.ANIM_REPLYING: # Rainbow fast (same as listening - no reverse on XVF3800) self._leds.set_effect(EFFECT_RAINBOW) self._leds.set_brightness(150) self._leds.set_speed(4) return 0.5 elif anim == self.ANIM_MUTED: # Solid red self._leds.set_effect(EFFECT_SINGLE) self._leds.set_color(COLOR_RED) self._leds.set_brightness(200) return 0.5 elif anim == self.ANIM_ERROR: # Breath magenta fast self._leds.set_effect(EFFECT_BREATH) self._leds.set_color(COLOR_MAGENTA) self._leds.set_brightness(200) self._leds.set_speed(5) return 0.5 elif anim == self.ANIM_TIMER_TICK: # DoA mode - shows direction self._leds.set_effect(EFFECT_DOA) self._leds.set_doa_color(0x303030, COLOR_CYAN) self._leds.set_brightness(150) return 0.5 elif anim == self.ANIM_TIMER_RING: # Breath fast self._leds.set_effect(EFFECT_BREATH) self._leds.set_color(COLOR_YELLOW) self._leds.set_brightness(200) self._leds.set_speed(5) return 0.5 return 0.5 def _loop(self) -> None: while not self._stop_event.is_set(): with self._anim_lock: anim = self._animation sleep = self._apply_animation() time.sleep(sleep) # =========================================================================== # 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 - красный", 8: "error - маджента быстрое", 9: "timer_tick - DoA режим", 10: "timer_ring - жёлтое быстрое", } def print_menu() -> None: print("\n" + "=" * 50) print(" ReSpeaker XVF3800 LED Controller") print(" (встроенные эффекты)") print("=" * 50) for idx, name in ANIM_NAMES.items(): print(f" {idx:2d} - {name}") print(" q - Quit") print("=" * 50) 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) print_menu() 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]}") except ValueError: print(f"❌ Invalid input: '{user_input}'. Choose 0-10 or q.") 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", ) args = parser.parse_args() logging.basicConfig( level=logging.DEBUG if args.debug else logging.INFO, format="%(asctime)s %(levelname)-8s %(name)s %(message)s", ) run_terminal_controller() if __name__ == "__main__": main()