diff --git a/linux_voice_assistant/event_led_grove.py b/linux_voice_assistant/event_led_grove.py new file mode 100644 index 0000000..b3d855c --- /dev/null +++ b/linux_voice_assistant/event_led_grove.py @@ -0,0 +1,128 @@ +# MODIFIED event_led.py + +import logging +import asyncio +from typing import Tuple, Any + +# --- MODIFICATION: Import new libraries and remove unused ones --- +import board +import adafruit_dotstar +from .event_bus import EventHandler, subscribe + +_LOGGER = logging.getLogger(__name__) + +# --- MODIFICATION: Make NUM_LEDS a configurable variable at the top --- +# 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 + +# Define some color constants +_OFF = (0, 0, 0) +_WHITE = (255, 255, 255) +_RED = (255, 0, 0) +_YELLOW = (255, 255, 0) +_BLUE = (0, 0, 255) +_GREEN = (0, 255, 0) + +# --- MODIFICATION: The old spidev-based APA102 class has been removed entirely --- + +class LedEvent(EventHandler): + def __init__(self, state): + super().__init__(state) + self.loop = self.state.loop + self.current_task = None + + # --- MODIFICATION: Initialize LEDs using Adafruit DotStar and GPIO pins --- + # The ReSpeaker 2-Mic HAT Grove connector uses GPIO13 (Clock) and GPIO12 (Data) + # If your wiring is different, change board.D13 and board.D12 accordingly. + try: + self.leds = adafruit_dotstar.DotStar( + clock=board.D13, + data=board.D12, + n=NUM_LEDS, + brightness=GLOBAL_BRIGHTNESS, + auto_write=False # We will call show() manually + ) + _LOGGER.info(f"Initialized {NUM_LEDS} APA102 LEDs on GPIO12/13.") + except Exception as e: + _LOGGER.error(f"Failed to initialize DotStar LEDs: {e}") + _LOGGER.error("Please ensure you have run 'pip install adafruit-circuitpython-dotstar' and that SPI is enabled if prompted.") + self.leds = None # Prevent crashes if initialization fails + + + def run_action(self, action_method_name: str, *args: Any) -> None: + if not self.leds: + _LOGGER.warning("LEDs not initialized, skipping action.") + 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) + + # --- MODIFICATION: Rewritten 'color' method to work with the new library --- + async def color(self, rgb: Tuple[int, int, int], brightness_override: float = None) -> None: + if brightness_override is not None: + self.leds.brightness = brightness_override + else: + self.leds.brightness = GLOBAL_BRIGHTNESS + + self.leds.fill(rgb) + self.leds.show() + + # --- MODIFICATION: Rewritten 'blink' method to be simpler --- + async def blink(self, rgb: Tuple[int, int, int], count=10000): + try: + for _ in range(count): + self.leds.fill(rgb) + self.leds.show() + await asyncio.sleep(0.3) + self.leds.fill(_OFF) + self.leds.show() + await asyncio.sleep(0.3) + except asyncio.CancelledError: + pass # Task was cancelled, which is normal + + # --- MODIFICATION: Rewritten 'pulse' method for the new library --- + async def pulse(self, color: Tuple[int, int, int], speed: float = 0.015): + try: + while True: + # Fade in + for i in range(101): + self.leds.brightness = i / 100.0 + 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 + self.leds.fill(color) + self.leds.show() + await asyncio.sleep(speed) + except asyncio.CancelledError: + self.leds.brightness = GLOBAL_BRIGHTNESS # Restore brightness on exit + + + # --- Event subscriptions remain the same --- + @subscribe + def ready(self, data: dict): + _LOGGER.debug('ready LED green blink') + # Blink 3 times and then turn off + self.run_action("blink", _GREEN, 3) + + @subscribe + def voice_wakeword(self, data: dict): + self.run_action("pulse", _BLUE) + + @subscribe + def voice_VOICE_ASSISTANT_STT_VAD_END(self, data: dict): + self.run_action("pulse", _YELLOW) + + @subscribe + def voice_play_tts(self, data: dict): + self.run_action("pulse", _GREEN) + + @subscribe + def voice__tts_finished(self, data: dict): + self.run_action("color", _OFF) diff --git a/linux_voice_assistant/event_led_grove.py_hal b/linux_voice_assistant/event_led_grove.py_hal new file mode 100644 index 0000000..bd84a4d --- /dev/null +++ b/linux_voice_assistant/event_led_grove.py_hal @@ -0,0 +1,158 @@ +# MODIFIED event_led_hal.py (v2 - with cancellation fix) + +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 = 1 +GLOBAL_BRIGHTNESS = 1.0 # Brightness as a value from 0.0 to 1.0 + +# --- Color and Timing Constants --- +_OFF = (0, 0, 0) +_RED = (255, 0, 0) +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 = _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() + + # --- MODIFICATION: This function is now more robust to cancellation --- + 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: + # This 'finally' block ensures brightness is restored even + # when the task is cancelled. + 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 RED") + 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 RED 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", _RED, 2, _RED, 0.02) + + @subscribe + def voice_VOICE_ASSISTANT_STT_START(self, data: dict): + _LOGGER.debug("VAD_START -> ensure RED pulsing (listening)") + self._cancel_watchdog() + self.run_action("pulse", _RED, 0.02) + + @subscribe + def voice_VOICE_ASSISTANT_STT_VAD_END(self, data: dict): + _LOGGER.debug("VAD_END -> stop pulsing -> solid RED") + self._cancel_watchdog() + self.run_action("color", self._base_color) + + @subscribe + def voice_play_tts(self, data: dict): + _LOGGER.debug("TTS -> solid RED") + self._cancel_watchdog() + self.run_action("color", self._base_color) + + @subscribe + def voice__tts_finished(self, data: dict): + _LOGGER.debug("TTS finished -> solid RED") + self._cancel_watchdog() + self.run_action("color", self._base_color) diff --git a/linux_voice_assistant/event_led_grove.py_marvin b/linux_voice_assistant/event_led_grove.py_marvin new file mode 100644 index 0000000..0258aa9 --- /dev/null +++ b/linux_voice_assistant/event_led_grove.py_marvin @@ -0,0 +1,156 @@ +# 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)