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
|
from dataclasses import dataclass
from datetime import datetime
from enum import IntEnum, Enum, auto
from typing import ClassVar, Generic, TypeVar
def pretty_name(name: str):
data = name.split("_")
return " ".join(f"{part[0].upper()}{part[1:]}" for part in data)
class ProductType(Enum):
MOWER = auto()
WATER_COMPUTER = auto()
VALVE = auto()
PUMP = auto()
@staticmethod
def from_manufacturer_data(data: "ManufacturerData") -> "ProductType":
if data.group == 10:
return ProductType.MOWER
if data.group == 18 and data.model in (0, 1) and data.variant == 1:
return ProductType.WATER_COMPUTER
if data.group == 18 and data.model == 2 and data.variant == 1:
return ProductType.VALVE
if data.group == 17 and data.model == 1:
return ProductType.PUMP
return None
CharacteristicType = TypeVar("CharacteristicType")
@dataclass
class Characteristic(Generic[CharacteristicType]):
uuid: str
name: str = ""
registry: ClassVar[dict[str, "Characteristic"]] = {}
def __set_name__(self, _, name: str):
self.name = pretty_name(name)
def __post_init__(self):
self.registry[self.uuid] = self
@classmethod
def decode(cls, data: bytes) -> CharacteristicType:
raise NotImplementedError(f"Decoding of {type(cls)} is not implemented")
@classmethod
def encode(cls, data: CharacteristicType) -> bytes:
raise NotImplementedError(f"Encoding of {type(cls)} is not implemented")
@dataclass
class CharacteristicBytes(Characteristic[bytes]):
@classmethod
def decode(cls, data: bytes) -> bytes:
return data
@classmethod
def encode(cls, value: bytes) -> bytes:
return value
@dataclass
class CharacteristicBool(Characteristic[bool]):
@classmethod
def decode(cls, data: bytes) -> bool:
return data[0] != 0
@classmethod
def encode(cls, data: bool) -> bytes:
if data:
return b"\x01"
return b"\x00"
@dataclass
class CharacteristicString(Characteristic[str]):
@classmethod
def decode(cls, data: bytes) -> str:
return data.decode("ASCII", "replace")
@classmethod
def encode(cls, value: str) -> bytes:
return value.encode("ASCII")
@dataclass
class CharacteristicNullString(Characteristic[str]):
@classmethod
def decode(cls, data: bytes) -> str:
return data.partition(b"\x00")[0].decode("ASCII", "replace")
@classmethod
def encode(cls, value: str) -> bytes:
return value.encode("ASCII")
@dataclass
class CharacteristicNullStringUf8(Characteristic[str]):
@classmethod
def decode(cls, data: bytes) -> str:
return data.partition(b"\x00")[0].decode("utf-8", "replace")
@classmethod
def encode(cls, value: str) -> bytes:
return value.encode("utf-8")
@dataclass
class CharacteristicInt(Characteristic[int]):
@classmethod
def decode(cls, data: bytes) -> int:
return int.from_bytes(data, "little", signed=True)
@classmethod
def encode(cls, value: int) -> bytes:
return value.to_bytes(1, "little", signed=True)
@dataclass
class CharacteristicLong(Characteristic[int]):
@classmethod
def decode(cls, data: bytes) -> int:
return int.from_bytes(data, "little", signed=True)
@classmethod
def encode(cls, value: int) -> bytes:
return value.to_bytes(4, "little", signed=True)
@dataclass
class CharacteristicUInt16(Characteristic[int]):
@classmethod
def decode(cls, data: bytes) -> int:
return int.from_bytes(data, "little", signed=False)
@classmethod
def encode(cls, value: int) -> bytes:
return value.to_bytes(2, "little", signed=False)
@dataclass
class CharacteristicLongArray(Characteristic[list[int]]):
@classmethod
def decode(cls, data: bytes) -> list[int]:
return [
int.from_bytes(data[i : i + 4], "little", signed=True)
for i in range(0, len(data), 4)
]
@dataclass
class CharacteristicTime(Characteristic[datetime]):
@classmethod
def decode(cls, data: bytes) -> datetime:
value = int.from_bytes(data, "little")
return datetime.fromtimestamp(value)
@classmethod
def encode(cls, value: datetime) -> bytes:
return int(value.timestamp()).to_bytes(4, "little", signed=True)
@dataclass
class CharacteristicTimeArray(Characteristic[list[datetime]]):
@classmethod
def decode(cls, data: bytes) -> list[datetime]:
return [
datetime.fromtimestamp(value)
for value in CharacteristicLongArray.decode(data)
]
class Service:
uuid: ClassVar[str]
registry: ClassVar[dict[str, "Service"]] = {}
def __init_subclass__(cls, /, **kwargs):
super().__init_subclass__(**kwargs)
cls.registry[cls.uuid] = cls
@classmethod
def characteristics(cls):
for value in vars(cls).values():
if isinstance(value, Characteristic):
yield value
class EnumOrInt(IntEnum):
@classmethod
def enum_or_int(cls, value: int):
try:
return cls(value)
except ValueError:
return value
class ProductGroup(EnumOrInt):
MOWER = 1
GARDEN_PUMP = 17
WATER_CONTROL = 18
@dataclass
class ManufacturerData:
company: ClassVar[int] = 0x0426
pairable: bool | None
serial: int | None
group: int | ProductGroup | None
model: int | None
variant: int | None
@staticmethod
def decode_dict(data: bytes):
res: dict[int, bytes] = {}
idx = 0
while idx < len(data):
size = data[idx]
key = data[idx + 1]
res[key] = data[idx + 2 : idx + size + 1]
idx += size + 1
return res
@staticmethod
def decode(data: bytes):
value = ManufacturerData.decode_dict(data)
info = dict(enumerate(value.get(6, b"")))
serial = value.get(4)
pairable = value.get(5)
return ManufacturerData(
pairable=bool.from_bytes(pairable, "little") if pairable else None,
serial=int.from_bytes(serial, "little") if serial else None,
group=ProductGroup.enum_or_int(info.get(0)),
model=info.get(1),
variant=info.get(2),
)
|
