#!/usr/bin/env python3 """ Wyoming LED Control - Простая надёжная версия для HA """ import argparse import asyncio import logging import subprocess 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 # ====================== НАСТРОЙКИ ====================== 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 XVF_HOST_PATH = str(SCRIPT_DIR / "xvf_host.py") 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" # ON/OFF 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 # ====================== CONFIG ====================== def load_brightness_config(): global BRIGHTNESS_CONFIG if CONFIG_FILE.exists(): try: with open(CONFIG_FILE) as f: BRIGHTNESS_CONFIG.update(json.load(f)) logger.info("✅ Загружены настройки яркости") 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) except Exception as e: logger.error(f"Ошибка сохранения config: {e}") # ====================== LED ====================== def run_xvf_command(command: str, values=None): args = [sys.executable, XVF_HOST_PATH, command] if values: args.extend(["--values"] + [str(v) for v in values]) try: subprocess.run(args, capture_output=True, text=True, check=True, timeout=5) return True except Exception as e: logger.error(f"LED error {command}: {e}") return False def get_effective_brightness(state: str) -> int: base = BRIGHTNESS_CONFIG.get(state, 100) return int(base * BRIGHTNESS_CONFIG.get("global_multiplier", 1.0)) def set_led_state(state: str, skip_publish=False): global current_state current_state = state brightness = get_effective_brightness(state) if state == "listening": run_xvf_command("LED_EFFECT", [1]) run_xvf_command("LED_COLOR", ["0x00BCD4"]) run_xvf_command("LED_BRIGHTNESS", [brightness]) run_xvf_command("LED_SPEED", [2]) elif state == "wake": run_xvf_command("LED_EFFECT", [1]) run_xvf_command("LED_COLOR", ["0xFF0000"]) run_xvf_command("LED_BRIGHTNESS", [brightness]) elif state == "thinking": run_xvf_command("LED_EFFECT", [2]) run_xvf_command("LED_BRIGHTNESS", [brightness]) run_xvf_command("LED_SPEED", [4]) elif state == "speaking": run_xvf_command("LED_EFFECT", [1]) run_xvf_command("LED_COLOR", ["0x4CAF50"]) run_xvf_command("LED_BRIGHTNESS", [brightness]) elif state == "idle": run_xvf_command("LED_EFFECT", [4]) run_xvf_command("LED_DOA_COLOR", ["0x303030", "0x00BCD4"]) run_xvf_command("LED_BRIGHTNESS", [brightness]) elif state == "paused": run_xvf_command("LED_EFFECT", [1]) run_xvf_command("LED_COLOR", ["0x404040"]) run_xvf_command("LED_BRIGHTNESS", [brightness]) elif state == "error": run_xvf_command("LED_EFFECT", [1]) run_xvf_command("LED_COLOR", ["0xFF00FF"]) run_xvf_command("LED_BRIGHTNESS", [brightness]) elif state == "off": run_xvf_command("LED_EFFECT", [0]) run_xvf_command("LED_BRIGHTNESS", [0]) if not skip_publish: publish_state(state, brightness) # ====================== MQTT ====================== def publish_discovery_config(): if not mqtt_client: return try: discovery = { "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, # ← ИЗМЕНИЛИ НА 100 "optimistic": False, "qos": 1, "device": MQTT_DISCOVERY_DEVICE, } mqtt_client.publish( f"{MQTT_DISCOVERY_PREFIX}/light/wyoming_led/config", json.dumps(discovery), retain=True, qos=1 ) logger.info("✅ Discovery опубликован (scale = 100)") except Exception as e: logger.error(f"Discovery error: {e}") def publish_state(state: str, brightness: int = None): """Публикуем в HA именно global_multiplier, а не effective brightness""" if not mqtt_client: return ha_state = "OFF" if (ha_disabled or state == "off") else "ON" # Главное: публикуем текущий global_multiplier в процентах (0-100) multiplier = BRIGHTNESS_CONFIG.get("global_multiplier", 1.0) ha_brightness = int(multiplier * 100) # 0..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) logger.info(f"📤 HA ← {ha_state} | Яркость: {ha_brightness}% (multiplier: {multiplier:.2f})") def on_mqtt_connect(client, userdata, flags, rc, properties=None): if rc == 0: logger.info("✅ MQTT подключён") client.publish(MQTT_TOPIC_AVAILABILITY, "online", retain=True, qos=1) client.subscribe(MQTT_TOPIC_COMMAND) client.subscribe(MQTT_TOPIC_BRIGHTNESS_SET) publish_discovery_config() # Отправляем текущее состояние brightness = get_effective_brightness(current_state) publish_state(current_state, brightness) 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() logger.info(f"📩 {msg.topic}: {payload}") # ==================== ЯРКОСТЬ ==================== if msg.topic == MQTT_TOPIC_BRIGHTNESS_SET: try: ha_value = max(0, min(100, int(payload))) # 0-100 new_multiplier = ha_value / 100.0 last_mqtt_brightness = ha_value BRIGHTNESS_CONFIG["global_multiplier"] = new_multiplier save_brightness_config() logger.info(f"🔆 HA установил яркость: {ha_value}% → multiplier = {new_multiplier:.2f}") if not ha_disabled: # Применяем новую яркость к текущему состоянию set_led_state(current_state, skip_publish=False) except ValueError: logger.warning(f"Неверное значение яркости: {payload}") # ==================== КОМАНДА (ON / OFF) ==================== elif msg.topic == MQTT_TOPIC_COMMAND: ha_state = payload.upper() if ha_state in ["ON", "OFF"]: if ha_state == "OFF": logger.info("🔴 HA → OFF") ha_disabled = True run_xvf_command("LED_EFFECT", [0]) run_xvf_command("LED_BRIGHTNESS", [0]) publish_state("off", 0) else: # ON logger.info("🟢 HA → ON") ha_disabled = False set_led_state(current_state) else: # Если вдруг пришёл JSON (редкий случай) try: cmd = json.loads(payload) ha_state = str(cmd.get("state", "")).upper() bright = cmd.get("brightness") if ha_state == "OFF": ha_disabled = True run_xvf_command("LED_EFFECT", [0]) run_xvf_command("LED_BRIGHTNESS", [0]) publish_state("off", 0) elif ha_state == "ON": ha_disabled = False if bright is not None: bright = max(0, min(255, int(bright))) last_mqtt_brightness = bright BRIGHTNESS_CONFIG["global_multiplier"] = bright / 255.0 save_brightness_config() set_led_state(current_state) except: logger.warning(f"Неизвестная команда: {payload}") except Exception as e: logger.error(f"❌ Ошибка обработки MQTT: {e}", exc_info=True) # ====================== START ====================== 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() class LedEventHandler(AsyncEventHandler): async def handle_event(self, event: Event) -> bool: if ha_disabled: return True et = event.type if Detection.is_type(et): set_led_state("wake") elif VoiceStarted.is_type(et): set_led_state("listening") elif VoiceStopped.is_type(et): set_led_state("thinking") elif et == "audio-start": set_led_state("speaking") elif Played.is_type(et): set_led_state("idle") elif et == "pause-satellite": set_led_state("paused") elif SatelliteConnected.is_type(et): set_led_state("idle") return True def cleanup_leds(): logger.info("🧹 Выключаем LED") run_xvf_command("LED_EFFECT", [0]) run_xvf_command("LED_BRIGHTNESS", [0]) async def main(args): logger.info("🚀 Запуск Wyoming LED Service") load_brightness_config() start_mqtt_client() run_xvf_command("VERSION") set_led_state("idle") uri = f"tcp://{args.host}:{args.port}" server = AsyncServer.from_uri(uri) await server.run(partial(LedEventHandler)) def signal_handler(sig, frame): logger.info(f"🛑 Сигнал {sig}") cleanup_leds() if mqtt_client: try: mqtt_client.publish(MQTT_TOPIC_AVAILABILITY, "offline", retain=True) mqtt_client.loop_stop() mqtt_client.disconnect() except: pass 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()