163 lines
6.2 KiB
Python
163 lines
6.2 KiB
Python
from led_cmds import ColorRGBW, EffectStaticConfig, EffectCircularConfig, EffectRandomTwoColorInterpolationConfig, EffectAlexaSwipeConfig, EffectSwipeAndChange
|
|
import asyncio
|
|
import asyncio_mqtt
|
|
import json
|
|
from copy import deepcopy
|
|
|
|
|
|
class ShelveLightMqtt:
|
|
def __init__(self, protocol, client):
|
|
self._protocol = protocol
|
|
self._mqtt_client = client
|
|
|
|
self._state = {
|
|
"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)
|
|
|
|
self._discovery_spec = self._create_discovery_msg_light()
|
|
|
|
async def init(self):
|
|
"""Init method, because constructor can't be async"""
|
|
self._protocol.shelve_led_effect(EffectStaticConfig(ColorRGBW(0, 0, 0, 0)))
|
|
await self._send_autodiscovery_msg()
|
|
await self._notify_mqtt_state({"state": "OFF"})
|
|
|
|
async def handle_light_message(self, msg):
|
|
if msg.topic == self._discovery_spec['command_topic']:
|
|
payload = msg.payload.decode()
|
|
new_state = json.loads(payload)
|
|
print("IN ", new_state)
|
|
await self._update_state(new_state)
|
|
await self._notify_mqtt_state(new_state)
|
|
|
|
async def _update_state(self, new_state):
|
|
"""Merges current state with new state, updates device"""
|
|
|
|
# memorize last color - this is used for effects that need 2 colors
|
|
if 'color' in new_state:
|
|
brightness = new_state.get('brightness', self._state['brightness'])
|
|
new_color = self._color_from_json(new_state['color'], brightness)
|
|
current_color = self._color_from_json(self._state['color'])
|
|
if new_color != current_color:
|
|
self._last_color = current_color
|
|
print("last color", self._last_color)
|
|
|
|
self._state.update(new_state)
|
|
self._update_device()
|
|
|
|
@staticmethod
|
|
def _color_from_json(json_color, brightness=255):
|
|
args = ((json_color[e] / 255) * (brightness / 255) for e in ('r', 'g', 'b', 'w'))
|
|
return ColorRGBW(*args)
|
|
|
|
def _update_device(self):
|
|
s = self._state
|
|
current_color = self._color_from_json(s['color'], brightness=s["brightness"])
|
|
|
|
if s['state'] == "OFF":
|
|
eff = EffectStaticConfig(ColorRGBW(0, 0, 0, 0))
|
|
elif s['effect'] == 'static':
|
|
eff = EffectStaticConfig(current_color)
|
|
elif s['effect'] == 'circular':
|
|
eff = EffectCircularConfig(speed=180, width=90, color=current_color)
|
|
elif s['effect'] == 'wipeup':
|
|
eff = EffectSwipeAndChange()
|
|
eff.swipe.secondary_color = current_color
|
|
eff.swipe.primary_color = self._last_color
|
|
eff.swipe.bell_curve_width_in_leds = 5
|
|
eff.swipe.swipe_speed = 180
|
|
eff.change.color1 = current_color
|
|
eff.change.color2 = self._last_color
|
|
elif s['effect'] == "twocolor":
|
|
eff = EffectRandomTwoColorInterpolationConfig()
|
|
eff.color1 = current_color
|
|
eff.color2 = self._last_color
|
|
eff.start_with_existing = True
|
|
elif s['effect'] == "twocolorrandom":
|
|
eff = EffectRandomTwoColorInterpolationConfig()
|
|
eff.color1 = current_color
|
|
eff.color2 = self._last_color
|
|
eff.hue1_random = True
|
|
eff.hue2_random = True
|
|
eff.start_with_existing = True
|
|
else:
|
|
print(f"Unknown effect {s['effect']}")
|
|
eff = EffectStaticConfig(ColorRGBW(0, 0, 0, 0))
|
|
self._protocol.shelve_led_effect(eff)
|
|
|
|
@staticmethod
|
|
def _create_discovery_msg_light(base_name="musicmouse_json",
|
|
display_name="Music Mouse Regal Licht"):
|
|
id = "shelve"
|
|
return {
|
|
'platform': 'mqtt',
|
|
'schema': 'json',
|
|
'name': display_name,
|
|
'unique_id': f'{base_name}_{id}',
|
|
'command_topic': f'{base_name}/lights_{id}/command',
|
|
'state_topic': f'{base_name}/lights_{id}/state',
|
|
'color_mode': True,
|
|
'brightness': True,
|
|
#'device': {
|
|
# 'manufacturer': 'bauer.tech',
|
|
# 'model': "SK6812 LED strip",
|
|
#},
|
|
'effect': True,
|
|
'effect_list': ['static', 'circular', 'wipeup', 'twocolor', 'twocolorrandom'],
|
|
'supported_color_modes': ['rgbw'],
|
|
}
|
|
|
|
async def _send_autodiscovery_msg(self):
|
|
topic = f"homeassistant/light/{self._discovery_spec['unique_id']}/config"
|
|
await self._mqtt_client.publish(topic, json.dumps(self._discovery_spec).encode())
|
|
|
|
async def _notify_mqtt_state(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())
|
|
return
|
|
|
|
direct_ack = False
|
|
if direct_ack == True:
|
|
state_payload = json.dumps(state)
|
|
else:
|
|
s = deepcopy(self._state)
|
|
if s['state'] == "OFF":
|
|
state_payload = json.dumps({"state": "OFF"})
|
|
else:
|
|
s['color_mode'] = "rgbw"
|
|
state_payload = json.dumps(s)
|
|
|
|
print("OUT ", state_payload)
|
|
await self._mqtt_client.publish(self._discovery_spec['state_topic'], state_payload.encode())
|
|
|
|
|
|
async def start_mqtt(music_mouse_protocol, server, username, password):
|
|
async with asyncio_mqtt.Client(hostname=server, username=username, password=password) as client:
|
|
shelve_light = ShelveLightMqtt(music_mouse_protocol, client)
|
|
await shelve_light.init()
|
|
async with client.filtered_messages("musicmouse_json/#") as messages:
|
|
await client.subscribe("musicmouse_json/#")
|
|
async for message in messages:
|
|
await shelve_light.handle_light_message(message)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
class DummyProtocol:
|
|
def shelve_led_effect(self, effect):
|
|
print("EFF ", repr(effect))
|
|
|
|
password = ""
|
|
asyncio.run(start_mqtt(DummyProtocol(), "homeassistant", "musicmouse", password))
|