391 lines
12 KiB
Python
391 lines
12 KiB
Python
#!/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'<I', color))
|
||
return self._ctrl_transfer(GPO_RESID, LED_COLOR, payload)
|
||
|
||
def set_doa_color(self, base_color: int, doa_color: int) -> bool:
|
||
payload = list(struct.pack(b'<II', base_color, doa_color))
|
||
return self._ctrl_transfer(GPO_RESID, LED_DOA_COLOR, payload)
|
||
|
||
def off(self) -> 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()
|