linux-voice-assistant/linux_voice_assistant/event_led_grove.py_marvin

# event_led_marvin.py
# Based on the corrected HAL file, with red changed to green.

import logging
import asyncio
from typing import Any, Tuple, Optional

import board
import adafruit_dotstar
from .event_bus import EventHandler, subscribe

_LOGGER = logging.getLogger(__name__)

# --- Configuration ---
# Set this to the number of APA102 LEDs you have connected to the Grove port.
NUM_LEDS = 2
GLOBAL_BRIGHTNESS = 1.0 # Brightness as a value from 0.0 to 1.0

# --- Color and Timing Constants ---
_OFF = (0, 0, 0)
_GREEN = (0, 255, 0) # Changed from RED to GREEN
LISTEN_IDLE_TIMEOUT_SEC = 15.0


class LedEvent(EventHandler):
    def __init__(self, state):
        super().__init__(state)
        self.loop = self.state.loop
        self.current_task: Optional[asyncio.Future] = None

        try:
            self.leds = adafruit_dotstar.DotStar(
                clock=board.D13,
                data=board.D12,
                n=NUM_LEDS,
                brightness=GLOBAL_BRIGHTNESS,
                auto_write=False
            )
            _LOGGER.info(f"Initialized {NUM_LEDS} APA102 LEDs on GPIO12/13.")
        except Exception as e:
            _LOGGER.error(f"Failed to initialize DotStar LEDs: {e}")
            self.leds = None

        self._base_color = _GREEN # Changed from _RED
        self._listen_idle_watchdog: Optional[asyncio.Task] = None
        self.run_action("color", _OFF)

    def run_action(self, action_method_name: str, *args: Any):
        if not self.leds:
            return

        if self.current_task and not self.current_task.done():
            self.current_task.cancel()
        self.current_task = asyncio.run_coroutine_threadsafe(
            getattr(self, action_method_name)(*args), self.loop
        )

    # --- LED Action Methods ---

    async def color(self, rgb: Tuple[int, int, int]):
        self.leds.brightness = GLOBAL_BRIGHTNESS
        self.leds.fill(rgb)
        self.leds.show()

    async def pulse(self, color: Tuple[int, int, int], speed=0.02):
        try:
            while True:
                # Fade in
                for i in range(101):
                    self.leds.brightness = (i / 100.0) * GLOBAL_BRIGHTNESS
                    self.leds.fill(color)
                    self.leds.show()
                    await asyncio.sleep(speed)

                # Fade out
                for i in range(100, -1, -1):
                    self.leds.brightness = (i / 100.0) * GLOBAL_BRIGHTNESS
                    self.leds.fill(color)
                    self.leds.show()
                    await asyncio.sleep(speed)
        finally:
            self.leds.brightness = GLOBAL_BRIGHTNESS

    async def flash(self, flash_color, times, on_time=0.1, off_time=0.1):
        for _ in range(times):
            await self.color(flash_color)
            await asyncio.sleep(on_time)
            await self.color(_OFF)
            await asyncio.sleep(off_time)

    async def flash_then_pulse(self, flash_color, times, pulse_color, pulse_speed=0.02):
        await self.flash(flash_color, times, on_time=0.05, off_time=0.05)
        await self.pulse(pulse_color, pulse_speed)


    # --- Watchdog Logic ---

    def _cancel_watchdog(self):
        if self._listen_idle_watchdog and not self._listen_idle_watchdog.done():
            self._listen_idle_watchdog.cancel()
        self._listen_idle_watchdog = None

    async def _listen_watchdog(self, timeout: float):
        try:
            await asyncio.sleep(timeout)
            _LOGGER.debug("Listen idle watchdog timeout -> return to base color")
            self.run_action("color", self._base_color)
        except asyncio.CancelledError:
            pass

    # --- Event Subscriptions ---

    @subscribe
    def ready(self, data: dict):
        _LOGGER.debug("ready -> solid GREEN")
        self._cancel_watchdog()
        self.run_action("color", self._base_color)

    @subscribe
    def disconnected(self, data: dict):
        _LOGGER.debug("disconnected -> LEDs OFF")
        self._cancel_watchdog()
        self.run_action("color", _OFF)

    @subscribe
    def voice_wakeword(self, data: dict):
        _LOGGER.debug("wakeword -> fast GREEN flash x2, then enter listening (pulse)")
        self._cancel_watchdog()
        self._listen_idle_watchdog = self.state.loop.create_task(
            self._listen_watchdog(LISTEN_IDLE_TIMEOUT_SEC)
        )
        self.run_action("flash_then_pulse", _GREEN, 2, _GREEN, 0.02) # Changed from _RED

    @subscribe
    def voice_VOICE_ASSISTANT_STT_START(self, data: dict):
        _LOGGER.debug("VAD_START -> ensure GREEN pulsing (listening)")
        self._cancel_watchdog()
        self.run_action("pulse", _GREEN, 0.02) # Changed from _RED

    @subscribe
    def voice_VOICE_ASSISTANT_STT_VAD_END(self, data: dict):
        _LOGGER.debug("VAD_END -> stop pulsing -> solid GREEN")
        self._cancel_watchdog()
        self.run_action("color", self._base_color)

    @subscribe
    def voice_play_tts(self, data: dict):
        _LOGGER.debug("TTS -> solid GREEN")
        self._cancel_watchdog()
        self.run_action("color", self._base_color)

    @subscribe
    def voice__tts_finished(self, data: dict):
        _LOGGER.debug("TTS finished -> solid GREEN")
        self._cancel_watchdog()
        self.run_action("color", self._base_color)