Replaced by led_controller.py

2025-10-11 12:41:58 -07:00
parent 6ce1c5f2b7
commit 688cc97d7d
1 changed files with 0 additions and 128 deletions
@@ -1,128 +0,0 @@
 # 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)