Replaced by led_controller.py
This commit is contained in:
imonlinux
@@ -1,158 +0,0 @@
|
||||
# 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)
|
||||
Reference in New Issue
Block a user