sirmrak
b1f7185a4f
DoA эффект: - base_color: 0x303030 (серый, без масштабирования) - doa_color: 0xD4BC00 (бирюзовый, без масштабирования) - brightness: управляется отдельно через set_led_brightness() Это как в оригинале до изменений яркости.
492 lines
18 KiB
Python
492 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Wyoming LED Control with MQTT Discovery for Home Assistant
|
||
Полная версия: Light + Sensor + 7 Number sliders
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import logging
|
||
import sys
|
||
import signal
|
||
import json
|
||
from functools import partial
|
||
from pathlib import Path
|
||
|
||
import paho.mqtt.client as mqtt
|
||
from wyoming.event import Event
|
||
from wyoming.satellite import SatelliteConnected
|
||
from wyoming.server import AsyncEventHandler, AsyncServer
|
||
from wyoming.snd import Played
|
||
from wyoming.vad import VoiceStarted, VoiceStopped
|
||
from wyoming.wake import Detection
|
||
|
||
# Импорт USB контроллера
|
||
from wyoming_satellite.xvf_usb import (
|
||
XVF3800Controller,
|
||
EFFECT_OFF, EFFECT_BREATH, EFFECT_RAINBOW, EFFECT_DOA,
|
||
rgb_to_bgr_int, scale_color
|
||
)
|
||
|
||
# ====================== НАСТРОЙКИ ======================
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S"
|
||
)
|
||
logger = logging.getLogger("XVF3800 LED Service")
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
CONFIG_FILE = SCRIPT_DIR / "led_brightness_config.json"
|
||
|
||
MQTT_BROKER = "192.168.200.100"
|
||
MQTT_PORT = 1883
|
||
MQTT_USER = "mqtt"
|
||
MQTT_PASS = "mqtt"
|
||
|
||
# Топики
|
||
MQTT_TOPIC_COMMAND = "wyoming/led/command"
|
||
MQTT_TOPIC_BRIGHTNESS_SET = "wyoming/led/brightness/set"
|
||
MQTT_TOPIC_STATE = "wyoming/led/state"
|
||
MQTT_TOPIC_BRIGHTNESS_STATE = "wyoming/led/brightness/state"
|
||
MQTT_TOPIC_SATELLITE_STATE = "wyoming/satellite/state"
|
||
MQTT_TOPIC_AVAILABILITY = "wyoming/led/availability"
|
||
|
||
MQTT_DISCOVERY_PREFIX = "homeassistant"
|
||
MQTT_CLIENT_ID = "wyoming-led-service"
|
||
|
||
MQTT_DISCOVERY_DEVICE = {
|
||
"identifiers": ["wyoming-led-repeaker"],
|
||
"name": "Wyoming LED ReSpeaker",
|
||
"manufacturer": "ReSpeaker",
|
||
"model": "XVF3800",
|
||
}
|
||
|
||
# Настройки яркости
|
||
BRIGHTNESS_CONFIG = {
|
||
"listening": 120, "wake": 200, "thinking": 150,
|
||
"speaking": 140, "idle": 80, "paused": 40, "error": 255,
|
||
"global_multiplier": 1.0
|
||
}
|
||
|
||
mqtt_client = None
|
||
current_state = "idle"
|
||
last_mqtt_brightness = None
|
||
ha_disabled = False
|
||
led_controller: XVF3800Controller | None = None
|
||
|
||
|
||
# ====================== CONFIG ======================
|
||
def load_brightness_config():
|
||
global BRIGHTNESS_CONFIG
|
||
if CONFIG_FILE.exists():
|
||
try:
|
||
with open(CONFIG_FILE) as f:
|
||
saved = json.load(f)
|
||
BRIGHTNESS_CONFIG.update(saved)
|
||
logger.info(f"📂 Загружены настройки яркости из {CONFIG_FILE}")
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ Ошибка загрузки config: {e}")
|
||
else:
|
||
save_brightness_config()
|
||
|
||
def save_brightness_config():
|
||
try:
|
||
with open(CONFIG_FILE, "w") as f:
|
||
json.dump(BRIGHTNESS_CONFIG, f, indent=2)
|
||
logger.info(f"💾 Сохранено в {CONFIG_FILE}: {BRIGHTNESS_CONFIG}")
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка сохранения config: {e}")
|
||
|
||
|
||
# ====================== LED ======================
|
||
def init_leds() -> bool:
|
||
"""Инициализация USB контроллера."""
|
||
global led_controller
|
||
try:
|
||
led_controller = XVF3800Controller()
|
||
if led_controller.is_connected():
|
||
logger.info("✅ XVF3800 USB подключён")
|
||
return True
|
||
else:
|
||
logger.warning("⚠️ XVF3800 не найден (USB)")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка инициализации LED: {e}")
|
||
return False
|
||
|
||
|
||
def get_effective_brightness(state: str) -> int:
|
||
base = BRIGHTNESS_CONFIG.get(state, 100)
|
||
multiplier = BRIGHTNESS_CONFIG.get("global_multiplier", 1.0)
|
||
return int(base * multiplier)
|
||
|
||
|
||
def set_led_state(state: str, skip_publish=False):
|
||
"""Установка LED состояния через USB контроллер."""
|
||
global current_state
|
||
current_state = state
|
||
brightness = get_effective_brightness(state)
|
||
|
||
if led_controller is None:
|
||
logger.warning("⚠️ LED контроллер не инициализирован")
|
||
if not skip_publish:
|
||
publish_state(state)
|
||
publish_satellite_state(state)
|
||
return
|
||
|
||
if state == "listening":
|
||
# Бирюзовое дыхание
|
||
led_controller.set_led_effect(EFFECT_BREATH)
|
||
led_controller.set_led_color(rgb_to_bgr_int(0, 188, 212)) # 0x00BCD4
|
||
led_controller.set_led_brightness(brightness)
|
||
led_controller.set_led_speed(2)
|
||
logger.info(f"🔵 Listening: бирюзовое дыхание ({brightness})")
|
||
|
||
elif state == "wake":
|
||
# Красное дыхание
|
||
led_controller.set_led_effect(EFFECT_BREATH)
|
||
led_controller.set_led_color(rgb_to_bgr_int(255, 0, 0)) # 0xFF0000
|
||
led_controller.set_led_brightness(brightness)
|
||
logger.info(f"🔴 Wake: красный ({brightness})")
|
||
|
||
elif state == "thinking":
|
||
# Радуга
|
||
led_controller.set_led_effect(EFFECT_RAINBOW)
|
||
led_controller.set_led_brightness(brightness)
|
||
led_controller.set_led_speed(4)
|
||
logger.info(f"🌈 Thinking: радуга ({brightness})")
|
||
|
||
elif state == "speaking":
|
||
# Зелёное дыхание
|
||
led_controller.set_led_effect(EFFECT_BREATH)
|
||
led_controller.set_led_color(rgb_to_bgr_int(76, 175, 80)) # 0x4CAF50
|
||
led_controller.set_led_brightness(brightness)
|
||
logger.info(f"🟢 Speaking: зелёный ({brightness})")
|
||
|
||
elif state == "idle":
|
||
# DoA эффект: яркость управляется отдельно через set_led_brightness
|
||
base_color = rgb_to_bgr_int(0x30, 0x30, 0x30) # 0x303030 - серый
|
||
doa_color = rgb_to_bgr_int(0x00, 0xBC, 0xD4) # 0xD4BC00 - бирюзовый (BGR)
|
||
led_controller.set_led_effect(EFFECT_DOA)
|
||
led_controller.set_led_doa_color(base_color, doa_color)
|
||
led_controller.set_led_brightness(brightness)
|
||
logger.info(f"⚪ Idle: DoA (brightness={brightness}, base=0x{base_color:06X}, doa=0x{doa_color:06X})")
|
||
|
||
elif state == "paused":
|
||
# Серое дыхание
|
||
led_controller.set_led_effect(EFFECT_BREATH)
|
||
led_controller.set_led_color(rgb_to_bgr_int(64, 64, 64)) # 0x404040
|
||
led_controller.set_led_brightness(brightness)
|
||
logger.info(f"⚫ Paused: серый ({brightness})")
|
||
|
||
elif state == "error":
|
||
# Маджента дыхание
|
||
led_controller.set_led_effect(EFFECT_BREATH)
|
||
led_controller.set_led_color(rgb_to_bgr_int(255, 0, 255)) # 0xFF00FF
|
||
led_controller.set_led_brightness(brightness)
|
||
logger.info(f"🟣 Error: маджента ({brightness})")
|
||
|
||
elif state == "off":
|
||
# Выключить
|
||
led_controller.set_led_effect(EFFECT_OFF)
|
||
logger.info("⬛ Off: выключено")
|
||
|
||
if not skip_publish:
|
||
publish_state(state)
|
||
publish_satellite_state(state)
|
||
|
||
|
||
# ====================== MQTT PUBLISH ======================
|
||
def publish_discovery_config():
|
||
if not mqtt_client:
|
||
return
|
||
try:
|
||
logger.info("🔄 Публикация Discovery...")
|
||
|
||
# Light
|
||
light_disc = {
|
||
"name": "Wyoming LED",
|
||
"unique_id": "wyoming_led_light",
|
||
"state_topic": MQTT_TOPIC_STATE,
|
||
"command_topic": MQTT_TOPIC_COMMAND,
|
||
"brightness_command_topic": MQTT_TOPIC_BRIGHTNESS_SET,
|
||
"brightness_state_topic": MQTT_TOPIC_BRIGHTNESS_STATE,
|
||
"availability_topic": MQTT_TOPIC_AVAILABILITY,
|
||
"payload_available": "online",
|
||
"payload_not_available": "offline",
|
||
"payload_on": "ON",
|
||
"payload_off": "OFF",
|
||
"brightness_scale": 100,
|
||
"optimistic": False,
|
||
"qos": 1,
|
||
"device": MQTT_DISCOVERY_DEVICE,
|
||
}
|
||
mqtt_client.publish(f"{MQTT_DISCOVERY_PREFIX}/light/wyoming_led/config",
|
||
json.dumps(light_disc), retain=True, qos=1)
|
||
|
||
# Sensor
|
||
sensor_disc = {
|
||
"name": "Wyoming Satellite State",
|
||
"unique_id": "wyoming_satellite_state",
|
||
"state_topic": MQTT_TOPIC_SATELLITE_STATE,
|
||
"availability_topic": MQTT_TOPIC_AVAILABILITY,
|
||
"payload_available": "online",
|
||
"payload_not_available": "offline",
|
||
"device": MQTT_DISCOVERY_DEVICE,
|
||
}
|
||
mqtt_client.publish(f"{MQTT_DISCOVERY_PREFIX}/sensor/wyoming_satellite_state/config",
|
||
json.dumps(sensor_disc), retain=True, qos=1)
|
||
|
||
# Number entities
|
||
publish_number_discovery()
|
||
|
||
logger.info("✅ Discovery опубликован успешно (Light + Sensor + 7 Number)")
|
||
# Состояния будут опубликованы с задержкой в on_mqtt_connect
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка Discovery: {e}", exc_info=True)
|
||
|
||
|
||
def publish_number_discovery():
|
||
if not mqtt_client:
|
||
return
|
||
try:
|
||
modes = {
|
||
"idle": "Idle", "listening": "Listening", "wake": "Wake",
|
||
"thinking": "Thinking", "speaking": "Speaking",
|
||
"paused": "Paused", "error": "Error"
|
||
}
|
||
for mode, name in modes.items():
|
||
disc = {
|
||
"name": f"{name}",
|
||
"unique_id": f"wyoming_led_{mode}_base",
|
||
"object_id": f"wyoming_led_{mode}_base",
|
||
"state_topic": f"wyoming/led/{mode}/state",
|
||
"command_topic": f"wyoming/led/{mode}/set",
|
||
"availability_topic": MQTT_TOPIC_AVAILABILITY,
|
||
"payload_available": "online",
|
||
"payload_not_available": "offline",
|
||
"min": 0, "max": 100, "step": 5,
|
||
"unit_of_measurement": "%",
|
||
|
||
"icon": "mdi:led-on",
|
||
"device": MQTT_DISCOVERY_DEVICE,
|
||
}
|
||
|
||
|
||
mqtt_client.publish(
|
||
f"{MQTT_DISCOVERY_PREFIX}/number/wyoming_led_{mode}/config",
|
||
json.dumps(disc), retain=True, qos=1
|
||
)
|
||
logger.info("✅ 7 Number entities опубликованы")
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка Number discovery: {e}")
|
||
|
||
|
||
def publish_state(state: str):
|
||
if not mqtt_client: return
|
||
ha_state = "OFF" if (ha_disabled or state == "off") else "ON"
|
||
multiplier = BRIGHTNESS_CONFIG.get("global_multiplier", 1.0)
|
||
ha_brightness = int(multiplier * 100)
|
||
|
||
mqtt_client.publish(MQTT_TOPIC_STATE, ha_state, retain=True, qos=1)
|
||
mqtt_client.publish(MQTT_TOPIC_BRIGHTNESS_STATE, str(ha_brightness), retain=True, qos=1)
|
||
|
||
|
||
def publish_satellite_state(state: str):
|
||
if mqtt_client:
|
||
mqtt_client.publish(MQTT_TOPIC_SATELLITE_STATE, state, retain=True, qos=0)
|
||
|
||
|
||
def publish_base_brightness_states():
|
||
if not mqtt_client: return
|
||
try:
|
||
logger.debug(f"📤 Brightness states: {BRIGHTNESS_CONFIG}")
|
||
for mode in ["idle","listening","wake","thinking","speaking","paused","error"]:
|
||
if mode in BRIGHTNESS_CONFIG:
|
||
percent = min(100, int(BRIGHTNESS_CONFIG[mode] / 2.55))
|
||
mqtt_client.publish(f"wyoming/led/{mode}/state", str(percent), retain=True, qos=1)
|
||
logger.debug(f"📤 {mode}/state = {percent}% (value={BRIGHTNESS_CONFIG[mode]})")
|
||
logger.info("📊 Brightness states опубликованы")
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка publish_base_brightness_states: {e}", exc_info=True)
|
||
|
||
|
||
# ====================== MQTT CALLBACKS ======================
|
||
def on_mqtt_connect(client, userdata, flags, rc, properties=None):
|
||
if rc == 0:
|
||
logger.info(f"✅ MQTT подключён к {MQTT_BROKER}:{MQTT_PORT}")
|
||
client.publish(MQTT_TOPIC_AVAILABILITY, "online", retain=True, qos=1)
|
||
client.subscribe(MQTT_TOPIC_COMMAND)
|
||
client.subscribe(MQTT_TOPIC_BRIGHTNESS_SET)
|
||
# Подписка на топики 7 режимов
|
||
for mode in ["idle","listening","wake","thinking","speaking","paused","error"]:
|
||
client.subscribe(f"wyoming/led/{mode}/set")
|
||
logger.debug(f"📡 Подписка на wyoming/led/{mode}/set")
|
||
publish_discovery_config()
|
||
publish_state(current_state)
|
||
publish_satellite_state(current_state)
|
||
publish_base_brightness_states()
|
||
logger.info(f"📡 Подписки: {MQTT_TOPIC_COMMAND}, {MQTT_TOPIC_BRIGHTNESS_SET}, 7 режимов")
|
||
else:
|
||
logger.error(f"❌ MQTT ошибка подключения: {rc}")
|
||
|
||
|
||
def on_mqtt_message(client, userdata, msg):
|
||
global ha_disabled, last_mqtt_brightness, current_state
|
||
try:
|
||
payload = msg.payload.decode('utf-8').strip()
|
||
topic = msg.topic
|
||
logger.info(f"📩 MQTT {topic}: {payload}")
|
||
|
||
# Глобальная яркость
|
||
if topic == MQTT_TOPIC_BRIGHTNESS_SET:
|
||
ha_value = max(0, min(100, int(payload)))
|
||
new_multiplier = ha_value / 100.0
|
||
last_mqtt_brightness = ha_value
|
||
BRIGHTNESS_CONFIG["global_multiplier"] = new_multiplier
|
||
save_brightness_config()
|
||
logger.info(f"🔆 Глобальная яркость: {ha_value}% (×{new_multiplier:.2f})")
|
||
if not ha_disabled:
|
||
set_led_state(current_state)
|
||
publish_base_brightness_states()
|
||
|
||
# Базовая яркость режима
|
||
elif topic.startswith("wyoming/led/") and topic.endswith("/set"):
|
||
mode = topic.split("/")[2]
|
||
logger.info(f"🎚️ Получена команда для режима: {mode}")
|
||
if mode in BRIGHTNESS_CONFIG:
|
||
percent = max(0, min(100, int(payload)))
|
||
real_value = int(percent * 2.55)
|
||
old_value = BRIGHTNESS_CONFIG[mode]
|
||
BRIGHTNESS_CONFIG[mode] = real_value
|
||
save_brightness_config()
|
||
logger.info(f"🎛️ {mode}: {old_value} → {real_value} ({percent}%)")
|
||
logger.info(f"📤 Публикация {mode}/state = {percent}")
|
||
publish_base_brightness_states()
|
||
if current_state == mode and not ha_disabled:
|
||
logger.info(f"🔄 Применяем новую яркость к текущему состоянию {mode}")
|
||
set_led_state(mode)
|
||
else:
|
||
logger.warning(f"⚠️ Неизвестный режим: {mode}")
|
||
|
||
# ON / OFF
|
||
elif topic == MQTT_TOPIC_COMMAND:
|
||
ha_state = payload.upper()
|
||
if ha_state == "OFF":
|
||
ha_disabled = True
|
||
if led_controller:
|
||
led_controller.set_led_effect(EFFECT_OFF)
|
||
publish_state("off")
|
||
logger.info("⬛ LED OFF")
|
||
elif ha_state == "ON":
|
||
ha_disabled = False
|
||
set_led_state(current_state)
|
||
logger.info("✅ LED ON")
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка MQTT: {e}", exc_info=True)
|
||
|
||
|
||
# ====================== MQTT CLIENT ======================
|
||
def start_mqtt_client():
|
||
global mqtt_client
|
||
mqtt_client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=MQTT_CLIENT_ID)
|
||
mqtt_client.username_pw_set(MQTT_USER, MQTT_PASS)
|
||
mqtt_client.on_connect = on_mqtt_connect
|
||
mqtt_client.on_message = on_mqtt_message
|
||
mqtt_client.will_set(MQTT_TOPIC_AVAILABILITY, "offline", retain=True, qos=1)
|
||
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
|
||
mqtt_client.loop_start()
|
||
|
||
|
||
# ====================== HANDLER & MAIN ======================
|
||
class LedEventHandler(AsyncEventHandler):
|
||
async def handle_event(self, event: Event) -> bool:
|
||
if ha_disabled:
|
||
return True
|
||
et = event.type
|
||
logger.debug(f"🔍 Событие: {et}")
|
||
if Detection.is_type(et):
|
||
logger.info("🎯 Detection → wake")
|
||
set_led_state("wake")
|
||
elif VoiceStarted.is_type(et):
|
||
logger.info("🎤 VoiceStarted → listening")
|
||
set_led_state("listening")
|
||
elif VoiceStopped.is_type(et):
|
||
logger.info("💭 VoiceStopped → thinking")
|
||
set_led_state("thinking")
|
||
elif et == "audio-start":
|
||
logger.info("🔊 audio-start → speaking")
|
||
set_led_state("speaking")
|
||
elif Played.is_type(et):
|
||
logger.info("✅ Played → idle")
|
||
set_led_state("idle")
|
||
elif et == "pause-satellite":
|
||
logger.info("⏸️ pause → paused")
|
||
set_led_state("paused")
|
||
elif SatelliteConnected.is_type(et):
|
||
logger.info("🔌 connected → idle")
|
||
set_led_state("idle")
|
||
return True
|
||
|
||
async def on_connected(self) -> None:
|
||
logger.info("✅ TCP подключён")
|
||
async def on_disconnected(self) -> None:
|
||
logger.info("❌ TCP отключён")
|
||
set_led_state("idle")
|
||
|
||
|
||
def cleanup_leds():
|
||
"""Очистка: выключаем LED и закрываем USB."""
|
||
logger.info("🧹 Очистка: выключаем LED...")
|
||
if led_controller:
|
||
try:
|
||
led_controller.set_led_effect(EFFECT_OFF)
|
||
led_controller.close()
|
||
logger.info("✅ LED выключены, USB закрыт")
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка очистки: {e}")
|
||
else:
|
||
logger.info("ℹ️ LED контроллер не был инициализирован")
|
||
|
||
|
||
async def main(args):
|
||
logger.info(f"🚀 Запуск на {args.host}:{args.port}")
|
||
load_brightness_config()
|
||
init_leds()
|
||
start_mqtt_client()
|
||
set_led_state("idle")
|
||
|
||
uri = f"tcp://{args.host}:{args.port}"
|
||
server = AsyncServer.from_uri(uri)
|
||
logger.info(f"📡 Слушаем {uri}")
|
||
await server.run(partial(LedEventHandler))
|
||
|
||
|
||
def signal_handler(sig, frame):
|
||
logger.info(f"\n🛑 Сигнал {sig}")
|
||
cleanup_leds()
|
||
if mqtt_client:
|
||
mqtt_client.publish(MQTT_TOPIC_AVAILABILITY, "offline", retain=True)
|
||
mqtt_client.loop_stop()
|
||
mqtt_client.disconnect()
|
||
logger.info("📡 MQTT остановлен")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--host", default="127.0.0.1")
|
||
parser.add_argument("--port", type=int, default=10701)
|
||
args = parser.parse_args()
|
||
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
signal.signal(signal.SIGTERM, signal_handler)
|
||
|
||
try:
|
||
asyncio.run(main(args))
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
cleanup_leds() |