summaryrefslogtreecommitdiff
path: root/src/home/mqtt
diff options
context:
space:
mode:
Diffstat (limited to 'src/home/mqtt')
-rw-r--r--src/home/mqtt/__init__.py4
-rw-r--r--src/home/mqtt/esp.py106
-rw-r--r--src/home/mqtt/mqtt.py76
-rw-r--r--src/home/mqtt/payload/__init__.py1
-rw-r--r--src/home/mqtt/payload/base_payload.py145
-rw-r--r--src/home/mqtt/payload/esp.py78
-rw-r--r--src/home/mqtt/payload/inverter.py73
-rw-r--r--src/home/mqtt/payload/relay.py22
-rw-r--r--src/home/mqtt/payload/sensors.py20
-rw-r--r--src/home/mqtt/payload/temphum.py15
-rw-r--r--src/home/mqtt/relay.py71
-rw-r--r--src/home/mqtt/temphum.py54
-rw-r--r--src/home/mqtt/util.py8
13 files changed, 0 insertions, 673 deletions
diff --git a/src/home/mqtt/__init__.py b/src/home/mqtt/__init__.py
deleted file mode 100644
index 982e2b6..0000000
--- a/src/home/mqtt/__init__.py
+++ /dev/null
@@ -1,4 +0,0 @@
-from .mqtt import MqttBase
-from .util import poll_tick
-from .relay import MqttRelay, MqttRelayState
-from .temphum import MqttTempHum \ No newline at end of file
diff --git a/src/home/mqtt/esp.py b/src/home/mqtt/esp.py
deleted file mode 100644
index 56ced83..0000000
--- a/src/home/mqtt/esp.py
+++ /dev/null
@@ -1,106 +0,0 @@
-import re
-import paho.mqtt.client as mqtt
-
-from .mqtt import MqttBase
-from typing import Optional, Union
-from .payload.esp import (
- OTAPayload,
- OTAResultPayload,
- DiagnosticsPayload,
- InitialDiagnosticsPayload
-)
-
-
-class MqttEspDevice:
- id: str
- secret: Optional[str]
-
- def __init__(self, id: str, secret: Optional[str] = None):
- self.id = id
- self.secret = secret
-
-
-class MqttEspBase(MqttBase):
- _devices: list[MqttEspDevice]
- _message_callback: Optional[callable]
- _ota_publish_callback: Optional[callable]
-
- TOPIC_LEAF = 'esp'
-
- def __init__(self,
- devices: Union[MqttEspDevice, list[MqttEspDevice]],
- subscribe_to_updates=True):
- super().__init__(clean_session=True)
- if not isinstance(devices, list):
- devices = [devices]
- self._devices = devices
- 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:
- for device in self._devices:
- topic = f'hk/{device.id}/{self.TOPIC_LEAF}/#'
- self._logger.debug(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(self.get_mqtt_topics(), msg.topic)
- self._logger.debug(f'topic: {msg.topic}')
- if not match:
- return
-
- device_id = match.group(1)
- subtopic = match.group(2)
-
- # try:
- next(d for d in self._devices if d.id == device_id)
- # except StopIteration:h
- # return
-
- message = None
- if subtopic == 'stat':
- message = DiagnosticsPayload.unpack(msg.payload)
- elif subtopic == 'stat1':
- message = InitialDiagnosticsPayload.unpack(msg.payload)
- elif subtopic == 'otares':
- message = OTAResultPayload.unpack(msg.payload)
-
- if message and self._message_callback:
- self._message_callback(device_id, message)
- return True
-
- except Exception as e:
- self._logger.exception(str(e))
-
- def push_ota(self,
- device_id,
- filename: str,
- publish_callback: callable,
- qos: int):
- device = next(d for d in self._devices if d.id == device_id)
- assert device.secret is not None, 'device secret not specified'
-
- self._ota_publish_callback = publish_callback
- payload = OTAPayload(secret=device.secret, filename=filename)
- publish_result = self._client.publish(f'hk/{device.id}/{self.TOPIC_LEAF}/admin/ota',
- payload=payload.pack(),
- qos=qos)
- self._ota_mid = publish_result.mid
- self._client.loop_write()
-
- @classmethod
- def get_mqtt_topics(cls, additional_topics: Optional[list[str]] = None):
- return rf'^hk/(.*?)/{cls.TOPIC_LEAF}/(stat|stat1|otares'+('|'+('|'.join(additional_topics)) if additional_topics else '')+')$' \ No newline at end of file
diff --git a/src/home/mqtt/mqtt.py b/src/home/mqtt/mqtt.py
deleted file mode 100644
index 4acd4f6..0000000
--- a/src/home/mqtt/mqtt.py
+++ /dev/null
@@ -1,76 +0,0 @@
-import os.path
-import paho.mqtt.client as mqtt
-import ssl
-import logging
-
-from typing import Tuple
-from ..config import config
-
-
-def username_and_password() -> Tuple[str, str]:
- username = config['mqtt']['username'] if 'username' in config['mqtt'] else None
- password = config['mqtt']['password'] if 'password' in config['mqtt'] else None
- return username, password
-
-
-class MqttBase:
- def __init__(self, clean_session=True):
- self._client = mqtt.Client(client_id=config['mqtt']['client_id'],
- protocol=mqtt.MQTTv311,
- clean_session=clean_session)
- 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):
- ca_certs = os.path.realpath(os.path.join(
- os.path.dirname(os.path.realpath(__file__)),
- '..',
- '..',
- '..',
- 'assets',
- 'mqtt_ca.crt'
- ))
- self._client.tls_set(ca_certs=ca_certs, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2)
-
- def connect_and_loop(self, loop_forever=True):
- host = config['mqtt']['host']
- port = config['mqtt']['port']
-
- self._client.connect(host, port, 60)
- if loop_forever:
- 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))
-
- 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.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
deleted file mode 100644
index eee6709..0000000
--- a/src/home/mqtt/payload/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .base_payload import MqttPayload \ No newline at end of file
diff --git a/src/home/mqtt/payload/base_payload.py b/src/home/mqtt/payload/base_payload.py
deleted file mode 100644
index 1abd898..0000000
--- a/src/home/mqtt/payload/base_payload.py
+++ /dev/null
@@ -1,145 +0,0 @@
-import abc
-import struct
-import re
-
-from typing import Optional, Tuple
-
-
-def pldstr(self) -> str:
- attrs = []
- for field in self.__class__.__annotations__:
- if hasattr(self, field):
- attr = getattr(self, field)
- attrs.append(f'{field}={attr}')
- if attrs:
- attrs_s = ' '
- attrs_s += ', '.join(attrs)
- else:
- attrs_s = ''
- return f'<%s{attrs_s}>' % (self.__class__.__name__,)
-
-
-class MqttPayload(abc.ABC):
- FORMAT = ''
- PACKER = {}
- UNPACKER = {}
-
- def __init__(self, **kwargs):
- for field in self.__class__.__annotations__:
- setattr(self, field, kwargs[field])
-
- def pack(self):
- args = []
- bf_number = -1
- bf_arg = 0
- bf_progress = 0
-
- for field, field_type in self.__class__.__annotations__.items():
- bfp = _bit_field_params(field_type)
- if bfp:
- n, s, b = bfp
- if n != bf_number:
- if bf_number != -1:
- args.append(bf_arg)
- bf_number = n
- bf_progress = 0
- bf_arg = 0
- bf_arg |= (getattr(self, field) & (2 ** b - 1)) << bf_progress
- bf_progress += b
-
- else:
- if bf_number != -1:
- args.append(bf_arg)
- bf_number = -1
- bf_progress = 0
- bf_arg = 0
-
- args.append(self._pack_field(field))
-
- if bf_number != -1:
- args.append(bf_arg)
-
- return struct.pack(self.FORMAT, *args)
-
- @classmethod
- def unpack(cls, buf: bytes):
- data = struct.unpack(cls.FORMAT, buf)
- kwargs = {}
- i = 0
- bf_number = -1
- bf_progress = 0
-
- for field, field_type in cls.__annotations__.items():
- bfp = _bit_field_params(field_type)
- if bfp:
- n, s, b = bfp
- if n != bf_number:
- bf_number = n
- bf_progress = 0
- kwargs[field] = (data[i] >> bf_progress) & (2 ** b - 1)
- bf_progress += b
- continue # don't increment i
-
- if bf_number != -1:
- bf_number = -1
- i += 1
-
- if issubclass(field_type, MqttPayloadCustomField):
- kwargs[field] = field_type.unpack(data[i])
- else:
- kwargs[field] = cls._unpack_field(field, data[i])
- i += 1
-
- return cls(**kwargs)
-
- def _pack_field(self, name):
- val = getattr(self, name)
- if self.PACKER and name in self.PACKER:
- return self.PACKER[name](val)
- else:
- return val
-
- @classmethod
- def _unpack_field(cls, name, val):
- if isinstance(val, MqttPayloadCustomField):
- return
- if cls.UNPACKER and name in cls.UNPACKER:
- return cls.UNPACKER[name](val)
- else:
- return val
-
- def __str__(self):
- return pldstr(self)
-
-
-class MqttPayloadCustomField(abc.ABC):
- def __init__(self, **kwargs):
- for field in self.__class__.__annotations__:
- setattr(self, field, kwargs[field])
-
- @abc.abstractmethod
- def __index__(self):
- pass
-
- @classmethod
- @abc.abstractmethod
- def unpack(cls, *args, **kwargs):
- pass
-
- def __str__(self):
- return pldstr(self)
-
-
-def bit_field(seq_no: int, total_bits: int, bits: int):
- return type(f'MQTTPayloadBitField_{seq_no}_{total_bits}_{bits}', (object,), {
- 'seq_no': seq_no,
- 'total_bits': total_bits,
- 'bits': bits
- })
-
-
-def _bit_field_params(cl) -> Optional[Tuple[int, ...]]:
- match = re.match(r'MQTTPayloadBitField_(\d+)_(\d+)_(\d)$', cl.__name__)
- if match is not None:
- return tuple([int(match.group(i)) for i in range(1, 4)])
- return None
diff --git a/src/home/mqtt/payload/esp.py b/src/home/mqtt/payload/esp.py
deleted file mode 100644
index 171cdb9..0000000
--- a/src/home/mqtt/payload/esp.py
+++ /dev/null
@@ -1,78 +0,0 @@
-import hashlib
-
-from .base_payload import MqttPayload, MqttPayloadCustomField
-
-
-class OTAResultPayload(MqttPayload):
- FORMAT = '=BB'
- result: int
- error_code: int
-
-
-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)
-
-
-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
diff --git a/src/home/mqtt/payload/inverter.py b/src/home/mqtt/payload/inverter.py
deleted file mode 100644
index 09388df..0000000
--- a/src/home/mqtt/payload/inverter.py
+++ /dev/null
@@ -1,73 +0,0 @@
-import struct
-
-from .base_payload import MqttPayload, bit_field
-from typing import Tuple
-
-_mult_10 = lambda n: int(n*10)
-_div_10 = lambda n: n/10
-
-
-class Status(MqttPayload):
- # 46 bytes
- FORMAT = 'IHHHHHHBHHHHHBHHHHHHHH'
-
- PACKER = {
- 'grid_voltage': _mult_10,
- 'grid_freq': _mult_10,
- 'ac_output_voltage': _mult_10,
- 'ac_output_freq': _mult_10,
- 'battery_voltage': _mult_10,
- 'battery_voltage_scc': _mult_10,
- 'battery_voltage_scc2': _mult_10,
- 'pv1_input_voltage': _mult_10,
- 'pv2_input_voltage': _mult_10
- }
- UNPACKER = {
- 'grid_voltage': _div_10,
- 'grid_freq': _div_10,
- 'ac_output_voltage': _div_10,
- 'ac_output_freq': _div_10,
- 'battery_voltage': _div_10,
- 'battery_voltage_scc': _div_10,
- 'battery_voltage_scc2': _div_10,
- 'pv1_input_voltage': _div_10,
- 'pv2_input_voltage': _div_10
- }
-
- time: int
- grid_voltage: float
- grid_freq: float
- ac_output_voltage: float
- ac_output_freq: float
- ac_output_apparent_power: int
- ac_output_active_power: int
- output_load_percent: int
- battery_voltage: float
- battery_voltage_scc: float
- battery_voltage_scc2: float
- battery_discharge_current: int
- battery_charge_current: int
- battery_capacity: int
- inverter_heat_sink_temp: int
- mppt1_charger_temp: int
- mppt2_charger_temp: int
- pv1_input_power: int
- pv2_input_power: int
- pv1_input_voltage: float
- pv2_input_voltage: float
-
- # H
- mppt1_charger_status: bit_field(0, 16, 2)
- mppt2_charger_status: bit_field(0, 16, 2)
- battery_power_direction: bit_field(0, 16, 2)
- dc_ac_power_direction: bit_field(0, 16, 2)
- line_power_direction: bit_field(0, 16, 2)
- load_connected: bit_field(0, 16, 1)
-
-
-class Generation(MqttPayload):
- # 8 bytes
- FORMAT = 'II'
-
- time: int
- wh: int
diff --git a/src/home/mqtt/payload/relay.py b/src/home/mqtt/payload/relay.py
deleted file mode 100644
index 4902991..0000000
--- a/src/home/mqtt/payload/relay.py
+++ /dev/null
@@ -1,22 +0,0 @@
-from .base_payload import MqttPayload
-from .esp import (
- OTAResultPayload,
- OTAPayload,
- InitialDiagnosticsPayload,
- DiagnosticsPayload
-)
-
-
-class PowerPayload(MqttPayload):
- FORMAT = '=12sB'
- PACKER = {
- 'state': lambda n: int(n),
- 'secret': lambda s: s.encode('utf-8')
- }
- UNPACKER = {
- 'state': lambda n: bool(n),
- 'secret': lambda s: s.decode('utf-8')
- }
-
- secret: str
- state: bool
diff --git a/src/home/mqtt/payload/sensors.py b/src/home/mqtt/payload/sensors.py
deleted file mode 100644
index f99b307..0000000
--- a/src/home/mqtt/payload/sensors.py
+++ /dev/null
@@ -1,20 +0,0 @@
-from .base_payload import MqttPayload
-
-_mult_100 = lambda n: int(n*100)
-_div_100 = lambda n: n/100
-
-
-class Temperature(MqttPayload):
- FORMAT = 'IhH'
- PACKER = {
- 'temp': _mult_100,
- 'rh': _mult_100,
- }
- UNPACKER = {
- 'temp': _div_100,
- 'rh': _div_100,
- }
-
- time: int
- temp: float
- rh: float
diff --git a/src/home/mqtt/payload/temphum.py b/src/home/mqtt/payload/temphum.py
deleted file mode 100644
index c0b744e..0000000
--- a/src/home/mqtt/payload/temphum.py
+++ /dev/null
@@ -1,15 +0,0 @@
-from .base_payload import MqttPayload
-
-two_digits_precision = lambda x: round(x, 2)
-
-
-class TempHumDataPayload(MqttPayload):
- FORMAT = '=ddb'
- UNPACKER = {
- 'temp': two_digits_precision,
- 'rh': two_digits_precision
- }
-
- temp: float
- rh: float
- error: int
diff --git a/src/home/mqtt/relay.py b/src/home/mqtt/relay.py
deleted file mode 100644
index a90f19c..0000000
--- a/src/home/mqtt/relay.py
+++ /dev/null
@@ -1,71 +0,0 @@
-import paho.mqtt.client as mqtt
-import re
-import datetime
-
-from .payload.relay import (
- PowerPayload,
-)
-from .esp import MqttEspBase
-
-
-class MqttRelay(MqttEspBase):
- TOPIC_LEAF = 'relay'
-
- def set_power(self, device_id, enable: bool, secret=None):
- device = next(d for d in self._devices if d.id == device_id)
- secret = secret if secret else device.secret
-
- assert secret is not None, 'device secret not specified'
-
- payload = PowerPayload(secret=secret,
- state=enable)
- self._client.publish(f'hk/{device.id}/{self.TOPIC_LEAF}/power',
- payload=payload.pack(),
- qos=1)
- self._client.loop_write()
-
- def on_message(self, client: mqtt.Client, userdata, msg):
- if super().on_message(client, userdata, msg):
- return
-
- try:
- match = re.match(self.get_mqtt_topics(['power']), msg.topic)
- if not match:
- return
-
- device_id = match.group(1)
- subtopic = match.group(2)
-
- message = None
- if subtopic == 'power':
- message = PowerPayload.unpack(msg.payload)
-
- if message and self._message_callback:
- self._message_callback(device_id, message)
-
- except Exception as e:
- self._logger.exception(str(e))
-
-
-class MqttRelayState:
- enabled: bool
- update_time: datetime.datetime
- rssi: int
- fw_version: int
- ever_updated: bool
-
- def __init__(self):
- self.ever_updated = False
- self.enabled = False
- self.rssi = 0
-
- def update(self,
- enabled: bool,
- rssi: int,
- fw_version=None):
- self.ever_updated = True
- self.enabled = enabled
- self.rssi = rssi
- self.update_time = datetime.datetime.now()
- if fw_version:
- self.fw_version = fw_version
diff --git a/src/home/mqtt/temphum.py b/src/home/mqtt/temphum.py
deleted file mode 100644
index 44810ef..0000000
--- a/src/home/mqtt/temphum.py
+++ /dev/null
@@ -1,54 +0,0 @@
-import paho.mqtt.client as mqtt
-import re
-
-from enum import auto
-from .payload.temphum import TempHumDataPayload
-from .esp import MqttEspBase
-from ..util import HashableEnum
-
-
-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 MqttTempHum(MqttEspBase):
- TOPIC_LEAF = 'temphum'
-
- def on_message(self, client: mqtt.Client, userdata, msg):
- if super().on_message(client, userdata, msg):
- return
-
- try:
- match = re.match(self.get_mqtt_topics(['data']), msg.topic)
- if not match:
- return
-
- device_id = match.group(1)
- subtopic = match.group(2)
-
- message = None
- if subtopic == 'data':
- message = TempHumDataPayload.unpack(msg.payload)
-
- if message and self._message_callback:
- self._message_callback(device_id, message)
-
- except Exception as e:
- self._logger.exception(str(e))
diff --git a/src/home/mqtt/util.py b/src/home/mqtt/util.py
deleted file mode 100644
index f71ffd8..0000000
--- a/src/home/mqtt/util.py
+++ /dev/null
@@ -1,8 +0,0 @@
-import time
-
-
-def poll_tick(freq):
- t = time.time()
- while True:
- t += freq
- yield max(t - time.time(), 0)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-21 14:56:20 +0000