From e7c9b58ef4ba483981ad32a60aa21613ed0162cd Mon Sep 17 00:00:00 2001 From: imonlinux <39863321+imonlinux@users.noreply.github.com> Date: Sat, 11 Oct 2025 12:42:36 -0700 Subject: [PATCH] Replaced by led_controller.py --- .../event_led_grove.py_marvin | 156 ------------------ 1 file changed, 156 deletions(-) delete mode 100644 linux_voice_assistant/event_led_grove.py_marvin diff --git a/linux_voice_assistant/event_led_grove.py_marvin b/linux_voice_assistant/event_led_grove.py_marvin deleted file mode 100644 index 0258aa9..0000000 --- a/linux_voice_assistant/event_led_grove.py_marvin +++ /dev/null @@ -1,156 +0,0 @@ -# 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)