blob: 5501f912a85cdd1a6d958f1d81a48baa9146e88b (
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
|
# from enum import auto
from .._node import MqttNode
from .._module import MqttModule
from .._payload import MqttPayload
# from ...util import HashableEnum
from typing import Optional
from ...temphum import BaseSensor
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, sensor: Optional[BaseSensor] = None, *args, **kwargs):
super().__init__(*args, **kwargs)
self._sensor = sensor
def on_connect(self, mqtt: MqttNode):
super().on_connect(mqtt)
mqtt.subscribe_module('temphum/data', self)
def tick(self):
pass
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
|
