summaryrefslogtreecommitdiff
path: root/src/home/mqtt/module/diagnostics.py
blob: c31cce2d3c1fb7b47f0e9d65bbf9ee92baee09cb (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
from ..mqtt import MqttPayload, MqttPayloadCustomField
from .._node import MqttNode, MqttModule
from typing import Optional

MODULE_NAME = 'MqttDiagnosticsModule'


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


class MqttDiagnosticsModule(MqttModule):
    def init(self, mqtt: MqttNode):
        for topic in ('diag', 'd1ag', 'stat', 'stat1'):
            mqtt.subscribe_module(topic, self)

    def handle_payload(self, mqtt: MqttNode, topic: str, payload: bytes) -> Optional[MqttPayload]:
        if topic in ('stat', 'diag'):
            message = DiagnosticsPayload.unpack(payload)
        elif topic in ('stat1', 'd1ag'):
            message = InitialDiagnosticsPayload.unpack(payload)
        self._logger.debug(message)
        return message