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) +