From 87a7cfa7b23bfeb8ac90abc30a55535337265a7f Mon Sep 17 00:00:00 2001 From: imonlinux <39863321+imonlinux@users.noreply.github.com> Date: Fri, 10 Oct 2025 18:45:40 -0700 Subject: [PATCH] Add files via upload --- linux_voice_assistant/event_bus.py | 67 +++++ linux_voice_assistant/event_led.py | 262 ++++++++++++++++++ linux_voice_assistant/event_led_grove.py | 128 +++++++++ linux_voice_assistant/event_led_grove.py_hal | 158 +++++++++++ .../event_led_grove.py_marvin | 156 +++++++++++ linux_voice_assistant/event_led_hal.py | 188 +++++++++++++ linux_voice_assistant/event_led_marvin.py | 198 +++++++++++++ 7 files changed, 1157 insertions(+) create mode 100644 linux_voice_assistant/event_bus.py create mode 100644 linux_voice_assistant/event_led.py create mode 100644 linux_voice_assistant/event_led_grove.py create mode 100644 linux_voice_assistant/event_led_grove.py_hal create mode 100644 linux_voice_assistant/event_led_grove.py_marvin create mode 100644 linux_voice_assistant/event_led_hal.py create mode 100644 linux_voice_assistant/event_led_marvin.py diff --git a/linux_voice_assistant/event_bus.py b/linux_voice_assistant/event_bus.py new file mode 100644 index 0000000..8ce5eca --- /dev/null +++ b/linux_voice_assistant/event_bus.py @@ -0,0 +1,67 @@ +import logging +from typing import Any, Callable, Dict, List, Optional + +_LOGGER = logging.getLogger(__name__) + + +class EventBus: + """A simple synchronous publish/subscribe event bus.""" + + def __init__(self): + # A dictionary to hold listeners for specific string topics + self.topics: Dict[str, List[Callable[[Any], None]]] = {} + + def subscribe(self, topic: str, listener: Callable[[Any], None]) -> None: + """ + Subscribes a listener to a topic. + """ + + # _LOGGER.debug(f'EventBus subscribe {topic}') + + if topic not in self.topics: + self.topics[topic] = [] + self.topics[topic].append(listener) + + def publish(self, topic: str, data: [dict, None]) -> None: + """ + Publishes an event to all subscribed listeners. + """ + + # _LOGGER.debug(f'EventBus publish {topic}') + + data['__topic'] = topic + + listeners = self.topics.get(topic, []) + for listener in listeners: + listener(data) + +# Client helpers for subscriptions + +# The decorator to mark methods for subscription. +def subscribe(func: Callable) -> Callable: + """Decorator to mark a method for event bus subscription.""" + func._event_bus_subscribe = True + return func + +class EventHandler: + """ + A base class for components that subscribe to events. + + Subclasses should define event handlers as methods decorated with `@subscribe`. + The method name will automatically be used as the event topic. + """ + + def __init__(self, state: Any): + self.state = state + self._subscribe_all_methods() + _LOGGER.debug(f"EventHandler {self.__class__.__name__} has subscribed to all decorated methods.") + + def _subscribe_all_methods(self): + """Finds and subscribes all methods decorated with @subscribe.""" + for method_name in dir(self): + method = getattr(self, method_name) + + if hasattr(method, '_event_bus_subscribe'): + # The topic is the name of the method itself. + self.state.event_bus.subscribe(method_name, method) + _LOGGER.debug(f"Subscribed method '{method_name}' to topic '{method_name}'") \ No newline at end of file diff --git a/linux_voice_assistant/event_led.py b/linux_voice_assistant/event_led.py new file mode 100644 index 0000000..651a34a --- /dev/null +++ b/linux_voice_assistant/event_led.py @@ -0,0 +1,262 @@ +import logging +from typing import Any, Callable + +from .event_bus import EventHandler, subscribe + +_LOGGER = logging.getLogger(__name__) + + + +"""Controls the LEDs on the ReSpeaker 2mic HAT.""" +from math import ceil +from typing import Tuple + +import time +import asyncio +import gpiozero +import spidev + + +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], +} + + + +_OFF = (0, 0, 0) +_WHITE = (255, 255, 255) +_RED = (255, 0, 0) +_YELLOW = (255, 255, 0) +_BLUE = (0, 0, 255) +_GREEN = (0, 255, 0) + +# ----------------------------------------------------------------------------- + + +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, + loop: asyncio.AbstractEventLoop, + 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 + print("LED brightness:", 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 + + self.current_task = None + self.loop = loop + + + def clock_start_frame(self): + """Sends a start frame to the LED strip. + + This method clocks out a start frame, telling the receiving LED + that it must update its own color now. + """ + self.spi.xfer2([0] * 4) # Start frame, 32 zero bits + + def clock_end_frame(self): + """Sends an end frame to the LED strip. + + As explained above, dummy data must be sent after the last real colour + information so that all of the data can reach its destination down the line. + The delay is not as bad as with the human example above. + It is only 1/2 bit per LED. This is because the SPI clock line + needs to be inverted. + + Say a bit is ready on the SPI data line. The sender communicates + this by toggling the clock line. The bit is read by the LED + and immediately forwarded to the output data line. When the clock goes + down again on the input side, the LED will toggle the clock up + on the output to tell the next LED that the bit is ready. + + After one LED the clock is inverted, and after two LEDs it is in sync + again, but one cycle behind. Therefore, for every two LEDs, one bit + of delay gets accumulated. For 300 LEDs, 150 additional bits must be fed to + the input of LED one so that the data can reach the last LED. + + Ultimately, we need to send additional numLEDs/2 arbitrary data bits, + in order to trigger numLEDs/2 additional clock changes. This driver + sends zeroes, which has the benefit of getting LED one partially or + fully ready for the next update to the strip. An optimized version + of the driver could omit the "clockStartFrame" method if enough zeroes have + been sent as part of "clockEndFrame". + """ + + self.spi.xfer2([0xFF] * 4) + + # Round up num_led/2 bits (or num_led/16 bytes) + # for _ in range((self.num_led + 15) // 16): + # self.spi.xfer2([0x00]) + + def set_pixel(self, led_num, red, green, blue, bright_percent=100): + """Sets the color of one pixel in the LED stripe. + + The changed pixel is not shown yet on the Stripe, it is only + written to the pixel buffer. Colors are passed individually. + If brightness is not set the global brightness setting is used. + """ + if led_num < 0: + return # Pixel is invisible, so ignore + if led_num >= self.num_led: + return # again, invisible + + # Calculate pixel brightness as a percentage of the + # defined global_brightness. Round up to nearest integer + # as we expect some brightness unless set to 0 + brightness = int(ceil(bright_percent * self.global_brightness / 100.0)) + + # LED startframe is three "1" bits, followed by 5 brightness bits + 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): + """Sets the color of one pixel in the LED stripe. + + The changed pixel is not shown yet on the Stripe, it is only + written to the pixel buffer. + Colors are passed combined (3 bytes concatenated) + If brightness is not set the global brightness setting is used. + """ + self.set_pixel( + led_num, + (rgb_color & 0xFF0000) >> 16, + (rgb_color & 0x00FF00) >> 8, + rgb_color & 0x0000FF, + bright_percent or self.global_brightness, + ) + + def rotate(self, positions=1): + """Rotate the LEDs by the specified number of positions. + + Treating the internal LED array as a circular buffer, rotate it by + the specified number of positions. The number could be negative, + which means rotating in the opposite direction. + """ + cutoff = 4 * (positions % self.num_led) + self.leds = self.leds[cutoff:] + self.leds[:cutoff] + + def show(self): + """Sends the content of the pixel buffer to the strip. + + Todo: More than 1024 LEDs requires more than one xfer operation. + """ + self.clock_start_frame() + # xfer2 kills the list, unfortunately. So it must be copied first + # SPI takes up to 4096 Integers. So we are fine for up to 1024 LEDs. + data = list(self.leds) + while data: + self.spi.xfer2(data[:32]) + data = data[32:] + self.clock_end_frame() + + def cleanup(self): + """Release the SPI device; Call this method at the end""" + + self.spi.close() # Close SPI port + + + def run_action(self, action_mentod_name: str, *args: Any) -> None: + if self.current_task and not self.current_task.done(): + self.current_task.cancel() + + self.current_task = asyncio.run_coroutine_threadsafe(getattr(self, action_mentod_name)(*args), self.loop) + + async def color(self, rgb: Tuple[int, int, int], brightness = None) -> None: + for i in range(NUM_LEDS): + self.set_pixel(i, rgb[0], rgb[1], rgb[2], brightness or self.global_brightness) + + self.show() + + async def blink(self, color, count=10000): + for _ in range(count): + await self.color(color) + await asyncio.sleep(0.3) + await self.color(_OFF) + await asyncio.sleep(0.3) + + async def pulse(self, color: Tuple[int, int, int], speed: float = 0.009): + """Asynchronously pulses the LEDs from off to full brightness and back.""" + # Fade in + while(1): + for brightness in range(1, self.global_brightness+1): + await self.color(color, brightness) + await asyncio.sleep(speed) + + # Fade out + for brightness in range(self.global_brightness+1, 0, -1): + await self.color(color, brightness) + await asyncio.sleep(speed) + + + + +class LedEvent(EventHandler): + def __init__(self, state): + super().__init__(state) + self.leds = APA102(num_led=3, global_brightness=31, loop=self.state.loop) + + @subscribe + def ready(self, data: dict): + _LOGGER.debug('ready LED green blink') + self.leds.run_action("blink", _GREEN, 3) + + @subscribe + def voice_wakeword(self, data: dict): + self.leds.run_action("pulse", _BLUE) + + @subscribe + def voice_VOICE_ASSISTANT_STT_VAD_END(self, data: dict): + self.leds.run_action("pulse", _YELLOW) + + @subscribe + def voice_play_tts(self, data: dict): + self.leds.run_action("pulse", _GREEN) + + # This event fires long before the TTS is done speaking + # @subscribe + # def voice_VOICE_ASSISTANT_RUN_END(self, data: dict): + # self.leds.run_action("color", _OFF, 0) + + @subscribe + def voice__tts_finished(self, data: dict): + self.leds.run_action("color", _OFF, 0) 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) diff --git a/linux_voice_assistant/event_led_hal.py b/linux_voice_assistant/event_led_hal.py new file mode 100644 index 0000000..d8e8103 --- /dev/null +++ b/linux_voice_assistant/event_led_hal.py @@ -0,0 +1,188 @@ + +import logging +from typing import Any, Tuple, Optional + +from .event_bus import EventHandler, subscribe + +_LOGGER = logging.getLogger(__name__) + +from math import ceil +import asyncio +import spidev + +NUM_LEDS = 3 +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], +} + +_OFF = (0, 0, 0) +_RED = (255, 0, 0) + +# Keep listening pulse alive for up to 15s after wakeword if user doesn't speak +LISTEN_IDLE_TIMEOUT_SEC = 15.0 + + +class APA102: + MAX_BRIGHTNESS = 0b11111 + LED_START = 0b11100000 + + def __init__( + self, + num_led: int, + global_brightness: int, + loop: asyncio.AbstractEventLoop, + order: str = "rgb", + bus: int = 0, + device: int = 1, + max_speed_hz: int = 8000000, + ): + self.num_led = num_led + order = order.lower() + self.rgb = RGB_MAP.get(order, RGB_MAP["rgb"]) + self.global_brightness = min(global_brightness, self.MAX_BRIGHTNESS) + _LOGGER.debug("LED brightness: %s", self.global_brightness) + + self.leds = [self.LED_START, 0, 0, 0] * self.num_led + self.spi = spidev.SpiDev() + self.spi.open(bus, device) + if max_speed_hz: + self.spi.max_speed_hz = max_speed_hz + + self.current_task: Optional[asyncio.Future] = None + self.loop = loop + + def clock_start_frame(self): + self.spi.xfer2([0] * 4) + + 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 or led_num >= self.num_led: + return + 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 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() + + def run_action(self, action_method_name: str, *args: Any): + 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 + ) + + async def color(self, rgb, brightness=None): + for i in range(NUM_LEDS): + self.set_pixel(i, rgb[0], rgb[1], rgb[2], brightness or self.global_brightness) + self.show() + + async def pulse(self, color, speed=0.02): + while True: + for brightness in range(1, self.MAX_BRIGHTNESS + 1): + await self.color(color, brightness) + await asyncio.sleep(speed) + for brightness in range(self.MAX_BRIGHTNESS, 0, -1): + await self.color(color, brightness) + await asyncio.sleep(speed) + + 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) + await self.pulse(pulse_color, pulse_speed) + + +class LedEvent(EventHandler): + def __init__(self, state): + super().__init__(state) + self.leds = APA102(num_led=3, global_brightness=31, loop=self.state.loop) + self._base_color = _RED + self.leds.run_action("color", _OFF) + self._listen_idle_watchdog: Optional[asyncio.Task] = None + + 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.leds.run_action("color", self._base_color) + except asyncio.CancelledError: + pass + + # Connectivity ------------------------------------------------------------------ + @subscribe + def ready(self, data: dict): + _LOGGER.debug("ready -> solid RED") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) + + @subscribe + def disconnected(self, data: dict): + _LOGGER.debug("disconnected -> LEDs OFF") + self._cancel_watchdog() + self.leds.run_action("color", _OFF) + + # Wakeword ---------------------------------------------------------------------- + @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.leds.run_action("flash_then_pulse", _RED, 2, _RED, 0.02) + + # Listening (VAD) --------------------------------------------------------------- + @subscribe + def voice_VOICE_ASSISTANT_STT_START(self, data: dict): + _LOGGER.debug("VAD_START -> ensure RED pulsing (listening)") + self._cancel_watchdog() + self.leds.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.leds.run_action("color", self._base_color) + + # TTS (speaking) ---------------------------------------------------------------- + @subscribe + def voice_play_tts(self, data: dict): + _LOGGER.debug("TTS -> solid RED") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) + + @subscribe + def voice__tts_finished(self, data: dict): + _LOGGER.debug("TTS finished -> solid RED") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) diff --git a/linux_voice_assistant/event_led_marvin.py b/linux_voice_assistant/event_led_marvin.py new file mode 100644 index 0000000..08ee6ad --- /dev/null +++ b/linux_voice_assistant/event_led_marvin.py @@ -0,0 +1,198 @@ + +import logging +from typing import Any, Tuple, Optional + +from .event_bus import EventHandler, subscribe + +_LOGGER = logging.getLogger(__name__) + +from math import ceil +import asyncio +import spidev + +NUM_LEDS = 2 +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], +} + +_OFF = (0, 0, 0) +_GREEN = (0, 255, 0) + +# How long to keep "listening" after wakeword if user never starts speaking +LISTEN_IDLE_TIMEOUT_SEC = 15.0 + + +class APA102: + MAX_BRIGHTNESS = 0b11111 + LED_START = 0b11100000 + + def __init__( + self, + num_led: int, + global_brightness: int, + loop: asyncio.AbstractEventLoop, + order: str = "rgb", + bus: int = 0, + device: int = 1, + max_speed_hz: int = 8000000, + ): + self.num_led = num_led + order = order.lower() + self.rgb = RGB_MAP.get(order, RGB_MAP["rgb"]) + self.global_brightness = min(global_brightness, self.MAX_BRIGHTNESS) + _LOGGER.debug("LED brightness: %s", self.global_brightness) + + self.leds = [self.LED_START, 0, 0, 0] * self.num_led + self.spi = spidev.SpiDev() + self.spi.open(bus, device) + if max_speed_hz: + self.spi.max_speed_hz = max_speed_hz + + self.current_task: Optional[asyncio.Future] = None + self.loop = loop + + def clock_start_frame(self): + self.spi.xfer2([0] * 4) + + 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 or led_num >= self.num_led: + return + 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 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() + + def run_action(self, action_method_name: str, *args: Any): + # Cancel any currently running animation task (pulse/blink/etc.) + 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 + ) + + async def color(self, rgb, brightness=None): + for i in range(NUM_LEDS): + self.set_pixel(i, rgb[0], rgb[1], rgb[2], brightness or self.global_brightness) + self.show() + + async def pulse(self, color, speed=0.02): + """Pulse continuously until cancelled.""" + while True: + for brightness in range(1, self.MAX_BRIGHTNESS + 1): + await self.color(color, brightness) + await asyncio.sleep(speed) + for brightness in range(self.MAX_BRIGHTNESS, 0, -1): + await self.color(color, brightness) + await asyncio.sleep(speed) + + async def flash(self, flash_color, times, on_time=0.1, off_time=0.1): + """Flash color on/off N times, then leave LEDs OFF (caller decides next state).""" + 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): + """Flash N times, then enter continuous pulse.""" + await self.flash(flash_color, times) + await self.pulse(pulse_color, pulse_speed) + + +class LedEvent(EventHandler): + def __init__(self, state): + super().__init__(state) + self.leds = APA102(num_led=3, global_brightness=31, loop=self.state.loop) + self._base_color = _GREEN + # Keep LEDs off until connected + self.leds.run_action("color", _OFF) + # Watchdog for the "no speech after wakeword" case + self._listen_idle_watchdog: Optional[asyncio.Task] = None + + 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.leds.run_action("color", self._base_color) + except asyncio.CancelledError: + pass + + # Connectivity ------------------------------------------------------------------ + @subscribe + def ready(self, data: dict): + _LOGGER.debug("ready -> solid GREEN") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) + + @subscribe + def disconnected(self, data: dict): + _LOGGER.debug("disconnected -> LEDs OFF") + self._cancel_watchdog() + self.leds.run_action("color", _OFF) + + # Wakeword ---------------------------------------------------------------------- + @subscribe + def voice_wakeword(self, data: dict): + _LOGGER.debug("wakeword -> fast GREEN flash x2, then enter listening (pulse)") + # Start a watchdog: if user never speaks, stop pulsing after timeout + self._cancel_watchdog() + self._listen_idle_watchdog = self.state.loop.create_task( + self._listen_watchdog(LISTEN_IDLE_TIMEOUT_SEC) + ) + # Flash twice quickly, then pulse + self.leds.run_action("flash_then_pulse", _GREEN, 2, _GREEN, 0.02) + + # Listening (VAD) --------------------------------------------------------------- + @subscribe + def voice_VOICE_ASSISTANT_STT_START(self, data: dict): + _LOGGER.debug("VAD_START -> ensure GREEN pulsing (listening)") + # User started speaking: cancel the idle watchdog and ensure pulse + self._cancel_watchdog() + self.leds.run_action("pulse", _GREEN, 0.02) + + @subscribe + def voice_VOICE_ASSISTANT_STT_VAD_END(self, data: dict): + _LOGGER.debug("VAD_END -> stop pulsing -> solid GREEN") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) + + # TTS (speaking) ---------------------------------------------------------------- + @subscribe + def voice_play_tts(self, data: dict): + _LOGGER.debug("TTS -> solid GREEN") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) + + @subscribe + def voice__tts_finished(self, data: dict): + _LOGGER.debug("TTS finished -> solid GREEN") + self._cancel_watchdog() + self.leds.run_action("color", self._base_color) +