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

Merge pull request #28 from imonlinux/neopixel_spi

Browse Source 
Feature - Add WS2812B (NeoPixel) LED Support
This commit is contained in:
imonlinux
2025-10-15 20:36:22 -07:00
committed by GitHub
parent ce4486896a 8bef7e0f7a
commit c4f632a48f
4 changed files with 145 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux_voice_assistant/__main__.py
+55 -120
View File
@@ -118,6 +118,12 @@ async def main() -> None:
parser.add_argument( parser.add_argument(
"--port", type=int, default=6053, help="Port for ESPHome server (default: 6053)" "--port", type=int, default=6053, help="Port for ESPHome server (default: 6053)"
) )
parser.add_argument(
"--led-type",
choices=["dotstar", "neopixel"],
default="dotstar",
help="Type of LED strip (default: dotstar for APA102)",
)
parser.add_argument( parser.add_argument(
"--led-interface", "--led-interface",
choices=["spi", "gpio"], choices=["spi", "gpio"],
@@ -136,7 +142,6 @@ async def main() -> None:
default=12, default=12,
help="GPIO pin for LED data (for Grove)", help="GPIO pin for LED data (for Grove)",
) )
# --- ADDED NUM_LEDS ARGUMENT ---
parser.add_argument( parser.add_argument(
"--num-leds", "--num-leds",
type=int, type=int,
@@ -151,16 +156,24 @@ async def main() -> None:
"--debug", action="store_true", help="Print DEBUG messages to console" "--debug", action="store_true", help="Print DEBUG messages to console"
) )
args = parser.parse_args() args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
_LOGGER.debug(args) _LOGGER.debug(args)
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
event_bus = EventBus() event_bus = EventBus()
wake_word_dirs = [Path(ww_dir) for ww_dir in args.wake_word_dir] preferences_path = Path(args.preferences_file)
if preferences_path.exists():
with open(preferences_path, "r", encoding="utf-8") as preferences_file:
preferences_dict = json.load(preferences_file)
preferences = Preferences(**preferences_dict)
else:
preferences = Preferences()
preferences.num_leds = getattr(preferences, 'num_leds', args.num_leds)
available_wake_words: Dict[str, AvailableWakeWord] = {} available_wake_words: Dict[str, AvailableWakeWord] = {}
for wake_word_dir in wake_word_dirs: for wake_word_dir in [Path(d) for d in args.wake_word_dir]:
for model_config_path in wake_word_dir.glob("*.json"): for model_config_path in wake_word_dir.glob("*.json"):
model_id = model_config_path.stem model_id = model_config_path.stem
if model_id == args.stop_model: if model_id == args.stop_model:
@@ -175,55 +188,34 @@ async def main() -> None:
trained_languages=model_config.get("trained_languages", []), trained_languages=model_config.get("trained_languages", []),
config_path=model_config_path, config_path=model_config_path,
) )
_LOGGER.debug("Available wake words: %s", list(sorted(available_wake_words.keys()))) _LOGGER.debug("Available wake words: %s", list(sorted(available_wake_words.keys())))
preferences_path = Path(args.preferences_file)
if preferences_path.exists():
_LOGGER.debug("Loading preferences: %s", preferences_path)
with open(preferences_path, "r", encoding="utf-8") as preferences_file:
preferences_dict = json.load(preferences_file)
preferences = Preferences(**preferences_dict)
else:
preferences = Preferences()
# Set number of LEDs from command line default, then override with saved preference
preferences.num_leds = getattr(preferences, 'num_leds', args.num_leds)
libtensorflowlite_c_path = _LIB_DIR / "libtensorflowlite_c.so" libtensorflowlite_c_path = _LIB_DIR / "libtensorflowlite_c.so"
_LOGGER.debug("libtensorflowlite_c path: %s", libtensorflowlite_c_path)
wake_models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {} wake_models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {}
if preferences.active_wake_words: if preferences.active_wake_words:
for wake_word_id in preferences.active_wake_words: for wake_word_id in preferences.active_wake_words:
wake_word = available_wake_words.get(wake_word_id) wake_word = available_wake_words.get(wake_word_id)
if wake_word is None: if wake_word is None: continue
_LOGGER.warning("Unrecognized wake word id: %s", wake_word_id)
continue
_LOGGER.debug("Loading wake model: %s", wake_word_id)
wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path) wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path)
if not wake_models: if not wake_models:
wake_word_id = args.wake_model wake_word_id = args.wake_model
wake_word = available_wake_words[wake_word_id] wake_word = available_wake_words.get(wake_word_id)
_LOGGER.debug("Loading wake model: %s", wake_word_id) if wake_word:
wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path) wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path)
stop_model: Optional[MicroWakeWord] = None stop_model: Optional[MicroWakeWord] = None
for wake_word_dir in wake_word_dirs: for wake_word_dir in [Path(d) for d in args.wake_word_dir]:
stop_config_path = wake_word_dir / f"{args.stop_model}.json" stop_config_path = wake_word_dir / f"{args.stop_model}.json"
if not stop_config_path.exists(): if not stop_config_path.exists(): continue
continue stop_model = MicroWakeWord.from_config(stop_config_path, libtensorflowlite_c_path)
_LOGGER.debug("Loading stop model: %s", stop_config_path)
stop_model = MicroWakeWord.from_config(
stop_config_path, libtensorflowlite_c_path
)
break break
assert stop_model is not None assert stop_model is not None
music_player = MpvMediaPlayer(
device=args.audio_output_device, initial_volume=preferences.volume_level music_player = MpvMediaPlayer(device=args.audio_output_device, initial_volume=preferences.volume_level)
) tts_player = MpvMediaPlayer(device=args.audio_output_device, initial_volume=preferences.volume_level)
tts_player = MpvMediaPlayer(
device=args.audio_output_device, initial_volume=preferences.volume_level
)
state = ServerState( state = ServerState(
name=args.name, name=args.name,
mac_address=get_mac(), mac_address=get_mac(),
@@ -248,69 +240,42 @@ async def main() -> None:
led_controller = LedController( led_controller = LedController(
state, state,
led_type=args.led_type,
interface=args.led_interface, interface=args.led_interface,
clock_pin=args.led_clock_pin, clock_pin=args.led_clock_pin,
data_pin=args.led_data_pin, data_pin=args.led_data_pin,
num_leds=state.preferences.num_leds, # Pass number of LEDs num_leds=state.preferences.num_leds,
) )
mqtt_controller: Optional[MqttController] = None mqtt_controller: Optional[MqttController] = None
if args.mqtt_host: if args.mqtt_host:
mqtt_controller = MqttController( mqtt_controller = MqttController(state=state, host=args.mqtt_host, port=args.mqtt_port, username=args.mqtt_username, password=args.mqtt_password)
state=state,
host=args.mqtt_host,
port=args.mqtt_port,
username=args.mqtt_username,
password=args.mqtt_password,
)
mqtt_controller.start() mqtt_controller.start()
mic_mute_handler = MicMuteHandler(state, mqtt_controller) mic_mute_handler = MicMuteHandler(state, mqtt_controller)
process_audio_thread = threading.Thread(target=process_audio, args=(state,), daemon=True)
process_audio_thread = threading.Thread(
target=process_audio, args=(state,), daemon=True
)
process_audio_thread.start() process_audio_thread.start()
def sd_callback(indata, _frames, _time, _status): def sd_callback(indata, _frames, _time, _status): state.audio_queue.put_nowait(bytes(indata))
state.audio_queue.put_nowait(bytes(indata))
server = await loop.create_server( server = await loop.create_server(lambda: VoiceSatelliteProtocol(state), host=args.host, port=args.port)
lambda: VoiceSatelliteProtocol(state), host=args.host, port=args.port
)
discovery = HomeAssistantZeroconf(port=args.port, name=args.name) discovery = HomeAssistantZeroconf(port=args.port, name=args.name)
await discovery.register_server() await discovery.register_server()
try: try:
_LOGGER.debug("Opening audio input device: %s", args.audio_input_device) with sd.RawInputStream(samplerate=16000, blocksize=args.audio_input_block_size, device=args.audio_input_device, dtype="int16", channels=1, callback=sd_callback):
with sd.RawInputStream(
samplerate=16000,
blocksize=args.audio_input_block_size,
device=args.audio_input_device,
dtype="int16",
channels=1,
callback=sd_callback,
):
async with server: async with server:
_LOGGER.info("Server started (host=%s, port=%s)", args.host, args.port) _LOGGER.info("Server started (host=%s, port=%s)", args.host, args.port)
await server.serve_forever() await server.serve_forever()
except KeyboardInterrupt: except KeyboardInterrupt: pass
pass
finally: finally:
if mqtt_controller: if mqtt_controller: mqtt_controller.stop()
mqtt_controller.stop()
state.audio_queue.put_nowait(None) state.audio_queue.put_nowait(None)
process_audio_thread.join() process_audio_thread.join()
_LOGGER.debug("Server stopped") _LOGGER.debug("Server stopped")
# -----------------------------------------------------------------------------
def process_audio(state: ServerState): def process_audio(state: ServerState):
"""Process audio chunks from the microphone."""
wake_words: List[Union[MicroWakeWord, OpenWakeWord]] = [] wake_words: List[Union[MicroWakeWord, OpenWakeWord]] = []
micro_features: Optional[MicroWakeWordFeatures] = None micro_features: Optional[MicroWakeWordFeatures] = None
micro_inputs: List[np.ndarray] = [] micro_inputs: List[np.ndarray] = []
@@ -318,75 +283,45 @@ def process_audio(state: ServerState):
oww_inputs: List[np.ndarray] = [] oww_inputs: List[np.ndarray] = []
has_oww = False has_oww = False
last_active: Optional[float] = None last_active: Optional[float] = None
try: try:
while True: while True:
if state.mic_muted: if state.mic_muted:
time.sleep(0.1) time.sleep(0.1)
continue continue
audio_chunk = state.audio_queue.get() audio_chunk = state.audio_queue.get()
if audio_chunk is None: if audio_chunk is None: break
break if state.satellite is None: continue
if state.satellite is None:
continue
if (not wake_words) or (state.wake_words_changed and state.wake_words): if (not wake_words) or (state.wake_words_changed and state.wake_words):
state.wake_words_changed = False state.wake_words_changed = False
wake_words = [ww for ww in state.wake_words.values() if ww.is_active] wake_words = [ww for ww in state.wake_words.values() if ww.is_active]
has_oww = False has_oww = any(isinstance(ww, OpenWakeWord) for ww in wake_words)
for wake_word in wake_words:
if isinstance(wake_word, OpenWakeWord):
has_oww = True
if micro_features is None: if micro_features is None:
micro_features = MicroWakeWordFeatures( micro_features = MicroWakeWordFeatures(libtensorflowlite_c_path=state.libtensorflowlite_c_path)
libtensorflowlite_c_path=state.libtensorflowlite_c_path,
)
if has_oww and (oww_features is None): if has_oww and (oww_features is None):
oww_features = OpenWakeWordFeatures( oww_features = OpenWakeWordFeatures(melspectrogram_model=state.oww_melspectrogram_path, embedding_model=state.oww_embedding_path, libtensorflowlite_c_path=state.libtensorflowlite_c_path)
melspectrogram_model=state.oww_melspectrogram_path,
embedding_model=state.oww_embedding_path,
libtensorflowlite_c_path=state.libtensorflowlite_c_path,
)
try: try:
state.satellite.handle_audio(audio_chunk) state.satellite.handle_audio(audio_chunk)
assert micro_features is not None assert micro_features is not None
micro_inputs.clear() micro_inputs.clear(); micro_inputs.extend(micro_features.process_streaming(audio_chunk))
micro_inputs.extend(micro_features.process_streaming(audio_chunk))
if has_oww: if has_oww:
assert oww_features is not None assert oww_features is not None
oww_inputs.clear() oww_inputs.clear(); oww_inputs.extend(oww_features.process_streaming(audio_chunk))
oww_inputs.extend(oww_features.process_streaming(audio_chunk))
for wake_word in wake_words: for wake_word in wake_words:
activated = False activated = False
if isinstance(wake_word, MicroWakeWord): if isinstance(wake_word, MicroWakeWord):
for micro_input in micro_inputs: if any(wake_word.process_streaming(mi) for mi in micro_inputs): activated = True
if wake_word.process_streaming(micro_input):
activated = True
elif isinstance(wake_word, OpenWakeWord): elif isinstance(wake_word, OpenWakeWord):
for oww_input in oww_inputs: if any(p > 0.5 for oi in oww_inputs for p in wake_word.process_streaming(oi)): activated = True
for prob in wake_word.process_streaming(oww_input):
if prob > 0.5:
activated = True
if activated: if activated:
now = time.monotonic() now = time.monotonic()
if (last_active is None) or ( if (last_active is None) or ((now - last_active) > state.refractory_seconds):
(now - last_active) > state.refractory_seconds
):
state.satellite.wakeup(wake_word) state.satellite.wakeup(wake_word)
last_active = now last_active = now
stopped = False if any(state.stop_word.process_streaming(mi) for mi in micro_inputs) and state.stop_word.is_active:
for micro_input in micro_inputs:
if state.stop_word.process_streaming(micro_input):
stopped = True
if stopped and state.stop_word.is_active:
state.satellite.stop() state.satellite.stop()
except Exception: except Exception: _LOGGER.exception("Unexpected error handling audio")
_LOGGER.exception("Unexpected error handling audio") except Exception: _LOGGER.exception("Unexpected error processing audio")
except Exception:
_LOGGER.exception("Unexpected error processing audio")
# -----------------------------------------------------------------------------
if __name__ == "__main__": if __name__ == "__main__":
print("--- __main__ block executing ---")
asyncio.run(main()) asyncio.run(main())
linux_voice_assistant/led_controller.py
+79 -77
View File
@@ -3,7 +3,6 @@ import logging
from typing import Any, Tuple from typing import Any, Tuple
import board import board
import adafruit_dotstar
from .event_bus import EventHandler, subscribe from .event_bus import EventHandler, subscribe
@@ -20,39 +19,64 @@ _PURPLE = (128, 0, 255)
class LedController(EventHandler): 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) super().__init__(state)
self.loop = state.loop self.loop = state.loop
self.num_leds = num_leds self.num_leds = num_leds
self.current_task = None self.current_task = None
self._is_ready = False self._is_ready = False
self.leds = None
# --- Central Configuration Dictionary ---
self.configs = { 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}, "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}, "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: try:
# Initialize hardware # Universal Hardware Initialization
if interface == "gpio": if led_type == "neopixel":
data_pin_obj = getattr(board, f"D{data_pin}") if interface == "spi":
clock_pin_obj = getattr(board, f"D{clock_pin}") # --- THIS IS THE FIX ---
self.leds = adafruit_dotstar.DotStar(clock_pin_obj, data_pin_obj, self.num_leds, auto_write=False) import busio
else: import neopixel_spi # The module name is just 'neopixel_spi'
self.leds = adafruit_dotstar.DotStar(board.SCLK, board.MOSI, self.num_leds, auto_write=False) _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._is_ready = True
self.leds.brightness = self.configs["idle"]["brightness"] _LOGGER.info(f"LED Controller initialized for {self.num_leds} {led_type} LEDs.")
_LOGGER.info(f"LED Controller initialized for {self.num_leds} LEDs using {interface} interface.")
self.run_action("startup_sequence") self.run_action("startup_sequence")
except Exception: except Exception:
_LOGGER.exception("Failed to initialize LED controller. LEDs will be disabled.") _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: def run_action(self, action_method_name: str, *args: Any) -> None:
if not self._is_ready: return if not self._is_ready: return
if self.current_task and not self.current_task.done(): self.current_task.cancel() 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): def _apply_state_effect(self, state_name: str, publish_state: bool = True):
config = self.configs.get(state_name, self.configs["idle"]) config = self.configs.get(state_name, self.configs["idle"])
_LOGGER.debug(f"Applying effect for state '{state_name}': {config['effect']}") _LOGGER.debug(f"Applying effect for state '{state_name}': {config['effect']}")
self.leds.brightness = config["brightness"] self.run_action(config["effect"], config["color"], config["brightness"])
self.run_action(config["effect"], config["color"])
if publish_state: if publish_state:
config["state_name"] = state_name config["state_name"] = state_name
self.state.event_bus.publish("publish_state_to_mqtt", config) 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)
# --- Animations --- async def off(self, color, brightness): self.leds.fill(_OFF); self.leds.show()
async def startup_sequence(self): await self.blink(_GREEN, 1) async def solid(self, color: Tuple[int, int, int], brightness: float):
async def off(self, color): self.leds.fill(_OFF); self.leds.show() r, g, b = color
async def solid(self, color: Tuple[int, int, int]): self.leds.fill(color); self.leds.show() self.leds.fill((int(r * brightness), int(g * brightness), int(b * brightness)))
async def blink(self, color, count=3): await self.medium_blink(color) self.leds.show()
async def _base_pulse(self, color: Tuple[int, int, int], speed: float): async def blink(self, color, brightness=1.0): await self.medium_blink(color, brightness)
async def _base_pulse(self, color: Tuple[int, int, int], brightness: float, speed: float):
try: try:
r, g, b = color r, g, b = color
while True: while True:
for i in range(0, 101, 10): 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): 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() 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 slow_pulse(self, color, brightness): await self._base_pulse(color, brightness, 0.05)
async def medium_pulse(self, color): await self._base_pulse(color, 0.02) async def medium_pulse(self, color, brightness): await self._base_pulse(color, 0.02)
async def fast_pulse(self, color): await self._base_pulse(color, 0.008) async def fast_pulse(self, color, brightness): await self._base_pulse(color, 0.008)
async def _base_blink(self, color: Tuple[int, int, int], speed: float):
async def _base_blink(self, color: Tuple[int, int, int], brightness: float, speed: float):
try: try:
r, g, b = color
bright_color = (int(r * brightness), int(g * brightness), int(b * brightness))
while True: 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) self.leds.fill(_OFF); self.leds.show(); await asyncio.sleep(speed)
except asyncio.CancelledError: self.leds.fill(_OFF); self.leds.show() 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 slow_blink(self, color, brightness): await self._base_blink(color, brightness, 1.0)
async def medium_blink(self, color): await self._base_blink(color, 0.5) async def medium_blink(self, color, brightness): await self._base_blink(color, brightness, 0.5)
async def fast_blink(self, color): await self._base_blink(color, 0.1) async def fast_blink(self, color, brightness): await self._base_blink(color, brightness, 0.1)
async def spin(self, color: Tuple[int, int, int], speed: float = 0.1):
async def spin(self, color: Tuple[int, int, int], brightness: float, speed: float = 0.1):
try: try:
i = 0 i = 0
bright_color = (int(color[0] * brightness), int(color[1] * brightness), int(color[2] * brightness))
while True: 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() except asyncio.CancelledError: self.leds.fill(_OFF); self.leds.show()
# --- Voice Event Subscribers ---
@subscribe @subscribe
def ha_connected(self, data: dict): self._apply_state_effect("idle") def ha_connected(self, data: dict): self._apply_state_effect("idle")
@subscribe @subscribe
@@ -121,55 +151,28 @@ class LedController(EventHandler):
@subscribe @subscribe
def voice_run_end(self, data: dict): self._apply_state_effect("idle") def voice_run_end(self, data: dict): self._apply_state_effect("idle")
@subscribe @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 @subscribe
def mic_unmuted(self, data: dict): self._apply_state_effect("idle") 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): def _update_config(self, state_name: str, data: dict, apply: bool):
config = self.configs[state_name] config = self.configs[state_name]; changed = False; is_retained = data.get("retained", False)
changed = False
is_retained = data.get("retained", False)
new_effect = data.get("effect") new_effect = data.get("effect")
if new_effect is not None and config["effect"] != new_effect: if new_effect is not None and config["effect"] != new_effect: config["effect"] = new_effect; changed = True
config["effect"] = new_effect
changed = True
new_color_data = data.get("color") new_color_data = data.get("color")
if new_color_data is not None: 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")) new_color_tuple = (new_color_data.get("r"), new_color_data.get("g"), new_color_data.get("b"))
if config["color"] != new_color_tuple: if config["color"] != new_color_tuple: config["color"] = new_color_tuple; changed = True
config["color"] = new_color_tuple
changed = True
new_brightness = data.get("brightness") new_brightness = data.get("brightness")
if new_brightness is not None: if new_brightness is not None:
new_brightness_float = new_brightness / 255.0 new_brightness_float = new_brightness / 255.0
if abs(config["brightness"] - new_brightness_float) > 0.001: if abs(config["brightness"] - new_brightness_float) > 0.001: config["brightness"] = new_brightness_float; changed = True
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 is_retained: if is_retained:
if state_name == "idle": if changed and state_name == "idle": self._apply_state_effect("idle", publish_state=False)
self._apply_state_effect("idle", publish_state=False) # Apply visuals, but DO NOT publish back return
# For other retained messages, we just update config silently. if changed:
elif apply: if apply: self._apply_state_effect(state_name)
self._apply_state_effect(state_name) # Apply visuals AND publish back else: config["state_name"] = state_name; self.state.event_bus.publish("publish_state_to_mqtt", config)
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)
@subscribe @subscribe
def set_idle_effect(self, data: dict): self._update_config("idle", data, True) 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) def set_error_effect(self, data: dict): self._update_config("error", data, False)
@subscribe @subscribe
def set_error_color(self, data: dict): self._update_config("error", data, False) def set_error_color(self, data: dict): self._update_config("error", data, False)
@subscribe @subscribe
def set_num_leds(self, data: dict): def set_num_leds(self, data: dict):
num_leds = data.get("num_leds") num_leds = data.get("num_leds")
pyproject.toml
+3
View File
@@ -32,6 +32,9 @@ dependencies = [
"python-mpv>=1,<2", "python-mpv>=1,<2",
"zeroconf", "zeroconf",
"paho-mqtt", "paho-mqtt",
"adafruit-circuitpython-dotstar",
"adafruit-circuitpython-neopixel",
"adafruit-circuitpython-neopixel-spi",
] ]
[project.optional-dependencies] [project.optional-dependencies]
script/setup
+8 -14
View File
@@ -17,20 +17,14 @@ builder = venv.EnvBuilder(with_pip=True)
context = builder.ensure_directories(_VENV_DIR) context = builder.ensure_directories(_VENV_DIR)
builder.create(_VENV_DIR) builder.create(_VENV_DIR)
# Upgrade dependencies # Upgrade pip and setuptools
pip = [context.env_exe, "-m", "pip"] pip = [context.env_exe, "-m", "pip"]
subprocess.check_call(pip + ["install", "--upgrade", "pip"]) subprocess.check_call(pip + ["install", "--upgrade", "pip", "setuptools", "wheel"])
subprocess.check_call(pip + ["install", "--upgrade", "setuptools", "wheel"])
subprocess.check_call(pip + ["install", "--upgrade", "python-mpv"])
subprocess.check_call(pip + ["install", "--upgrade", "gpiozero"])
subprocess.check_call(pip + ["install", "--upgrade", "spidev"])
subprocess.check_call(pip + ["install", "--upgrade", "adafruit-circuitpython-dotstar"])
# Install requirements
subprocess.check_call(pip + ["install", "-e", str(_PROGRAM_DIR)])
# Install the project and all its dependencies from pyproject.toml
if args.dev: if args.dev:
# Install dev requirements # Install with dev dependencies
subprocess.check_call( subprocess.check_call(pip + ["install", "-e", f"{_PROGRAM_DIR}[dev]"])
pip + ["install", "-e", f"{_PROGRAM_DIR}[dev]"] else:
) # Install with regular dependencies
subprocess.check_call(pip + ["install", "-e", str(_PROGRAM_DIR)])