#!/usr/bin/env python3
"""Controls the LEDs on the ReSpeaker 2mic HAT, inspired by smart speakers like Alexa and Alice."""
import argparse
import asyncio
import logging
import time
from functools import partial
from math import ceil
from typing import Tuple, List, Optional
import gpiozero
import spidev
from wyoming.asr import Transcript
from wyoming.event import Event
from wyoming.satellite import (
    RunSatellite,
    SatelliteConnected,
    SatelliteDisconnected,
    StreamingStarted,
    StreamingStopped,
)
from wyoming.server import AsyncEventHandler, AsyncServer
from wyoming.vad import VoiceStarted, VoiceStopped
from wyoming.wake import Detection
_LOGGER = logging.getLogger()
NUM_LEDS = 3
LEDS_GPIO = 12
RGB_MAP = {
    "rgb": [3, 2, 1],
    "rbg": [3, 1, 2],
    "grb": [2, 3, 1],
    "gbr": [2, 1, 3],
    "brg": [1, 3, 2],
    "bgr": [1, 2, 3],
}
async def main() -> None:
    """Main entry point."""
    parser = argparse.ArgumentParser()
    parser.add_argument("--uri", required=True, help="unix:// or tcp://")
    parser.add_argument("--debug", action="store_true", help="Log DEBUG messages")
    parser.add_argument(
        "--led-brightness",
        type=int,
        choices=range(1, 32),
        default=31,
        help="LED brightness (integer from 1 to 31)",
    )
    args = parser.parse_args()
    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
    _LOGGER.debug(args)
    _LOGGER.info("Ready")
    # Turn on power to LEDs
    led_power = gpiozero.LED(LEDS_GPIO, active_high=False)
    led_power.on()
    leds = APA102(num_led=NUM_LEDS, global_brightness=args.led_brightness)
    # Start server
    server = AsyncServer.from_uri(args.uri)
    try:
        await server.run(partial(LEDsEventHandler, args, leds))
    except KeyboardInterrupt:
        pass
    finally:
        leds.cleanup()
        led_power.off()
# -----------------------------------------------------------------------------
_BLACK = (0, 0, 0)
_WHITE = (255, 255, 255)
_RED = (255, 0, 0)
_YELLOW = (255, 255, 0)
_BLUE = (0, 0, 255)
_GREEN = (0, 255, 0)
_PURPLE = (128, 0, 128)
_ORANGE = (255, 165, 0)
_CYAN = (0, 255, 255)
class LEDsEventHandler(AsyncEventHandler):
    """Event handler for clients, with animations inspired by Alexa and Alice."""
    def __init__(
        self,
        cli_args: argparse.Namespace,
        leds: "APA102",
        *args,
        **kwargs,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.cli_args = cli_args
        self.client_id = str(time.monotonic_ns())
        self.leds = leds
        self.voice_active = False
        self.voice_animation_task: Optional[asyncio.Task] = None
        self.streaming_active = False
        self.streaming_animation_task: Optional[asyncio.Task] = None
        _LOGGER.debug("Client connected: %s", self.client_id)
    async def handle_event(self, event: Event) -> bool:
        _LOGGER.debug(event)
        if StreamingStarted.is_type(event.type):
            self.streaming_active = True
            if self.streaming_animation_task is None or self.streaming_animation_task.done():
                self.streaming_animation_task = asyncio.create_task(self.streaming_animation_loop())
        elif Detection.is_type(event.type):
            # Как Alexa: directional blue spotlight for wake/listening
            await self.rotate_colors([_BLUE, _CYAN, _BLACK], rotations=5, delay=0.2)  # Вращение для "просыпания"
            self.color(_BLACK)
        elif VoiceStarted.is_type(event.type):
            self.voice_active = True
            if self.voice_animation_task is None or self.voice_animation_task.done():
                self.voice_animation_task = asyncio.create_task(self.voice_animation_loop())
        elif Transcript.is_type(event.type):
            # Как Alexa/Alice: Зеленый успех или эмоция (мигание зеленым для подтверждения)
            await self.pulse_color(_GREEN, pulses=2, duration=0.8)
            self.color(_BLACK)
        elif VoiceStopped.is_type(event.type):
            self.voice_active = False
            if self.voice_animation_task:
                self.voice_animation_task.cancel()
                self.voice_animation_task = None
            await self.fade_out(_BLUE, steps=10, delay=0.15)  # Затухание синего, как конец обработки
        elif StreamingStopped.is_type(event.type):
            self.streaming_active = False
            if self.streaming_animation_task:
                self.streaming_animation_task.cancel()
                self.streaming_animation_task = None
            await self.fade_out(_BLUE, steps=10, delay=0.15)
        elif RunSatellite.is_type(event.type):
            self.color(_BLACK)
        elif SatelliteConnected.is_type(event.type):
            # Разовая анимация при подключении, без loop, чтобы в idle не светилось
            await self.pulse_color(_GREEN, pulses=2, duration=1.5)
            self.color(_BLACK)
        elif SatelliteDisconnected.is_type(event.type):
            await self.pulse_color(_RED, pulses=3, duration=0.5)  # Красный пульс для ошибки, как Alexa purple/red
            self.color(_BLACK)
        return True
    async def voice_animation_loop(self) -> None:
        """Loop анимации для VoiceStarted: как Alexa listening/thinking - alternating blue/cyan."""
        try:
            while self.voice_active:
                await self.gradient([_BLUE, _CYAN, _BLUE], duration=1.5)  # Градиент для "thinking"
                await self.rotate_colors([_CYAN, _BLUE, _BLACK], rotations=1, delay=0.3)  # Вращение для активности
                await asyncio.sleep(0.2)
        except asyncio.CancelledError:
            self.color(_BLACK)
    async def streaming_animation_loop(self) -> None:
        """Loop для StreamingStarted: pulsing blue как Alexa processing."""
        try:
            while self.streaming_active:
                await self.pulse_color(_BLUE, pulses=1, duration=0.8)
                await asyncio.sleep(0.2)
        except asyncio.CancelledError:
            self.color(_BLACK)
    def color(self, rgb: Tuple[int, int, int]) -> None:
        for i in range(NUM_LEDS):
            self.leds.set_pixel(i, rgb[0], rgb[1], rgb[2])
        self.leds.show()
    def set_colors(self, colors: List[Tuple[int, int, int]]) -> None:
        for i in range(min(NUM_LEDS, len(colors))):
            self.leds.set_pixel(i, colors[i][0], colors[i][1], colors[i][2])
        self.leds.show()
    async def pulse_color(self, rgb: Tuple[int, int, int], pulses: int = 3, duration: float = 0.5) -> None:
        steps = 10
        for _ in range(pulses):
            for bright in range(steps):
                self.set_pixel_bright(rgb, (bright + 1) / steps * 100)
                await asyncio.sleep(duration / (2 * steps))
            for bright in range(steps, 0, -1):
                self.set_pixel_bright(rgb, (bright - 1) / steps * 100)
                await asyncio.sleep(duration / (2 * steps))
    def set_pixel_bright(self, rgb: Tuple[int, int, int], bright_percent: float) -> None:
        for i in range(NUM_LEDS):
            self.leds.set_pixel(i, rgb[0], rgb[1], rgb[2], bright_percent)
        self.leds.show()
    async def rainbow_cycle(self, steps: int = 20, delay: float = 0.05) -> None:
        for step in range(steps):
            colors = []
            for i in range(NUM_LEDS):
                r = int((step + i * 85) % 255)
                g = int((step + i * 85 + 85) % 255)
                b = int((step + i * 85 + 170) % 255)
                colors.append((r, g, b))
            self.set_colors(colors)
            await asyncio.sleep(delay)
        self.color(_BLACK)
    async def gradient(self, colors: List[Tuple[int, int, int]], duration: float = 1.0) -> None:
        steps = 10
        for step in range(steps):
            blended_colors = []
            for i in range(NUM_LEDS):
                idx = i % len(colors)
                next_idx = (idx + 1) % len(colors)
                r = int(colors[idx][0] + (colors[next_idx][0] - colors[idx][0]) * step / steps)
                g = int(colors[idx][1] + (colors[next_idx][1] - colors[idx][1]) * step / steps)
                b = int(colors[idx][2] + (colors[next_idx][2] - colors[idx][2]) * step / steps)
                blended_colors.append((r, g, b))
            self.set_colors(blended_colors)
            await asyncio.sleep(duration / steps)
    async def fade_out(self, start_rgb: Tuple[int, int, int], steps: int = 10, delay: float = 0.1) -> None:
        for step in range(steps, 0, -1):
            r = int(start_rgb[0] * step / steps)
            g = int(start_rgb[1] * step / steps)
            b = int(start_rgb[2] * step / steps)
            self.color((r, g, b))
            await asyncio.sleep(delay)
        self.color(_BLACK)
    async def rotate_colors(self, colors: List[Tuple[int, int, int]], rotations: int = 3, delay: float = 0.2) -> None:
        current_colors = colors[:NUM_LEDS]
        for _ in range(rotations):
            self.set_colors(current_colors)
            await asyncio.sleep(delay)
            current_colors = current_colors[-1:] + current_colors[:-1]
# -----------------------------------------------------------------------------
class APA102:
    """
    Driver for APA102 LEDS (aka "DotStar").
    (c) Martin Erzberger 2016-2017
    """
    # Constants
    MAX_BRIGHTNESS = 0b11111 # Safeguard: Set to a value appropriate for your setup
    LED_START = 0b11100000 # Three "1" bits, followed by 5 brightness bits
    def __init__(
        self,
        num_led,
        global_brightness,
        order="rgb",
        bus=0,
        device=1,
        max_speed_hz=8000000,
    ):
        self.num_led = num_led # The number of LEDs in the Strip
        order = order.lower()
        self.rgb = RGB_MAP.get(order, RGB_MAP["rgb"])
        # Limit the brightness to the maximum if it's set higher
        if global_brightness > self.MAX_BRIGHTNESS:
            self.global_brightness = self.MAX_BRIGHTNESS
        else:
            self.global_brightness = global_brightness
        _LOGGER.debug("LED brightness: %d", self.global_brightness)
        self.leds = [self.LED_START, 0, 0, 0] * self.num_led # Pixel buffer
        self.spi = spidev.SpiDev() # Init the SPI device
        self.spi.open(bus, device) # Open SPI port 0, slave device (CS) 1
        # Up the speed a bit, so that the LEDs are painted faster
        if max_speed_hz:
            self.spi.max_speed_hz = max_speed_hz
    def clock_start_frame(self):
        self.spi.xfer2([0] * 4) # Start frame, 32 zero bits
    def clock_end_frame(self):
        self.spi.xfer2([0xFF] * 4)
    def set_pixel(self, led_num, red, green, blue, bright_percent=100):
        if led_num < 0:
            return # Pixel is invisible, so ignore
        if led_num >= self.num_led:
            return # again, invisible
        brightness = int(ceil(bright_percent * self.global_brightness / 100.0))
        ledstart = (brightness & 0b00011111) | self.LED_START
        start_index = 4 * led_num
        self.leds[start_index] = ledstart
        self.leds[start_index + self.rgb[0]] = red
        self.leds[start_index + self.rgb[1]] = green
        self.leds[start_index + self.rgb[2]] = blue
    def set_pixel_rgb(self, led_num, rgb_color, bright_percent=100):
        self.set_pixel(
            led_num,
            (rgb_color & 0xFF0000) >> 16,
            (rgb_color & 0x00FF00) >> 8,
            rgb_color & 0x0000FF,
            bright_percent,
        )
    def rotate(self, positions=1):
        cutoff = 4 * (positions % self.num_led)
        self.leds = self.leds[cutoff:] + self.leds[:cutoff]
    def show(self):
        self.clock_start_frame()
        data = list(self.leds)
        while data:
            self.spi.xfer2(data[:32])
            data = data[32:]
        self.clock_end_frame()
    def cleanup(self):
        self.spi.close() # Close SPI port
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass