Mercurial > code > home > repos > light-bridge

comparison protocols.py @ 20:24a574108365

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
more protocols; bugs in setColor
author drewp@bigasterisk.com
date Mon, 29 Jan 2024 11:52:43 -0800
parents 61d4ccecfed8
children b8201490c731
comparison
equal deleted inserted replaced
19:4e350dcdc4fe 20:24a574108365
1 import logging
1 import json 2 import json
3 import aiohttp
2 from color_convert import DeviceColor 4 from color_convert import DeviceColor
3 from mqtt_io import MqttIo 5 from mqtt_io import MqttIo
4 6
7 log = logging.getLogger('prot')
5 8
6 class Transport: 9 class Transport:
7 10
8 def linked(self): 11 def linked(self):
9 return {'label': str(self)} 12 return {'label': str(self)}
25 self.mqtt = mqtt 28 self.mqtt = mqtt
26 self.name = name 29 self.name = name
27 self.ieee = ieee 30 self.ieee = ieee
28 31
29 def linked(self): 32 def linked(self):
30 return {'url': f'https://bigasterisk.com/zigbee/console/#/device/{self.ieee}/info', 'label': 'do-bar'} 33 return {'url': f'https://bigasterisk.com/zigbee/console/#/device/{self.ieee}/info', 'label': self.name}
31 34
32 async def send(self, dc: DeviceColor): 35 async def send(self, dc: DeviceColor):
33 await self.mqtt.publish(f'zigbee/{self.name}/set', json.dumps(zigbeeHexMessage(dc))) 36 await self.mqtt.publish(f'zigbee/{self.name}/set', json.dumps(zigbeeHexMessage(dc)))
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}