189 lines
6.2 KiB
Python
189 lines
6.2 KiB
Python
|
|
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)
|