-
Notifications
You must be signed in to change notification settings - Fork 0
/
bulbs.py
executable file
·178 lines (147 loc) · 5.07 KB
/
bulbs.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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
#!/usr/bin/env python
from colors import colors
from functools import partial
from multiprocessing import Pool
from wyze_sdk import Client
from wyze_sdk.errors import WyzeApiError
import json
import logging
import os
import sys
def create_client():
if os.path.exists('tokens.json'):
with open('tokens.json') as f:
data = json.load(f)
token = data['access_token']
refresh_token = data['refresh_token']
return Client(token=token, refresh_token=refresh_token)
with open('credentials.json') as fp:
cred = json.load(fp)
response = Client().login(
email=cred['email'],
password=cred['password']
)
return Client(token=response['access_token'],
refresh_token=response['refresh_token'])
def write_tokens(client):
with open('tokens.json', 'w') as f:
data = {
'access_token': client._token,
'refresh_token': client._refresh_token
}
json.dump(data, f)
def update_token(client):
"""Return if the token was updated."""
try:
client.user_get_info()
except WyzeApiError as e:
if 'refresh the token' in str(e):
logging.info('refreshing auth token')
client.refresh_token()
write_tokens(client)
return True
else:
raise e
return False
class Bulb:
def __init__(self, client, bulb):
self.client = client
self.nickname = bulb.nickname
self.mac = bulb.mac
self.model = bulb.product.model
self.info = None
def get_info(self):
self.info = self.client.bulbs.info(device_mac=self.mac)
self.brightness = self.info.brightness
self.color = self.info.color
self.color_temp = self.info.color_temp
self.is_on = self.info.is_on
def turn_on(self):
self.client.bulbs.turn_on(device_mac=self.mac, device_model=self.model)
def turn_off(self):
self.client.bulbs.turn_off(device_mac=self.mac, device_model=self.model)
def set_values(self, state):
if not state['is_on']:
self.turn_off()
return
self.turn_on()
if 'color_name' in state:
color = colors[state['color_name']]['color']
color_temp = colors[state['color_name']]['color_temp']
else:
color = state['color']
color_temp = state['color_temp']
self.client.bulbs.set_color(device_mac=self.mac,
device_model=self.model,
color=color)
self.client.bulbs.set_color_temp(device_mac=self.mac,
device_model=self.model,
color_temp=color_temp)
self.client.bulbs.set_brightness(device_mac=self.mac,
device_model=self.model,
brightness=state['brightness'])
def to_dict(self):
return {
'nickname': self.nickname,
'is_on': self.is_on,
'brightness': self.brightness,
'color': self.color,
'color_temp': self.color_temp
}
def __repr__(self):
return str(self.to_dict())
def confirm(message):
r = input(f'{message} [Y/n] ').lower()
return r == 'y' or r == 'yes'
def info_dict(bulb):
bulb.get_info()
return bulb.to_dict()
def save_state(bulbs, filename):
filename += '.json'
scene_file = os.path.join('scenes', filename)
if (os.path.exists(scene_file) and
not confirm(f'{filename} already exists, override?')):
return
with Pool(20) as p:
state = p.map(info_dict, bulbs)
with open(scene_file, 'w') as fp:
json.dump(state, fp, sort_keys=True, indent=2)
def set_bulb(client, state):
bulb = bulb_from_nickname(client, state['nickname'])
bulb.set_values(state)
def load_state(client, filename):
filename += '.json'
try:
with open(os.path.join('scenes', filename)) as fp:
states = json.load(fp)
except FileNotFoundError:
raise KeyError(f'No scene named {filename}')
with Pool(20) as p:
p.map(partial(set_bulb, client), states)
def filter_bulbs(client, prefix=''):
"""Return bulbs whose name start with @prefix."""
for bulb in client.bulbs.list():
if bulb.nickname.startswith(prefix):
yield Bulb(client, bulb)
def bulb_from_nickname(client, name):
for bulb in client.bulbs.list():
if bulb.nickname == name:
return Bulb(client, bulb)
if __name__ == '__main__':
if len(sys.argv) < 3:
print('load or save')
sys.exit(1)
client = create_client()
update_token(client)
filename = sys.argv[2]
try:
if sys.argv[1] == 'save':
bulbs = filter_bulbs(client)
save_state(bulbs, filename)
elif sys.argv[1] == 'load':
load_state(client, filename)
else:
print(f'Unrecognized command {sys.argv[1]}')
sys.exit(1)
except WyzeApiError as e:
print(e)