|
20
|
1 import logging
|
|
15
|
2 import json
|
|
20
|
3 import aiohttp
|
|
15
|
4 from color_convert import DeviceColor
|
|
|
5 from mqtt_io import MqttIo
|
|
|
6
|
|
20
|
7 log = logging.getLogger('prot')
|
|
15
|
8
|
|
|
9 class Transport:
|
|
|
10
|
|
|
11 def linked(self):
|
|
|
12 return {'label': str(self)}
|
|
|
13
|
|
|
14 async def send(self, dc: DeviceColor):
|
|
|
15 raise TypeError
|
|
|
16
|
|
|
17
|
|
|
18 def zigbeeHexMessage(dc: DeviceColor) -> dict:
|
|
|
19 msg: dict = {"transition": 0, "brightness": int(255 * dc.brightness)}
|
|
|
20 c = "#%02x%02x%02x" % (int(dc.r * 255), int(dc.g * 255), int(dc.b * 255))
|
|
|
21 msg["color"] = {"hex": c}
|
|
|
22 return msg
|
|
|
23
|
|
|
24
|
|
|
25 class ZigbeeTransport(Transport):
|
|
|
26
|
|
|
27 def __init__(self, mqtt: MqttIo, name: str, ieee: str):
|
|
|
28 self.mqtt = mqtt
|
|
|
29 self.name = name
|
|
|
30 self.ieee = ieee
|
|
|
31
|
|
|
32 def linked(self):
|
|
20
|
33 return {'url': f'https://bigasterisk.com/zigbee/console/#/device/{self.ieee}/info', 'label': self.name}
|
|
15
|
34
|
|
|
35 async def send(self, dc: DeviceColor):
|
|
|
36 await self.mqtt.publish(f'zigbee/{self.name}/set', json.dumps(zigbeeHexMessage(dc)))
|
|
20
|
37
|
|
|
38
|
|
|
39 class SonoffRelayTransport(Transport):
|
|
|
40
|
|
|
41 def __init__(self, mqtt: MqttIo, name: str):
|
|
|
42 self.mqtt = mqtt
|
|
|
43 self.name = name
|
|
|
44
|
|
|
45 def linked(self):
|
|
|
46 return {'label': self.name}
|
|
|
47
|
|
|
48 async def send(self, dc: DeviceColor):
|
|
|
49 topic = f'{self.name}/switch/sonoff_basic_relay/command'
|
|
|
50 msg = 'ON' if dc.brightness else 'OFF'
|
|
|
51 log.info(f'sonoff {topic=} {msg=}')
|
|
|
52 await self.mqtt.publish(topic, msg)
|
|
|
53
|
|
|
54
|
|
|
55 class _WebTransport(Transport):
|
|
|
56
|
|
|
57 def __init__(self, hostname: str):
|
|
|
58 self.hostname = hostname
|
|
|
59 self._session = aiohttp.ClientSession()
|
|
|
60
|
|
|
61 def linked(self):
|
|
|
62 return {'url': f'http://{self.hostname}/', 'label': self.hostname}
|
|
|
63
|
|
|
64
|
|
|
65 class TasmotaWebTransport(_WebTransport):
|
|
|
66
|
|
|
67 async def send(self, dc: DeviceColor):
|
|
|
68 cmnd = 'Color ' + ','.join(str(int(x * 255)) for x in (dc.r, dc.g, dc.b, dc.cw, dc.ww))
|
|
|
69 async with self._session.get(f'http://{self.hostname}/cm', params={'cmnd': cmnd}) as resp:
|
|
|
70 await resp.text()
|
|
|
71 # {"POWER":"ON","Dimmer":21,"Color":"3636363600","HSBColor":"0,0,21","White":21,"CT":153,"Channel":[21,21,21,21,0]}
|
|
|
72
|
|
|
73
|
|
|
74 class ShellyGen1WebTransport(_WebTransport):
|
|
|
75
|
|
|
76 async def send(self, dc: DeviceColor):
|
|
|
77 # also see https://shelly-api-docs.shelly.cloud/gen1/#shelly-rgbw2-color-status for metrics
|
|
|
78 async with self._session.get(f'http://{self.hostname}/light/0',
|
|
|
79 params={
|
|
|
80 'red': int(dc.r * 255),
|
|
|
81 'green': int(dc.g * 255),
|
|
|
82 'blue': int(dc.b * 255),
|
|
|
83 'white': int(dc.w * 255),
|
|
|
84 }) as resp:
|
|
|
85 await resp.text()
|
|
|
86 # {..."mode":"color","red":255,"green":242,"blue":0,"white":255,"gain":59,"effect":0,"transition":0,"power":18.00,"overpower":false}
|
