-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
83 lines (60 loc) · 2.69 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
#!/usr/bin/env python
import os
import logging
from mqttwrapper import run_script
from asgiref.sync import async_to_sync
from pytradfri.api.aiocoap_api import APIFactory
from pytradfri import Gateway
GATEWAY_IP = os.environ['GATEWAY_IP']
GATEWAY_ID = os.environ['GATEWAY_ID']
GATEWAY_PSK = os.environ['GATEWAY_PSK']
LIGHT_IDS = os.environ['LIGHT_IDS']
OUTLET_IDS = os.environ['OUTLET_IDS']
def handle_message(topic: str, payload: bytes, lights, outlets):
device_type, device, attribute = topic.split("/")[1:]
device_id = lights[device] if device_type == "lighting" else outlets[device]
if attribute == "power":
value = payload == b'1'
elif attribute == "brightness":
value = max(0, min(254, int(float(payload))))
if device_type == "switch":
tradfri_set_outlet_state(device_id, value)
elif device_type == "lighting" and attribute == "power":
tradfri_set_light_state(device_id, value)
elif device_type == "lighting" and attribute == "brightness":
tradfri_set_light_dimmer(device_id, value)
async def tradfri_get_api_device(device_id):
api_factory = APIFactory(host=GATEWAY_IP, psk_id=GATEWAY_ID, psk=GATEWAY_PSK)
api = api_factory.request
gateway = Gateway()
device = await api(gateway.get_device(device_id))
return api, device
@async_to_sync
async def tradfri_set_outlet_state(device_id, value):
api, device = await tradfri_get_api_device(device_id)
await api(device.socket_control.set_state(value))
@async_to_sync
async def tradfri_set_light_state(device_id, value):
api, device = await tradfri_get_api_device(device_id)
await api(device.light_control.set_state(value))
@async_to_sync
async def tradfri_set_light_dimmer(device_id, value):
api, device = await tradfri_get_api_device(device_id)
await api(device.light_control.set_dimmer(value))
def get_topics():
lights = []
for light in (x.split(":")[1] for x in LIGHT_IDS.split(",")):
lights.append("control/lighting/{}/power".format(light))
lights.append("control/lighting/{}/brightness".format(light))
outlets = []
for outlet in (x.split(":")[1] for x in OUTLET_IDS.split(",")):
outlets.append("control/switch/{}/power".format(outlet))
return lights + outlets
def main():
formatter = "[%(asctime)s] %(name)s %(levelname)s - %(message)s"
logging.basicConfig(level=logging.DEBUG, format=formatter)
lights = {name: device_id for device_id, name in (i.split(":") for i in LIGHT_IDS.split(","))}
outlets = {name: device_id for device_id, name in (i.split(":") for i in OUTLET_IDS.split(","))}
run_script(handle_message, topics=get_topics(), lights=lights, outlets=outlets)
if __name__ == '__main__':
main()