satellite/examples/wyoming-led.py

#!/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 эффект: яркость управляется через значения цвета
        base_rgb = scale_color((0x30, 0x30, 0x30), brightness)
        doa_rgb = scale_color((0x00, 0xBC, 0xD4), brightness)
        base_color = rgb_to_bgr_int(*base_rgb)
        doa_color = rgb_to_bgr_int(*doa_rgb)
        led_controller.set_led_effect(EFFECT_DOA)
        led_controller.set_led_doa_color(base_color, doa_color)
        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()