color parsing cleanup

This commit is contained in:
2026-08-26 11:28:48 +02:00
parent 1cb3387fbe
commit 55bcc9448c
2 changed files with 24 additions and 19 deletions

View File

@@ -14,7 +14,7 @@ import os
from hass_client import HomeAssistantClient from hass_client import HomeAssistantClient
from ruamel.yaml import YAML from ruamel.yaml import YAML
import warnings import warnings
from typing import Optional from typing import Optional, NamedTuple
from mqtt_json import start_mqtt from mqtt_json import start_mqtt
yaml = YAML(typ='safe') yaml = YAML(typ='safe')
@@ -22,6 +22,13 @@ yaml = YAML(typ='safe')
OFF_COLOR = ColorRGBW(0, 0, 0, 0) OFF_COLOR = ColorRGBW(0, 0, 0, 0)
class FigureColors(NamedTuple):
primary: ColorRGBW
secondary: ColorRGBW
bg: ColorRGBW
accent: ColorRGBW
def parse_color(color_str: str): def parse_color(color_str: str):
if isinstance(color_str, ColorRGBW): if isinstance(color_str, ColorRGBW):
return color_str return color_str
@@ -32,6 +39,8 @@ def parse_color(color_str: str):
elif color_str.startswith("w"): elif color_str.startswith("w"):
color_str = color_str.lstrip("w") color_str = color_str.lstrip("w")
return ColorRGBW(0, 0, 0, int(color_str, 16) / 255) return ColorRGBW(0, 0, 0, int(color_str, 16) / 255)
else:
raise ValueError(f"Unrecognized color format: {color_str!r}")
def load_config(config_path): def load_config(config_path):
@@ -39,7 +48,7 @@ def load_config(config_path):
with open(os.path.join(config_path, "config.yml")) as cfg_file: with open(os.path.join(config_path, "config.yml")) as cfg_file:
cfg = yaml.load(cfg_file) cfg = yaml.load(cfg_file)
for figure_name, figure_cfg in cfg["figures"].items(): for figure_name, figure_cfg in cfg["figures"].items():
figure_cfg["colors"] = [parse_color(c) for c in figure_cfg["colors"]] figure_cfg["colors"] = FigureColors(*(parse_color(c) for c in figure_cfg["colors"]))
if 'media_files' not in figure_cfg: if 'media_files' not in figure_cfg:
figure_cfg['media_files'] = sorted(glob(os.path.join(config_path, figure_name))) figure_cfg['media_files'] = sorted(glob(os.path.join(config_path, figure_name)))
return cfg return cfg
@@ -131,9 +140,8 @@ class Controller:
self.mmstate.active_figure = None self.mmstate.active_figure = None
elif tagid in self._rfid_to_figure_name: elif tagid in self._rfid_to_figure_name:
newly_placed_figure = self._rfid_to_figure_name[tagid] newly_placed_figure = self._rfid_to_figure_name[tagid]
primary_color, secondary_color, *rest = self.cfg["figures"][newly_placed_figure][ colors = self.cfg["figures"][newly_placed_figure]["colors"]
"colors"] self._start_animation(colors.primary, colors.secondary)
self._start_animation(primary_color, secondary_color)
self.mmstate.button_leds(self.cfg["general"].get("button_leds_brightness", 0.5)) self.mmstate.button_leds(self.cfg["general"].get("button_leds_brightness", 0.5))
if newly_placed_figure in self.cfg['figures']: if newly_placed_figure in self.cfg['figures']:
@@ -143,8 +151,7 @@ class Controller:
else: else:
print("Restarting playlist") print("Restarting playlist")
self.audio_player.set_playlist( self.audio_player.set_playlist(
self.audio_player.create_playlist( self.audio_player.create_playlist(self.cfg['figures'][newly_placed_figure]['media_files']))
self.cfg['figures'][newly_placed_figure]['media_files']))
self.audio_player.play_from_start() self.audio_player.play_from_start()
self.mmstate.active_figure = newly_placed_figure self.mmstate.active_figure = newly_placed_figure
@@ -156,8 +163,7 @@ class Controller:
if isinstance(message, RfidTokenRead): if isinstance(message, RfidTokenRead):
self.handle_rfid_event(message.id) self.handle_rfid_event(message.id)
elif isinstance(message, RotaryEncoderEvent): elif isinstance(message, RotaryEncoderEvent):
volume_increment = self.cfg["general"].get("volume_increment", 2) * abs( volume_increment = self.cfg["general"].get("volume_increment", 2) * abs(message.increment)
message.increment)
if message.direction == 2: if message.direction == 2:
self.audio_player.change_volume(volume_increment) self.audio_player.change_volume(volume_increment)
elif message.direction == 1: elif message.direction == 1:
@@ -173,9 +179,9 @@ class Controller:
elif isinstance(message, TouchButtonPress): elif isinstance(message, TouchButtonPress):
figure = self.mmstate.active_figure figure = self.mmstate.active_figure
if figure and self.audio_player.is_playing(): if figure and self.audio_player.is_playing():
primary_color, secondary_color, bg, accent = self.cfg["figures"][figure]["colors"] figure_colors = self.cfg["figures"][figure]["colors"]
self.protocol.mouse_led_effect( self.protocol.mouse_led_effect(
EffectStaticConfig(accent, *mouse_leds_index_ranges[message.touch_button])) EffectStaticConfig(figure_colors.accent, *mouse_leds_index_ranges[message.touch_button]))
colors = { colors = {
TouchButton.RIGHT_FOOT: { TouchButton.RIGHT_FOOT: {
@@ -204,14 +210,14 @@ class Controller:
eff_static = EffectStaticConfig(ColorRGBW(0, 0, 0, 0), eff_static = EffectStaticConfig(ColorRGBW(0, 0, 0, 0),
*mouse_leds_index_ranges[message.touch_button]) *mouse_leds_index_ranges[message.touch_button])
if figure and self.audio_player.is_playing(): if figure and self.audio_player.is_playing():
primary_color, secondary_color, bg, accent = self.cfg["figures"][figure]["colors"] colors = self.cfg["figures"][figure]["colors"]
eff_static.color = primary_color eff_static.color = colors.primary
self.protocol.mouse_led_effect(eff_static) self.protocol.mouse_led_effect(eff_static)
if figure and self.audio_player.is_playing(): if figure and self.audio_player.is_playing():
primary_color, secondary_color, bg, accent = self.cfg["figures"][figure]["colors"] colors = self.cfg["figures"][figure]["colors"]
eff_change.color1 = primary_color eff_change.color1 = colors.primary
eff_change.color2 = secondary_color eff_change.color2 = colors.secondary
eff_change.start_with_existing = True eff_change.start_with_existing = True
self.protocol.mouse_led_effect(eff_change) self.protocol.mouse_led_effect(eff_change)

View File

@@ -152,7 +152,6 @@ class ShelveLightMqtt:
async def _notify_mqtt_state(self, state): async def _notify_mqtt_state(self, state):
state_payload = json.dumps(self._state) state_payload = json.dumps(self._state)
print("OUT ", state_payload)
await self._mqtt_client.publish(self._discovery_spec['state_topic'], state_payload.encode()) await self._mqtt_client.publish(self._discovery_spec['state_topic'], state_payload.encode())