|
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
|
|
21
|
9
|
|
15
|
10 class Transport:
|
|
|
11
|
|
|
12 def linked(self):
|
|
|
13 return {'label': str(self)}
|
|
|
14
|
|
|
15 async def send(self, dc: DeviceColor):
|
|
|
16 raise TypeError
|
|
|
17
|
|
|
18
|
|
21
|
19 def to8(x: float):
|
|
|
20 return int(x * 255)
|
|
|
21
|
|
|
22
|
|
|
23 def zbColorMessage(dc: DeviceColor) -> dict:
|
|
|
24 return {
|
|
|
25 "transition": 0,
|
|
|
26 "brightness": to8(dc.brightness),
|
|
|
27 "color": {
|
|
|
28 "hex": "#%02x%02x%02x" % (to8(dc.r), to8(dc.g), to8(dc.b))
|
|
|
29 },
|
|
|
30 }
|
|
|
31
|
|
15
|
32
|
|
21
|
33 def zbBrightnessMessage(dc: DeviceColor) -> dict:
|
|
|
34 return {
|
|
|
35 "transition": 0,
|
|
|
36 "brightness": to8(dc.brightness),
|
|
|
37 }
|
|
|
38
|
|
|
39
|
|
|
40 def zbWhiteSpectrumMessage(dc: DeviceColor) -> dict:
|
|
|
41 return {
|
|
|
42 "transition": 0,
|
|
|
43 "brightness": to8(dc.brightness),
|
|
|
44 # temperature todo
|
|
|
45 }
|
|
|
46
|
|
|
47 def zbRelayMessage(dc: DeviceColor) -> dict:
|
|
|
48 return {'state': 'ON' if dc.brightness else 'OFF'}
|
|
|
49
|
|
|
50 def z2mSet(name):
|
|
|
51 return f'zigbee/{name}/set'
|
|
15
|
52
|
|
|
53 class ZigbeeTransport(Transport):
|
|
|
54
|
|
21
|
55 def __init__(self, mqtt: MqttIo, name: str, ieee: str, topic=z2mSet, msg=zbColorMessage):
|
|
15
|
56 self.mqtt = mqtt
|
|
|
57 self.name = name
|
|
|
58 self.ieee = ieee
|
|
21
|
59 self.topic=topic
|
|
|
60 self.msg = msg
|
|
15
|
61
|
|
|
62 def linked(self):
|
|
20
|
63 return {'url': f'https://bigasterisk.com/zigbee/console/#/device/{self.ieee}/info', 'label': self.name}
|
|
15
|
64
|
|
|
65 async def send(self, dc: DeviceColor):
|
|
21
|
66 await self.mqtt.publish(self.topic(self.name), json.dumps(self.msg(dc)))
|
|
20
|
67
|
|
|
68
|
|
|
69 class SonoffRelayTransport(Transport):
|
|
|
70
|
|
|
71 def __init__(self, mqtt: MqttIo, name: str):
|
|
|
72 self.mqtt = mqtt
|
|
|
73 self.name = name
|
|
|
74
|
|
|
75 def linked(self):
|
|
|
76 return {'label': self.name}
|
|
|
77
|
|
|
78 async def send(self, dc: DeviceColor):
|
|
|
79 topic = f'{self.name}/switch/sonoff_basic_relay/command'
|
|
|
80 msg = 'ON' if dc.brightness else 'OFF'
|
|
|
81 log.info(f'sonoff {topic=} {msg=}')
|
|
|
82 await self.mqtt.publish(topic, msg)
|
|
|
83
|
|
|
84
|
|
|
85 class _WebTransport(Transport):
|
|
|
86
|
|
|
87 def __init__(self, hostname: str):
|
|
|
88 self.hostname = hostname
|
|
|
89 self._session = aiohttp.ClientSession()
|
|
|
90
|
|
|
91 def linked(self):
|
|
|
92 return {'url': f'http://{self.hostname}/', 'label': self.hostname}
|
|
|
93
|
|
|
94
|
|
|
95 class TasmotaWebTransport(_WebTransport):
|
|
|
96
|
|
|
97 async def send(self, dc: DeviceColor):
|
|
|
98 cmnd = 'Color ' + ','.join(str(int(x * 255)) for x in (dc.r, dc.g, dc.b, dc.cw, dc.ww))
|
|
|
99 async with self._session.get(f'http://{self.hostname}/cm', params={'cmnd': cmnd}) as resp:
|
|
|
100 await resp.text()
|
|
|
101 # {"POWER":"ON","Dimmer":21,"Color":"3636363600","HSBColor":"0,0,21","White":21,"CT":153,"Channel":[21,21,21,21,0]}
|
|
|
102
|
|
|
103
|
|
|
104 class ShellyGen1WebTransport(_WebTransport):
|
|
|
105
|
|
|
106 async def send(self, dc: DeviceColor):
|
|
|
107 # also see https://shelly-api-docs.shelly.cloud/gen1/#shelly-rgbw2-color-status for metrics
|
|
|
108 async with self._session.get(f'http://{self.hostname}/light/0',
|
|
|
109 params={
|
|
|
110 'red': int(dc.r * 255),
|
|
|
111 'green': int(dc.g * 255),
|
|
|
112 'blue': int(dc.b * 255),
|
|
|
113 'white': int(dc.w * 255),
|
|
|
114 }) as resp:
|
|
|
115 await resp.text()
|
|
|
116 # {..."mode":"color","red":255,"green":242,"blue":0,"white":255,"gain":59,"effect":0,"transition":0,"power":18.00,"overpower":false}
|
