Martin Bauer
d44c24ec97
- event bus systen - all components are independent - preparation for web frontend
251 lines
9.1 KiB
Python
251 lines
9.1 KiB
Python
"""Each LED zone as a Home-Assistant-discoverable JSON light.
|
|
|
|
Two things changed from the old ``ShelveLightMqtt``:
|
|
|
|
* The ``side_*``/``top_*`` effect names are parsed rather than enumerated, so adding a
|
|
width or an increment is a data change (see :data:`WIDTHS`, :data:`INCREMENTS`).
|
|
* State is published from :class:`~musicmouse.events.LedEffectChanged` - the device's
|
|
report of what it actually did - instead of echoing back the command. When a figure
|
|
animation overrides an MQTT-set colour, Home Assistant now follows along.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
from musicmouse.bus import EventBus
|
|
from musicmouse.color import ColorRGBW
|
|
from musicmouse.config import MqttConfig
|
|
from musicmouse.devices.mouse import MusicMouseDevice
|
|
from musicmouse.effects import (
|
|
EffectCircularConfig,
|
|
EffectRandomTwoColorInterpolationConfig,
|
|
EffectStaticConfig,
|
|
EffectStaticDetailedConfig,
|
|
EffectSwipeAndChange,
|
|
LedEffect,
|
|
)
|
|
from musicmouse.events import LedEffectChanged
|
|
from musicmouse.hardware import LedZone
|
|
from musicmouse.services.mqtt.entity import Entity
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
__all__ = ["LightEntity", "effect_names", "parse_positional_effect"]
|
|
|
|
BLACK = ColorRGBW(0, 0, 0, 0)
|
|
|
|
#: Effects that are not simply "light this fraction of the strip".
|
|
BASE_EFFECTS = ("static", "circular", "wipeup", "twocolor", "twocolorrandom")
|
|
|
|
#: Fraction of the strip lit by a positional effect.
|
|
WIDTHS = (0.2, 0.5)
|
|
#: Light every n-th LED. 1 is solid.
|
|
INCREMENTS = (1, 4, 8)
|
|
|
|
_POSITIONAL = re.compile(r"^(?P<position>side|top)_(?P<width>\d+(?:\.\d+)?)(?:_inc(?P<inc>\d+))?$")
|
|
|
|
DEFAULT_TRANSITION_S = 0.3
|
|
|
|
|
|
def effect_names() -> list[str]:
|
|
"""Every effect name this entity accepts, for the discovery ``effect_list``."""
|
|
positional = [
|
|
f"{position}_{width:g}" + ("" if increment == 1 else f"_inc{increment}")
|
|
for position in ("side", "top")
|
|
for width in WIDTHS
|
|
for increment in INCREMENTS
|
|
]
|
|
return [*BASE_EFFECTS, *positional]
|
|
|
|
|
|
def parse_positional_effect(name: str) -> tuple[float, float, int] | None:
|
|
"""``"side_0.2_inc4"`` -> ``(begin, end, increment)``, or ``None`` if not one.
|
|
|
|
``side`` lights a band around the far end of the strip and wraps; ``top`` lights a
|
|
band centred on the middle.
|
|
"""
|
|
match = _POSITIONAL.match(name)
|
|
if match is None:
|
|
return None
|
|
width = float(match["width"])
|
|
increment = int(match["inc"] or 1)
|
|
if match["position"] == "side":
|
|
return 1.0 - width / 2, width / 2, increment
|
|
return 0.5 - width / 2, 0.5 + width / 2, increment
|
|
|
|
|
|
class LightEntity(Entity):
|
|
component = "light"
|
|
|
|
def __init__(
|
|
self,
|
|
bus: EventBus,
|
|
config: MqttConfig,
|
|
mouse: MusicMouseDevice,
|
|
zone: LedZone,
|
|
name: str,
|
|
) -> None:
|
|
self.zone = zone
|
|
self.mouse = mouse
|
|
self._state: dict[str, Any] = {
|
|
"state": "OFF",
|
|
"color": {"r": 255, "g": 255, "b": 255, "w": 0},
|
|
"color_mode": "rgbw",
|
|
"brightness": 30,
|
|
"effect": "static",
|
|
}
|
|
self._last_color = ColorRGBW(0.5, 0.5, 0.5, 0)
|
|
super().__init__(bus, config, object_id=f"light_{zone}", name=name)
|
|
|
|
# ---------------------------------------------------------------- discovery
|
|
|
|
def discovery_payload(self) -> dict[str, Any]:
|
|
return {
|
|
"schema": "json",
|
|
"name": self.name,
|
|
"unique_id": self.unique_id,
|
|
"command_topic": self.command_topic,
|
|
"state_topic": self.state_topic,
|
|
"brightness": True,
|
|
"color_mode": True,
|
|
"supported_color_modes": ["rgbw"],
|
|
"effect": True,
|
|
"effect_list": effect_names(),
|
|
"device": self.device_block(),
|
|
}
|
|
|
|
def command_topics(self) -> tuple[str, ...]:
|
|
return (self.command_topic,)
|
|
|
|
def subscribe(self) -> None:
|
|
self.bus.subscribe(LedEffectChanged, self._on_led_changed)
|
|
|
|
# ----------------------------------------------------------------- commands
|
|
|
|
async def handle(self, topic: str, payload: str) -> None:
|
|
try:
|
|
command = json.loads(payload)
|
|
except json.JSONDecodeError:
|
|
_log.warning("Ignoring non-JSON command on %s: %r", topic, payload[:120])
|
|
return
|
|
if not isinstance(command, dict):
|
|
_log.warning("Ignoring command on %s: expected an object, got %r", topic, command)
|
|
return
|
|
|
|
self._remember_previous_color(command)
|
|
self._state.update(command)
|
|
self.mouse.set_effect(self.zone, self._build_effect(), origin="mqtt")
|
|
# No publish here: LedEffectChanged will report what the device actually did.
|
|
|
|
def _remember_previous_color(self, command: dict[str, Any]) -> None:
|
|
"""Two-colour effects interpolate from the colour that was set before."""
|
|
if "color" not in command:
|
|
return
|
|
brightness = command.get("brightness", self._state["brightness"])
|
|
new_color = _color_from_json(command["color"], brightness)
|
|
current = _color_from_json(self._state["color"], self._state["brightness"])
|
|
if new_color != current:
|
|
self._last_color = current
|
|
|
|
def _build_effect(self) -> LedEffect:
|
|
state = self._state
|
|
color = _color_from_json(state["color"], state["brightness"])
|
|
transition_ms = float(state.get("transition", DEFAULT_TRANSITION_S)) * 1000
|
|
effect = str(state.get("effect", "static"))
|
|
|
|
if state["state"] == "OFF":
|
|
return _static(BLACK, transition_ms)
|
|
|
|
if (positional := parse_positional_effect(effect)) is not None:
|
|
begin, end, increment = positional
|
|
return EffectStaticDetailedConfig(
|
|
color,
|
|
increment=increment,
|
|
begin=begin,
|
|
end=end,
|
|
transition_time_in_ms=transition_ms,
|
|
)
|
|
|
|
match effect:
|
|
case "static":
|
|
return _static(color, transition_ms)
|
|
case "circular":
|
|
return EffectCircularConfig(speed=180, width=90, color=color)
|
|
case "wipeup":
|
|
swipe_and_change = EffectSwipeAndChange()
|
|
swipe_and_change.swipe.primary_color = self._last_color
|
|
swipe_and_change.swipe.secondary_color = color
|
|
swipe_and_change.swipe.bell_curve_width_in_leds = 10
|
|
swipe_and_change.swipe.transition_width = 30
|
|
swipe_and_change.swipe.start_position = 0
|
|
swipe_and_change.swipe.swipe_speed = 260
|
|
swipe_and_change.change.color1 = color
|
|
swipe_and_change.change.color2 = self._last_color
|
|
return swipe_and_change
|
|
case "twocolor" | "twocolorrandom":
|
|
random_hues = effect == "twocolorrandom"
|
|
return EffectRandomTwoColorInterpolationConfig(
|
|
color1=color,
|
|
color2=self._last_color,
|
|
hue1_random=random_hues,
|
|
hue2_random=random_hues,
|
|
start_with_existing=True,
|
|
)
|
|
case _:
|
|
_log.warning("Unknown effect %r on %s, turning it off", effect, self.zone)
|
|
return _static(BLACK, transition_ms)
|
|
|
|
# -------------------------------------------------------------------- state
|
|
|
|
async def _on_led_changed(self, event: LedEffectChanged) -> None:
|
|
if event.zone is not self.zone:
|
|
return
|
|
if event.origin != "mqtt":
|
|
self._reconcile(event.effect)
|
|
await self.publish_state()
|
|
|
|
def _reconcile(self, effect: LedEffect) -> None:
|
|
"""Fold an effect this entity did not ask for into the reported state.
|
|
|
|
The mapping is lossy - the firmware has richer effects than the HA light
|
|
schema - so only on/off and a colour are taken. The effect *name* is left
|
|
alone, since reporting one outside ``effect_list`` would confuse HA.
|
|
"""
|
|
color = getattr(effect, "color", None)
|
|
if isinstance(effect, EffectStaticConfig | EffectStaticDetailedConfig) and color == BLACK:
|
|
self._state["state"] = "OFF"
|
|
return
|
|
|
|
self._state["state"] = "ON"
|
|
if isinstance(color, ColorRGBW):
|
|
self._state["color"] = _color_to_json(color)
|
|
self._state["brightness"] = 255
|
|
|
|
async def publish_state(self) -> None:
|
|
await self.publish(self.state_topic, self._state)
|
|
|
|
|
|
def _static(color: ColorRGBW, transition_ms: float) -> LedEffect:
|
|
if transition_ms > 0:
|
|
return EffectStaticDetailedConfig(color, transition_time_in_ms=transition_ms)
|
|
return EffectStaticConfig(color)
|
|
|
|
|
|
def _color_from_json(color: dict[str, int], brightness: int = 255) -> ColorRGBW:
|
|
scale = brightness / 255
|
|
r, g, b, w = ((color.get(channel, 0) / 255) * scale for channel in "rgbw")
|
|
return ColorRGBW(r, g, b, w)
|
|
|
|
|
|
def _color_to_json(color: ColorRGBW) -> dict[str, int]:
|
|
return {
|
|
"r": round(color.r * 255),
|
|
"g": round(color.g * 255),
|
|
"b": round(color.b * 255),
|
|
"w": round(color.w * 255),
|
|
}
|