forked from blynkkk/lib-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
blynklib.py
357 lines (301 loc) · 12.5 KB
/
blynklib.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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
# Copyright (c) 2019 Anton Morozenko
# Copyright (c) 2015-2019 Volodymyr Shymanskyy.
# See the file LICENSE for copying permission.
__version__ = '0.2.4'
try:
import usocket as socket
import utime as time
import ustruct as struct
import uselect as select
from micropython import const
ticks_ms = time.ticks_ms
sleep_ms = time.sleep_ms
IOError = OSError
except ImportError:
import socket
import time
import struct
import select
const = lambda x: x
ticks_ms = lambda: int(time.time() * 1000)
sleep_ms = lambda x: time.sleep(x // 1000)
LOGO = """
___ __ __
/ _ )/ /_ _____ / /__
/ _ / / // / _ \\/ '_/
/____/_/\\_, /_//_/_/\\_\\
/___/ for Python v{}\n""".format(__version__)
def stub_log(*args):
pass
class BlynkError(Exception):
pass
class Protocol(object):
MSG_RSP = const(0)
MSG_LOGIN = const(2)
MSG_PING = const(6)
MSG_TWEET = const(12)
MSG_EMAIL = const(13)
MSG_NOTIFY = const(14)
MSG_BRIDGE = const(15)
MSG_HW_SYNC = const(16)
MSG_INTERNAL = const(17)
MSG_PROPERTY = const(19)
MSG_HW = const(20)
MSG_HEAD_LEN = const(5)
STATUS_INVALID_TOKEN = const(9)
STATUS_OK = const(200)
VPIN_MAX_NUM = const(32)
_msg_id = 1
def _get_msg_id(self, **kwargs):
if 'msg_id' in kwargs:
return kwargs['msg_id']
self._msg_id += 1
return self._msg_id if self._msg_id <= 0xFFFF else 1
def _pack_msg(self, msg_type, *args, **kwargs):
data = ('\0'.join([str(curr_arg) for curr_arg in args])).encode('utf-8')
return struct.pack('!BHH', msg_type, self._get_msg_id(**kwargs), len(data)) + data
def parse_response(self, rsp_data, msg_buffer):
msg_args = []
try:
msg_type, msg_id, h_data = struct.unpack('!BHH', rsp_data[:self.MSG_HEAD_LEN])
except Exception as p_err:
raise BlynkError('Message parse error: {}'.format(p_err))
if msg_id == 0:
raise BlynkError('invalid msg_id == 0')
elif h_data >= msg_buffer:
raise BlynkError('Command too long. Length = {}'.format(h_data))
elif msg_type in (self.MSG_RSP, self.MSG_PING, self.MSG_INTERNAL):
pass
elif msg_type in (self.MSG_HW, self.MSG_BRIDGE):
msg_body = rsp_data[self.MSG_HEAD_LEN: self.MSG_HEAD_LEN + h_data]
msg_args = [itm.decode('utf-8') for itm in msg_body.split(b'\0')]
else:
raise BlynkError("Unknown message type: '{}'".format(msg_type))
return msg_type, msg_id, h_data, msg_args
def heartbeat_msg(self, heartbeat, rcv_buffer):
return self._pack_msg(self.MSG_INTERNAL, 'ver', __version__, 'buff-in', rcv_buffer, 'h-beat', heartbeat,
'dev', 'python')
def login_msg(self, token):
return self._pack_msg(self.MSG_LOGIN, token)
def ping_msg(self):
return self._pack_msg(self.MSG_PING)
def response_msg(self, *args, **kwargs):
return self._pack_msg(self.MSG_RSP, *args, **kwargs)
def virtual_write_msg(self, v_pin, *val):
return self._pack_msg(self.MSG_HW, 'vw', v_pin, *val)
def virtual_sync_msg(self, *pins):
return self._pack_msg(self.MSG_HW_SYNC, 'vr', *pins)
def email_msg(self, to, subject, body):
return self._pack_msg(self.MSG_EMAIL, to, subject, body)
def tweet_msg(self, msg):
return self._pack_msg(self.MSG_TWEET, msg)
def notify_msg(self, msg):
return self._pack_msg(self.MSG_NOTIFY, msg)
def set_property_msg(self, pin, prop, *val):
return self._pack_msg(self.MSG_PROPERTY, pin, prop, *val)
class Connection(Protocol):
SOCK_MAX_TIMEOUT = const(5)
SOCK_TIMEOUT = 0.05
EAGAIN = const(11)
ETIMEDOUT = const(60)
RETRIES_TX_DELAY = const(2)
RETRIES_TX_MAX_NUM = const(3)
RECONNECT_SLEEP = const(1)
TASK_PERIOD_RES = const(50)
DISCONNECTED = const(0)
CONNECTING = const(1)
AUTHENTICATING = const(2)
AUTHENTICATED = const(3)
_state = None
_socket = None
_last_rcv_time = 0
_last_ping_time = 0
_last_send_time = 0
def __init__(self, token, server='blynk-cloud.com', port=80, heartbeat=10, rcv_buffer=1024, log=stub_log):
self.token = token
self.server = server
self.port = port
self.heartbeat = heartbeat
self.rcv_buffer = rcv_buffer
self.log = log
def _set_socket_timeout(self, timeout):
if getattr(self._socket, 'settimeout', None):
self._socket.settimeout(timeout)
else:
p = select.poll()
p.register(self._socket)
p.poll(int(timeout * 1000))
def send(self, data):
retries = self.RETRIES_TX_MAX_NUM
while retries > 0:
try:
retries -= 1
self._last_send_time = ticks_ms()
return self._socket.send(data)
except (IOError, OSError):
sleep_ms(self.RETRIES_TX_DELAY)
def receive(self, length, timeout):
d_buff = b''
try:
self._set_socket_timeout(timeout)
d_buff += self._socket.recv(length)
if len(d_buff) >= length:
d_buff = d_buff[:length]
return d_buff
except (IOError, OSError) as err:
if str(err) == 'timed out':
return b''
if str(self.EAGAIN) in str(err) or str(self.ETIMEDOUT) in str(err):
return b''
raise
def is_server_alive(self):
now = ticks_ms()
h_beat_ms = self.heartbeat * 1000
rcv_delta = now - self._last_rcv_time
ping_delta = now - self._last_ping_time
send_delta = now - self._last_send_time
if rcv_delta > h_beat_ms + (h_beat_ms // 2):
return False
if (ping_delta > h_beat_ms // 10) and (send_delta > h_beat_ms or rcv_delta > h_beat_ms):
self.send(self.ping_msg())
self.log('Heartbeat time: {}'.format(now))
self._last_ping_time = now
return True
def _get_socket(self):
try:
self._state = self.CONNECTING
self._socket = socket.socket()
self._socket.connect(socket.getaddrinfo(self.server, self.port)[0][4])
self._set_socket_timeout(self.SOCK_TIMEOUT)
self.log('Connected to blynk server')
except Exception as g_exc:
raise BlynkError('Connection with the Blynk server failed: {}'.format(g_exc))
def _authenticate(self):
self.log('Authenticating device...')
self._state = self.AUTHENTICATING
self.send(self.login_msg(self.token))
rsp_data = self.receive(self.rcv_buffer, self.SOCK_MAX_TIMEOUT)
if not rsp_data:
raise BlynkError('Auth stage timeout')
_, _, status, _ = self.parse_response(rsp_data, self.rcv_buffer)
if status != self.STATUS_OK:
if status == self.STATUS_INVALID_TOKEN:
raise BlynkError('Invalid Auth Token')
raise BlynkError('Auth stage failed. Status={}'.format(status))
self._state = self.AUTHENTICATED
self.log('Access granted')
def _set_heartbeat(self):
self.send(self.heartbeat_msg(self.heartbeat, self.rcv_buffer))
rcv_data = self.receive(self.rcv_buffer, self.SOCK_MAX_TIMEOUT)
if not rcv_data:
raise BlynkError('Heartbeat stage timeout')
_, _, status, _ = self.parse_response(rcv_data, self.rcv_buffer)
if status != self.STATUS_OK:
raise BlynkError('Set heartbeat returned code={}'.format(status))
self.log('Heartbeat = {} sec. MaxCmdBuffer = {} bytes'.format(self.heartbeat, self.rcv_buffer))
def connected(self):
return True if self._state == self.AUTHENTICATED else False
class Blynk(Connection):
_CONNECT_TIMEOUT = const(30) # 30sec
_VPIN_WILDCARD = '*'
_VPIN_READ = 'read v'
_VPIN_WRITE = 'write v'
_INTERNAL = 'internal_'
_CONNECT = 'connect'
_DISCONNECT = 'disconnect'
_VPIN_READ_ALL = '{}{}'.format(_VPIN_READ, _VPIN_WILDCARD)
_VPIN_WRITE_ALL = '{}{}'.format(_VPIN_WRITE, _VPIN_WILDCARD)
_events = {}
def __init__(self, token, **kwargs):
Connection.__init__(self, token, **kwargs)
self._start_time = ticks_ms()
self._last_rcv_time = ticks_ms()
self._last_send_time = ticks_ms()
self._last_ping_time = ticks_ms()
self._state = self.DISCONNECTED
print(LOGO)
def connect(self, timeout=_CONNECT_TIMEOUT):
end_time = time.time() + timeout
while not self.connected():
if self._state == self.DISCONNECTED:
try:
self._get_socket()
self._authenticate()
self._set_heartbeat()
self.log('Registered events: {}\n'.format(list(self._events.keys())))
self.call_handler(self._CONNECT)
return True
except BlynkError as b_err:
self.disconnect(b_err)
sleep_ms(self.TASK_PERIOD_RES)
if time.time() >= end_time:
return False
def disconnect(self, err_msg=None):
if self._socket:
self._socket.close()
self._state = self.DISCONNECTED
if err_msg:
self.log('[ERROR]: {}\nConnection closed'.format(err_msg))
time.sleep(self.RECONNECT_SLEEP)
self.call_handler(self._DISCONNECT)
def virtual_write(self, v_pin, *val):
return self.send(self.virtual_write_msg(v_pin, *val))
def virtual_sync(self, *v_pin):
return self.send(self.virtual_sync_msg(*v_pin))
def email(self, to, subject, body):
return self.send(self.email_msg(to, subject, body))
def tweet(self, msg):
return self.send(self.tweet_msg(msg))
def notify(self, msg):
return self.send(self.notify_msg(msg))
def set_property(self, v_pin, property_name, *val):
return self.send(self.set_property_msg(v_pin, property_name, *val))
def handle_event(blynk, event_name):
class Deco(object):
def __init__(self, func):
self.func = func
# wildcard 'read V*' and 'write V*' events handling
if str(event_name).lower() in (blynk._VPIN_READ_ALL, blynk._VPIN_WRITE_ALL):
event_base_name = str(event_name).split(blynk._VPIN_WILDCARD)[0]
for i in range(1, blynk.VPIN_MAX_NUM + 1):
blynk._events['{}{}'.format(event_base_name.lower(), i)] = func
else:
blynk._events[str(event_name).lower()] = func
def __call__(self):
return self.func()
return Deco
def call_handler(self, event, *args, **kwargs):
if event in self._events.keys():
self.log("Event: ['{}'] -> {}".format(event, args))
self._events[event](*args, **kwargs)
def process(self, msg_type, msg_id, msg_len, msg_args):
if msg_type == self.MSG_RSP:
self.log('Response status: {}'.format(msg_len))
elif msg_type == self.MSG_PING:
self.send(self.response_msg(self.STATUS_OK, msg_id=msg_id))
elif msg_type in (self.MSG_HW, self.MSG_BRIDGE, self.MSG_INTERNAL):
if msg_type == self.MSG_INTERNAL and len(msg_args) >= const(3):
self.call_handler("{}{}".format(self._INTERNAL, msg_args[1]), msg_args[2:])
elif len(msg_args) >= const(3) and msg_args[0] == 'vw':
self.call_handler("{}{}".format(self._VPIN_WRITE, msg_args[1]), int(msg_args[1]), msg_args[2:])
elif len(msg_args) == const(2) and msg_args[0] == 'vr':
self.call_handler("{}{}".format(self._VPIN_READ, msg_args[1]), int(msg_args[1]))
def run(self):
if not self.connected():
self.connect()
else:
try:
rsp_data = self.receive(self.rcv_buffer, self.SOCK_TIMEOUT)
self._last_rcv_time = ticks_ms()
if rsp_data:
msg_type, msg_id, h_data, msg_args = self.parse_response(rsp_data, self.rcv_buffer)
self.process(msg_type, msg_id, h_data, msg_args)
if not self.is_server_alive():
self.disconnect('Blynk server is offline')
except KeyboardInterrupt:
raise
except BlynkError as b_err:
self.log(b_err)
self.disconnect()
except Exception as g_exc:
self.log(g_exc)