sirmrak/linux-voice-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Support for NEOPixel via SPI

Browse Source
This commit is contained in:
imonlinux
2025-10-15 20:34:12 -07:00
committed by GitHub
parent d427d138e6
commit 8bef7e0f7a
1 changed files with 79 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux_voice_assistant/led_controller.py
+77 -75
View File
@@ -3,7 +3,6 @@ import logging
from typing import Any, Tuple
import board
import adafruit_dotstar
from .event_bus import EventHandler, subscribe
@@ -20,39 +19,64 @@ _PURPLE = (128, 0, 255)
class LedController(EventHandler):
def __init__(self, state, interface: str = "spi", clock_pin: int = 13, data_pin: int = 12, num_leds: int = 3):
def __init__(
self,
state,
led_type: str = "dotstar",
interface: str = "spi",
clock_pin: int = 13,
data_pin: int = 12,
num_leds: int = 3,
):
super().__init__(state)
self.loop = state.loop
self.num_leds = num_leds
self.current_task = None
self._is_ready = False
self.leds = None
# --- Central Configuration Dictionary ---
self.configs = {
"idle": {"effect": "off", "color": _PURPLE, "brightness": 0.5},
"idle": {"effect": "off", "color": _PURPLE, "brightness": 0.5},
"listening": {"effect": "medium_pulse", "color": _BLUE, "brightness": 0.5},
"thinking": {"effect": "spin", "color": _YELLOW, "brightness": 0.8},
"thinking": {"effect": "spin", "color": _YELLOW, "brightness": 0.8},
"responding": {"effect": "medium_pulse", "color": _GREEN, "brightness": 0.5},
"error": {"effect": "fast_blink", "color": _ORANGE, "brightness": 1.0},
"error": {"effect": "fast_blink", "color": _ORANGE, "brightness": 1.0},
}
try:
# Initialize hardware
if interface == "gpio":
data_pin_obj = getattr(board, f"D{data_pin}")
clock_pin_obj = getattr(board, f"D{clock_pin}")
self.leds = adafruit_dotstar.DotStar(clock_pin_obj, data_pin_obj, self.num_leds, auto_write=False)
else:
self.leds = adafruit_dotstar.DotStar(board.SCLK, board.MOSI, self.num_leds, auto_write=False)
# Universal Hardware Initialization
if led_type == "neopixel":
if interface == "spi":
# --- THIS IS THE FIX ---
import busio
import neopixel_spi # The module name is just 'neopixel_spi'
_LOGGER.debug(f"Initializing {self.num_leds} NeoPixel LEDs on hardware SPI")
spi = busio.SPI(board.SCLK, MOSI=board.MOSI)
self.leds = neopixel_spi.NeoPixel_SPI(spi, self.num_leds, auto_write=False)
else: # GPIO
import neopixel
_LOGGER.debug(f"Initializing {self.num_leds} NeoPixel LEDs on GPIO data={data_pin}")
pin_object = getattr(board, f"D{data_pin}")
self.leds = neopixel.NeoPixel(pin_object, self.num_leds, auto_write=False)
else: # dotstar
import adafruit_dotstar
if interface == "gpio":
_LOGGER.debug(f"Initializing {self.num_leds} DotStar LEDs on GPIO data={data_pin}, clock={clock_pin}")
data_pin_obj = getattr(board, f"D{data_pin}")
clock_pin_obj = getattr(board, f"D{clock_pin}")
self.leds = adafruit_dotstar.DotStar(clock_pin_obj, data_pin_obj, self.num_leds, auto_write=False)
else: # SPI
_LOGGER.debug(f"Initializing {self.num_leds} DotStar LEDs on hardware SPI")
self.leds = adafruit_dotstar.DotStar(board.SCLK, board.MOSI, self.num_leds, auto_write=False)
self._is_ready = True
self.leds.brightness = self.configs["idle"]["brightness"]
_LOGGER.info(f"LED Controller initialized for {self.num_leds} LEDs using {interface} interface.")
_LOGGER.info(f"LED Controller initialized for {self.num_leds} {led_type} LEDs.")
self.run_action("startup_sequence")
except Exception:
_LOGGER.exception("Failed to initialize LED controller. LEDs will be disabled.")
# ... (The rest of the file is identical and does not need to be changed)
def run_action(self, action_method_name: str, *args: Any) -> None:
if not self._is_ready: return
if self.current_task and not self.current_task.done(): self.current_task.cancel()
@@ -63,47 +87,53 @@ class LedController(EventHandler):
def _apply_state_effect(self, state_name: str, publish_state: bool = True):
config = self.configs.get(state_name, self.configs["idle"])
_LOGGER.debug(f"Applying effect for state '{state_name}': {config['effect']}")
self.leds.brightness = config["brightness"]
self.run_action(config["effect"], config["color"])
self.run_action(config["effect"], config["color"], config["brightness"])
if publish_state:
config["state_name"] = state_name
self.state.event_bus.publish("publish_state_to_mqtt", config)
async def startup_sequence(self, color=None, brightness=None): await self.blink(_GREEN, 1.0)
async def off(self, color, brightness): self.leds.fill(_OFF); self.leds.show()
async def solid(self, color: Tuple[int, int, int], brightness: float):
r, g, b = color
self.leds.fill((int(r * brightness), int(g * brightness), int(b * brightness)))
self.leds.show()
async def blink(self, color, brightness=1.0): await self.medium_blink(color, brightness)
# --- Animations ---
async def startup_sequence(self): await self.blink(_GREEN, 1)
async def off(self, color): self.leds.fill(_OFF); self.leds.show()
async def solid(self, color: Tuple[int, int, int]): self.leds.fill(color); self.leds.show()
async def blink(self, color, count=3): await self.medium_blink(color)
async def _base_pulse(self, color: Tuple[int, int, int], speed: float):
async def _base_pulse(self, color: Tuple[int, int, int], brightness: float, speed: float):
try:
r, g, b = color
while True:
for i in range(0, 101, 10):
mod = i / 100.0; self.leds.fill((int(r*mod), int(g*mod), int(b*mod))); self.leds.show(); await asyncio.sleep(speed)
mod = i / 100.0 * brightness
self.leds.fill((int(r*mod), int(g*mod), int(b*mod))); self.leds.show(); await asyncio.sleep(speed)
for i in range(100, -1, -10):
mod = i / 100.0; self.leds.fill((int(r*mod), int(g*mod), int(b*mod))); self.leds.show(); await asyncio.sleep(speed)
mod = i / 100.0 * brightness
self.leds.fill((int(r*mod), int(g*mod), int(b*mod))); self.leds.show(); await asyncio.sleep(speed)
except asyncio.CancelledError: self.leds.fill(_OFF); self.leds.show()
async def slow_pulse(self, color): await self._base_pulse(color, 0.05)
async def medium_pulse(self, color): await self._base_pulse(color, 0.02)
async def fast_pulse(self, color): await self._base_pulse(color, 0.008)
async def _base_blink(self, color: Tuple[int, int, int], speed: float):
async def slow_pulse(self, color, brightness): await self._base_pulse(color, brightness, 0.05)
async def medium_pulse(self, color, brightness): await self._base_pulse(color, 0.02)
async def fast_pulse(self, color, brightness): await self._base_pulse(color, 0.008)
async def _base_blink(self, color: Tuple[int, int, int], brightness: float, speed: float):
try:
r, g, b = color
bright_color = (int(r * brightness), int(g * brightness), int(b * brightness))
while True:
self.leds.fill(color); self.leds.show(); await asyncio.sleep(speed)
self.leds.fill(bright_color); self.leds.show(); await asyncio.sleep(speed)
self.leds.fill(_OFF); self.leds.show(); await asyncio.sleep(speed)
except asyncio.CancelledError: self.leds.fill(_OFF); self.leds.show()
async def slow_blink(self, color): await self._base_blink(color, 1.0)
async def medium_blink(self, color): await self._base_blink(color, 0.5)
async def fast_blink(self, color): await self._base_blink(color, 0.1)
async def spin(self, color: Tuple[int, int, int], speed: float = 0.1):
async def slow_blink(self, color, brightness): await self._base_blink(color, brightness, 1.0)
async def medium_blink(self, color, brightness): await self._base_blink(color, brightness, 0.5)
async def fast_blink(self, color, brightness): await self._base_blink(color, brightness, 0.1)
async def spin(self, color: Tuple[int, int, int], brightness: float, speed: float = 0.1):
try:
i = 0
bright_color = (int(color[0] * brightness), int(color[1] * brightness), int(color[2] * brightness))
while True:
self.leds.fill(_OFF); self.leds[i % self.num_leds] = color; self.leds.show(); i += 1; await asyncio.sleep(speed)
self.leds.fill(_OFF); self.leds[i % self.num_leds] = bright_color; self.leds.show(); i += 1; await asyncio.sleep(speed)
except asyncio.CancelledError: self.leds.fill(_OFF); self.leds.show()
# --- Voice Event Subscribers ---
@subscribe
def ha_connected(self, data: dict): self._apply_state_effect("idle")
@subscribe
@@ -121,55 +151,28 @@ class LedController(EventHandler):
@subscribe
def voice_run_end(self, data: dict): self._apply_state_effect("idle")
@subscribe
def mic_muted(self, data: dict): self.run_action("solid", _DIM_RED)
def mic_muted(self, data: dict): self.run_action("solid", _DIM_RED, 1.0)
@subscribe
def mic_unmuted(self, data: dict): self._apply_state_effect("idle")
# --- MQTT Config Subscribers ---
def _update_config(self, state_name: str, data: dict, apply: bool):
config = self.configs[state_name]
changed = False
is_retained = data.get("retained", False)
config = self.configs[state_name]; changed = False; is_retained = data.get("retained", False)
new_effect = data.get("effect")
if new_effect is not None and config["effect"] != new_effect:
config["effect"] = new_effect
changed = True
if new_effect is not None and config["effect"] != new_effect: config["effect"] = new_effect; changed = True
new_color_data = data.get("color")
if new_color_data is not None:
new_color_tuple = (new_color_data.get("r"), new_color_data.get("g"), new_color_data.get("b"))
if config["color"] != new_color_tuple:
config["color"] = new_color_tuple
changed = True
if config["color"] != new_color_tuple: config["color"] = new_color_tuple; changed = True
new_brightness = data.get("brightness")
if new_brightness is not None:
new_brightness_float = new_brightness / 255.0
if abs(config["brightness"] - new_brightness_float) > 0.001:
config["brightness"] = new_brightness_float
changed = True
# --- THIS IS THE FIX ---
if not changed:
# If it's a retained message for the idle state on startup, we still need to apply it once.
# But since nothing changed, we can assume it's already correct.
if is_retained and state_name == "idle":
self._apply_state_effect("idle", publish_state=False) # Apply visuals, but DO NOT publish back
return
# If we get here, the configuration has changed.
if abs(config["brightness"] - new_brightness_float) > 0.001: config["brightness"] = new_brightness_float; changed = True
if is_retained:
if state_name == "idle":
self._apply_state_effect("idle", publish_state=False) # Apply visuals, but DO NOT publish back
# For other retained messages, we just update config silently.
elif apply:
self._apply_state_effect(state_name) # Apply visuals AND publish back
else:
# Config was updated for a non-active state, just publish it back to confirm.
config["state_name"] = state_name
self.state.event_bus.publish("publish_state_to_mqtt", config)
if changed and state_name == "idle": self._apply_state_effect("idle", publish_state=False)
return
if changed:
if apply: self._apply_state_effect(state_name)
else: config["state_name"] = state_name; self.state.event_bus.publish("publish_state_to_mqtt", config)
@subscribe
def set_idle_effect(self, data: dict): self._update_config("idle", data, True)
@@ -191,7 +194,6 @@ class LedController(EventHandler):
def set_error_effect(self, data: dict): self._update_config("error", data, False)
@subscribe
def set_error_color(self, data: dict): self._update_config("error", data, False)
@subscribe
def set_num_leds(self, data: dict):
num_leds = data.get("num_leds")