sirmrakandQwen-Coder
19a11aac6e
Port LED effects from respeaker_usb_mic_array.py to example_led.py:
- Replace built-in XVF3800 effects with custom frame-by-frame animations
- Use LED_RING_COLOR command (resid=20, cmdid=19) for direct LED control
- Implement 11 animations matching home-assistant-voice.yaml:
- off, twinkle, twinkle_blue, waiting, listening, thinking,
replying, muted, error, timer_tick, timer_ring
- Muted indicator: 4 red LEDs at mic positions (0°, 90°, 180°, 270°)
- Timer tick: arc shows remaining time with sweep marker
- Add --test mode for hardware verification
- Add --verbose flag for INFO logging
- Enhanced USB diagnostics and error reporting
Technical changes:
- set_led_frame() sends 12×uint32 colors in 0xBBGGRR format
- RGB tuples with _scale() for brightness control
- _pulse_step() for smooth pulsing effects
- _apply_mic_indicators() for muted state visualization
Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
723 lines
25 KiB
Python
723 lines
25 KiB
Python
#!/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:
|
||
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)",
|
||
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'<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 set_led_frame(self, pixels: list[RGB], brightness: float = 1.0) -> 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('<I', color_int)
|
||
|
||
_LOGGER.debug("LED frame: brightness=%.2f, pixels=%s, payload_len=%d",
|
||
brightness, pixels, len(payload))
|
||
|
||
# Send via vendor control transfer
|
||
# resid=20 (GPO_SERVICER_RESID), cmdid=19 (LED_RING_COLOR)
|
||
with self._dev_lock:
|
||
if self._dev is None:
|
||
_LOGGER.warning("❌ LEDs [simulated]: device not available")
|
||
_LOGGER.warning(" Payload would be: %s", payload.hex())
|
||
return False
|
||
try:
|
||
result = self._dev.ctrl_transfer(
|
||
usb.util.CTRL_OUT | usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_RECIPIENT_DEVICE,
|
||
0, # request
|
||
19, # wValue = cmdid
|
||
20, # wIndex = resid
|
||
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)
|
||
_LOGGER.error(" Payload was: %s", payload.hex())
|
||
self._dev = None
|
||
return False
|
||
|
||
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 - 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()
|