hubee/controller.py at main · esimioni/hubee · GitHub

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
import gc
import time

from umachine import WDT

import debug_utils
import hubee
from device import Device
from sensor.base import Sensor
from xbee_device import XBeeDevice
from zigbee import Zigbee


class Controller:

    def __init__(self, sensors: list[Sensor]):
        self.wdt = WDT(timeout = (150000 if hubee.IS_PROD else 9999999))
        self.zigbee = Zigbee()
        self.xbee_device = XBeeDevice()
        self.sensors = sensors
        self.devices = {}
        self._setup_devices()

    def _setup_devices(self):
        self.sensors.insert(0, self.xbee_device)
        for sensor in self.sensors:
            if isinstance(sensor, Device):
                sensor.start(self.zigbee)
                self.devices[sensor.get_endpoint()] = sensor

    def _check_sensors_and_devices(self):
        time_now = time.ticks_ms()
        # print('Check Sensors')
        for sensor in self.sensors:
            try:
                sensor.check_updates(time_now)
                sensor.reset_error()
                # print(sensor.__class__.__name__ + ': Error reset')
            except OSError as e:
                # print(sensor.__class__.__name__ + ': Exception raised: ' + str(e))
                if hubee.IS_PROD:
                    if sensor.report_error(time_now):
                        sensor.transmit_error(sensor.__class__.__name__ + ': Persistent error updating status')
                        sensor.reset_error()
                        # print(sensor.__class__.__name__ + ': Notifying error on sensor')
                        debug_utils.notify_exception(e, self.zigbee)
                else:
                    raise

    def _check_zigbee_cmd(self):
        zgb_msg = self.zigbee.receive()
        if zgb_msg:
            endp = zgb_msg['endpoint']
            cmd = zgb_msg['command']
            payld = zgb_msg['payload']
            cluster = zgb_msg['cluster']
            device = self.devices.get(endp)
            if device:
                device.handle_command(cmd, payld)
            else:
                self.xbee_device.handle_unknown_command(cmd, endp, cluster, payld)

    def run(self):
        try:
            while True:
                gc.collect()
                self.wdt.feed()
                self._check_sensors_and_devices()
                self._check_zigbee_cmd()
                self.xbee_device.ble_disable()
        except BaseException as e:
            if hubee.IS_PROD:
                debug_utils.notify_exception(e, self.zigbee)
            else:
                raise