Add support for Grove Interface

Add Grove interface support for APA102 LEDs instead of the built in spi.

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)

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)

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)