Полная синхронизация

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
2026-05-19 19:40:42 +03:00
co-authored by Qwen-Coder
parent df5ce8a683
commit cae22004b1
17 changed files with 1139 additions and 138 deletions
+390
View File
@@ -0,0 +1,390 @@
#!/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()