blob: 0e43f1b0d40736a826e68816e3508197eac4ca1e (
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
|
from enum import auto
from .._node import MqttNode
from .._module import MqttModule
from .._payload import MqttPayload
from ...util import HashableEnum
from typing import Optional
two_digits_precision = lambda x: round(x, 2)
MODULE_NAME = 'MqttTempHumModule'
class MqttTemphumDataPayload(MqttPayload):
FORMAT = '=ddb'
UNPACKER = {
'temp': two_digits_precision,
'rh': two_digits_precision
}
temp: float
rh: float
error: int
class MqttTempHumNodes(HashableEnum):
KBN_SH_HALL = auto()
KBN_SH_BATHROOM = auto()
KBN_SH_LIVINGROOM = auto()
KBN_SH_BEDROOM = auto()
KBN_BH_2FL = auto()
KBN_BH_2FL_STREET = auto()
KBN_BH_1FL_LIVINGROOM = auto()
KBN_BH_1FL_BEDROOM = auto()
KBN_BH_1FL_BATHROOM = auto()
KBN_NH_1FL_INV = auto()
KBN_NH_1FL_CENTER = auto()
KBN_NH_1LF_KT = auto()
KBN_NH_1FL_DS = auto()
KBN_NH_1FS_EZ = auto()
SPB_FLAT120_CABINET = auto()
class MqttTempHumModule(MqttModule):
def init(self, mqtt: MqttNode):
mqtt.subscribe_module('temphum/data', self)
def handle_payload(self,
mqtt: MqttNode,
topic: str,
payload: bytes) -> Optional[MqttPayload]:
if topic == 'temphum/data':
message = MqttTemphumDataPayload.unpack(payload)
self._logger.debug(message)
return message
|
