blob: 171cdb994dc8d1bd37cc09d74f53a9c2fe4e2f88 (
plain)
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
|
import hashlib
from .base_payload import MqttPayload, MqttPayloadCustomField
class OTAResultPayload(MqttPayload):
FORMAT = '=BB'
result: int
error_code: int
class OTAPayload(MqttPayload):
secret: str
filename: str
# structure of returned data:
#
# uint8_t[len(secret)] secret;
# uint8_t[16] md5;
# *uint8_t data
def pack(self):
buf = bytearray(self.secret.encode())
m = hashlib.md5()
with open(self.filename, 'rb') as fd:
content = fd.read()
m.update(content)
buf.extend(m.digest())
buf.extend(content)
return buf
def unpack(cls, buf: bytes):
raise RuntimeError(f'{cls.__class__.__name__}.unpack: not implemented')
# secret = buf[:12].decode()
# filename = buf[12:].decode()
# return OTAPayload(secret=secret, filename=filename)
class DiagnosticsFlags(MqttPayloadCustomField):
state: bool
config_changed_value_present: bool
config_changed: bool
@staticmethod
def unpack(flags: int):
# _logger.debug(f'StatFlags.unpack: flags={flags}')
state = flags & 0x1
ccvp = (flags >> 1) & 0x1
cc = (flags >> 2) & 0x1
# _logger.debug(f'StatFlags.unpack: state={state}')
return DiagnosticsFlags(state=(state == 1),
config_changed_value_present=(ccvp == 1),
config_changed=(cc == 1))
def __index__(self):
bits = 0
bits |= (int(self.state) & 0x1)
bits |= (int(self.config_changed_value_present) & 0x1) << 1
bits |= (int(self.config_changed) & 0x1) << 2
return bits
class InitialDiagnosticsPayload(MqttPayload):
FORMAT = '=IBbIB'
ip: int
fw_version: int
rssi: int
free_heap: int
flags: DiagnosticsFlags
class DiagnosticsPayload(MqttPayload):
FORMAT = '=bIB'
rssi: int
free_heap: int
flags: DiagnosticsFlags
|
