-
Notifications
You must be signed in to change notification settings - Fork 0
/
bulb_ip.py
113 lines (96 loc) · 2.8 KB
/
bulb_ip.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
from yeelight import Bulb, BulbException
from os import remove, environ as env
IP_PATH = env['HOME'] + '/.cache/bulb_ip'
def discover_bulbs():
import concurrent.futures
def test_ip(i):
try:
ip = f'192.168.1.{i}'
b = Bulb(ip)
print(b.get_properties())
return ip
except BulbException as e:
return None
ips = []
print("Looking for a new ip")
with concurrent.futures.ThreadPoolExecutor(max_workers=255/2) as executor:
futs = [executor.submit(test_ip, i) for i in range(1, 255)]
for f in futs:
ip = f.result()
if ip:
ips.append(ip)
return ips
class Cache:
@staticmethod
def refresh_ip():
try:
ip = discover_bulbs()[0]
except StopIteration:
print('No bulbs found')
return None
print(f'Found ip {ip}')
with open(IP_PATH, 'w') as f:
f.write(ip)
return ip
@staticmethod
def get_ip():
try:
with open(IP_PATH, 'r') as f:
ip = f.read().strip()
print(f'Trying ip \'{ip}\'')
return ip if ip else Cache.refresh_ip()
except FileNotFoundError as e:
return Cache.refresh_ip()
@staticmethod
def invalidate_cache():
print("cache not invalidated because trash api")
Cache.refresh_ip()
#remove(IP_PATH)
class BulbIp:
def with_ip(ip):
b = BulbIp()
b.bulb = Bulb(ip)
return b
def __init__(self):
self.bulb = None
def run(self, func, retry=True):
bulb = self.load_bulb()
try:
return func(bulb)
except BulbException as e:
print('Failed with', e)
if retry:
print('Retrying...')
Cache.invalidate_cache()
self.bulb = None
return self.run(func, retry=False)
def load_bulb(self):
if self.bulb is None:
ip = Cache.get_ip()
if not ip:
ip = Cache.refresh_ip()
if not ip:
print('Giving up...')
return None
self.bulb = Bulb(ip)
return self.bulb
instance = BulbIp()
def run(func):
global instance
try:
return instance.run(func)
except Exception as e:
print('Failed with', e)
class BulbProxy:
def __getattribute__(self, name):
global instance
attr = Bulb.__getattribute__(instance.load_bulb(), name)
if hasattr(attr, '__call__'):
def f(*args, **kwargs):
return instance.run(lambda _: attr(*args, **kwargs))
return f
else:
return attr
def __dir__(self):
return dir(Bulb)
B = BulbProxy()