diff options
Diffstat (limited to 'src/home/mqtt')
| -rw-r--r-- | src/home/mqtt/__init__.py | 1 | ||||
| -rw-r--r-- | src/home/mqtt/mqtt.py | 21 | ||||
| -rw-r--r-- | src/home/mqtt/payload/__init__.py | 1 | ||||
| -rw-r--r-- | src/home/mqtt/payload/relay.py | 45 | ||||
| -rw-r--r-- | src/home/mqtt/relay.py | 93 |
5 files changed, 153 insertions, 8 deletions
diff --git a/src/home/mqtt/__init__.py b/src/home/mqtt/__init__.py index c0ef9ba..a6f5f5e 100644 --- a/src/home/mqtt/__init__.py +++ b/src/home/mqtt/__init__.py @@ -1,2 +1,3 @@ from .mqtt import MQTTBase from .util import poll_tick +from .relay import MQTTRelay
\ No newline at end of file diff --git a/src/home/mqtt/mqtt.py b/src/home/mqtt/mqtt.py index b3334b5..9dd973b 100644 --- a/src/home/mqtt/mqtt.py +++ b/src/home/mqtt/mqtt.py @@ -6,8 +6,6 @@ import logging from typing import Tuple from ..config import config -logger = logging.getLogger(__name__) - def username_and_password() -> Tuple[str, str]: username = config['mqtt']['username'] if 'username' in config['mqtt'] else None @@ -23,11 +21,15 @@ class MQTTBase: self._client.on_connect = self.on_connect self._client.on_disconnect = self.on_disconnect self._client.on_message = self.on_message + self._client.on_log = self.on_log + self._client.on_publish = self.on_publish + self._loop_started = False self._logger = logging.getLogger(self.__class__.__name__) username, password = username_and_password() if username and password: + self._logger.debug(f'username={username} password={password}') self._client.username_pw_set(username, password) def configure_tls(self): @@ -50,6 +52,12 @@ class MQTTBase: self._client.loop_forever() else: self._client.loop_start() + self._loop_started = True + + def disconnect(self): + self._client.disconnect() + self._client.loop_write() + self._client.loop_stop() def on_connect(self, client: mqtt.Client, userdata, flags, rc): self._logger.info("Connected with result code " + str(rc)) @@ -57,5 +65,12 @@ class MQTTBase: def on_disconnect(self, client: mqtt.Client, userdata, rc): self._logger.info("Disconnected with result code " + str(rc)) + def on_log(self, client: mqtt.Client, userdata, level, buf): + level = mqtt.LOGGING_LEVEL[level] if level in mqtt.LOGGING_LEVEL else logging.INFO + self._logger.log(level, f'MQTT: {buf}') + def on_message(self, client: mqtt.Client, userdata, msg): - self._logger.info(msg.topic + ": " + str(msg.payload)) + self._logger.debug(msg.topic + ": " + str(msg.payload)) + + def on_publish(self, client: mqtt.Client, userdata, mid): + self._logger.debug(f'publish done, mid={mid}')
\ No newline at end of file diff --git a/src/home/mqtt/payload/__init__.py b/src/home/mqtt/payload/__init__.py index e69de29..9fcaf3e 100644 --- a/src/home/mqtt/payload/__init__.py +++ b/src/home/mqtt/payload/__init__.py @@ -0,0 +1 @@ +from .base_payload import MQTTPayload
\ No newline at end of file diff --git a/src/home/mqtt/payload/relay.py b/src/home/mqtt/payload/relay.py index 2a327ba..debc2c8 100644 --- a/src/home/mqtt/payload/relay.py +++ b/src/home/mqtt/payload/relay.py @@ -1,6 +1,10 @@ +import hashlib + from .base_payload import MQTTPayload, MQTTPayloadCustomField +# _logger = logging.getLogger(__name__) + class StatFlags(MQTTPayloadCustomField): state: bool config_changed_value_present: bool @@ -8,9 +12,11 @@ class StatFlags(MQTTPayloadCustomField): @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 StatFlags(state=(state == 1), config_changed_value_present=(ccvp == 1), config_changed=(cc == 1)) @@ -24,7 +30,7 @@ class StatFlags(MQTTPayloadCustomField): class InitialStatPayload(MQTTPayload): - FORMAT = 'IBbIB' + FORMAT = '=IBbIB' ip: int fw_version: int @@ -34,7 +40,7 @@ class InitialStatPayload(MQTTPayload): class StatPayload(MQTTPayload): - FORMAT = 'bIB' + FORMAT = '=bIB' rssi: int free_heap: int @@ -42,13 +48,42 @@ class StatPayload(MQTTPayload): class PowerPayload(MQTTPayload): - FORMAT = '12sB' + FORMAT = '=12sB' PACKER = { - 'state': lambda n: int(n) + 'state': lambda n: int(n), + 'secret': lambda s: s.encode('utf-8') } UNPACKER = { - 'state': lambda n: bool(n) + 'state': lambda n: bool(n), + 'secret': lambda s: s.decode('utf-8') } secret: str state: bool + + +class OTAPayload(MQTTPayload): + secret: str + filename: str + + # structure of returned data: + # + # uint8_t[len(secret)] secret; + # uint8_t[16] md5; + # *uint8_t data + + def pack(self): + buf = bytearray(self.secret.encode()) + m = hashlib.md5() + with open(self.filename, 'rb') as fd: + content = fd.read() + m.update(content) + buf.extend(m.digest()) + buf.extend(content) + return buf + + def unpack(cls, buf: bytes): + raise RuntimeError(f'{cls.__class__.__name__}.unpack: not implemented') + # secret = buf[:12].decode() + # filename = buf[12:].decode() + # return OTAPayload(secret=secret, filename=filename) diff --git a/src/home/mqtt/relay.py b/src/home/mqtt/relay.py new file mode 100644 index 0000000..0f97b5b --- /dev/null +++ b/src/home/mqtt/relay.py @@ -0,0 +1,93 @@ +import paho.mqtt.client as mqtt +import re + +from .mqtt import MQTTBase +from typing import Optional, Union +from .payload.relay import ( + InitialStatPayload, + StatPayload, + PowerPayload, + OTAPayload +) + + +class MQTTRelay(MQTTBase): + _home_id: Union[str, int] + _secret: str + _message_callback: Optional[callable] + _ota_publish_callback: Optional[callable] + + def __init__(self, + home_id: Union[str, int], + secret: str, + subscribe_to_updates=True): + super().__init__(clean_session=True) + self._home_id = home_id + self._secret = secret + self._message_callback = None + self._ota_publish_callback = None + self._subscribe_to_updates = subscribe_to_updates + self._ota_mid = None + + def on_connect(self, client: mqtt.Client, userdata, flags, rc): + super().on_connect(client, userdata, flags, rc) + + if self._subscribe_to_updates: + topic = f'hk/{self._home_id}/relay/#' + self._logger.info(f"subscribing to {topic}") + + client.subscribe(topic, qos=1) + + def on_publish(self, client: mqtt.Client, userdata, mid): + if self._ota_mid is not None and mid == self._ota_mid and self._ota_publish_callback: + self._ota_publish_callback() + + def set_message_callback(self, callback: callable): + self._message_callback = callback + + def on_message(self, client: mqtt.Client, userdata, msg): + try: + match = re.match(r'^hk/(.*?)/relay/(stat|stat1|power|otares)$', msg.topic) + self._logger.debug(f'topic: {msg.topic}') + if not match: + return + + name = match.group(1) + subtopic = match.group(2) + + if name != self._home_id: + return + + message = None + if subtopic == 'stat': + message = StatPayload.unpack(msg.payload) + elif subtopic == 'stat1': + message = InitialStatPayload.unpack(msg.payload) + elif subtopic == 'power': + message = PowerPayload.unpack(msg.payload) + + if message and self._message_callback: + self._message_callback(message) + + except Exception as e: + self._logger.exception(str(e)) + + def set_power(self, enable: bool): + payload = PowerPayload(secret=self._secret, + state=enable) + self._client.publish(f'hk/{self._home_id}/relay/power', + payload=payload.pack(), + qos=1) + self._client.loop_write() + + def push_ota(self, + filename: str, + publish_callback: callable, + qos: int): + self._ota_publish_callback = publish_callback + payload = OTAPayload(secret=self._secret, filename=filename) + publish_result = self._client.publish(f'hk/{self._home_id}/relay/admin/ota', + payload=payload.pack(), + qos=qos) + self._ota_mid = publish_result.mid + self._client.loop_write() |