Add files via upload
This commit is contained in:
imonlinux
@@ -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}'")
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user