Hook MQTT up to the live data feed

.gitignore

@@ -1,5 +1,8 @@
 *.py[ocd]
 
+.mypy_cache/
+__pycache__/
+
 *.log
 *.zip
 

consumers/mqtt.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+import json
 from typing import Any, Dict, List, Optional
 from uuid import uuid4
 
@@ -34,11 +35,13 @@ MAP_VALUES: Dict[str, Dict[str, Any]] = {
         "unit": "Wh",
         "type": "energy",
         "state_class": "total_increasing",
+        "expiry": 180,
     },
     "total_consumption_power": {
         "unit": "Wh",
         "type": "energy",
         "state_class": "total_increasing",
+        "expiry": 180,
     },
     #
     "battery_charge": {"unit": "%", "type": "battery", "state_class": "measurement"},
@@ -128,6 +131,7 @@ class MqttConsumer(BaseConsumer):
                 # TODO: Get charger serial and use for identifier instead
                 # See: https://www.home-assistant.io/integrations/sensor.mqtt/#device
                 # "via_device": self.settings["device_id"],
+                "suggested_area": "Solar panel",
             },
             "force_update": True,
             "expire_after": expiry,
@@ -140,6 +144,8 @@ class MqttConsumer(BaseConsumer):
         if state_class:
             res["state_class"] = state_class
 
+        return res
+
     # The callback for when the client receives a CONNACK response from the server.
     @staticmethod
     def on_connect(client: mqtt.Client, userdata: "MqttConsumer", flags, rc):
@@ -162,15 +168,17 @@ class MqttConsumer(BaseConsumer):
         return super().poll()
 
     def write(self, data: Dict[str, Any]):
-        self.client.publish(f"{self.topic_prefix}/raw", payload=data)
+        self.client.publish(f"{self.topic_prefix}/raw", payload=json.dumps(data))
 
         for k, v in data.items():
             if k in MAP_VALUES:
                 if k not in self.initialized:
                     pretty_name = k.replace("_", " ").capitalize()
                     self.client.publish(
-                        f"{self.settings['discovery_prefix']}/sensor/{self.settings['device_id']}/config",  # noqa: E501
+                        f"{self.settings['discovery_prefix']}/sensor/{self.settings['device_id']}_{k}/config",  # noqa: E501
-                        self.get_ha_config(k, pretty_name, **MAP_VALUES[k]),
+                        payload=json.dumps(
+                            self.get_ha_config(k, pretty_name, **MAP_VALUES[k])
+                        ),
                         retain=True,
                     )
                     self.initialized.append(k)

solar_ble.py

@@ -12,6 +12,7 @@ from bluepy import btle
 from libscrc import modbus
 
 from feasycom_ble import BTLEUart
+from test_config import get_config, get_consumers
 
 MAC = "DC:0D:30:9C:61:BA"
 
@@ -366,68 +367,92 @@ class Periodical:
 
 
 if __name__ == "__main__":
+    conf = get_config()
+    consumers = get_consumers(conf)
 
     per_voltages = Periodical(interval=15)
     per_current_hist = Periodical(interval=60)
 
-    while True:
+    try:
-        try:
+        while True:
-            log("Connecting...")
+            try:
-            with BTLEUart(MAC, timeout=10) as dev:
+                log("Connecting...")
-                log("Connected.")
+                with BTLEUart(MAC, timeout=10) as dev:
+                    log("Connected.")
 
-                # write(dev, construct_request(0, 32))
+                    # write(dev, construct_request(0, 32))
 
-                # Memory dump
+                    # Memory dump
-                # for address in range(0, 0x10000, 16):
+                    # for address in range(0, 0x10000, 16):
-                #    log(f"Reading 0x{address:04X}...")
+                    #    log(f"Reading 0x{address:04X}...")
-                #    write(wd, construct_request(address, 16))
+                    #    write(wd, construct_request(address, 16))
-                days = 7
+                    days = 7
-                res = readMemory(dev, 0x010B, 21)
+                    res = readMemory(dev, 0x010B, 21)
-                if res:
-                    d = parse_historical_entry(res)
-                    log(d)
-                    days = cast(int, d.get("run_days", 7))
-
-                for i in range(days):
-                    res = readMemory(dev, 0xF000 + i, 10)
                     if res:
                         d = parse_historical_entry(res)
-                        log({i: d})
+                        log(d)
+                        for consumer in consumers:
+                            consumer.write(d)
+                        days = cast(int, d.get("run_days", 7))
 
-                while True:
+                    for i in range(days):
-                    now = time.time()
+                        res = readMemory(dev, 0xF000 + i, 10)
-                    if per_voltages(now):
-                        # CMD_GET_BATTERY_STATE + CMD_GET_PANEL_STATUS
-                        res = readMemory(dev, 0x0100, 11)
                         if res:
-                            try:
+                            d = parse_historical_entry(res)
-                                d = parse_battery_state(res)
+                            log({i: d})
-                                log(d)
+                            for consumer in consumers:
-                            except struct.error as e:
+                                consumer.write({str(i): d})
-                                log(e)
-                                log("0x0100 Unpack error:", len(res), res)
-                                log("Flushed from read buffer; ", dev.read(timeout=0.5))
-                    if per_current_hist(now):
-                        res = readMemory(dev, 0x010B, 21)
-                        if res:
-                            try:
-                                d = parse_historical_entry(res)
-                                log(d)
-                            except struct.error as e:
-                                log(e)
-                                log("0x010B Unpack error:", len(res), res)
-                                log("Flushed from read buffer; ", dev.read(timeout=0.5))
-                    # print(".")
-                    time.sleep(1)
 
-                # if STATUS.get('load_enabled'):
+                    while True:
-                #    write(wd, CMD_DISABLE_LOAD)
+                        now = time.time()
-                # else:
+                        if per_voltages(now):
-                #    write(wd, CMD_ENABLE_LOAD)
+                            # CMD_GET_BATTERY_STATE + CMD_GET_PANEL_STATUS
+                            res = readMemory(dev, 0x0100, 11)
+                            if res:
+                                try:
+                                    d = parse_battery_state(res)
+                                    log(d)
+                                    for consumer in consumers:
+                                        consumer.write(d)
+                                except struct.error as e:
+                                    log(e)
+                                    log("0x0100 Unpack error:", len(res), res)
+                                    log(
+                                        "Flushed from read buffer; ",
+                                        dev.read(timeout=0.5),
+                                    )
+                        if per_current_hist(now):
+                            res = readMemory(dev, 0x010B, 21)
+                            if res:
+                                try:
+                                    d = parse_historical_entry(res)
+                                    log(d)
+                                    for consumer in consumers:
+                                        consumer.write(d)
+                                except struct.error as e:
+                                    log(e)
+                                    log("0x010B Unpack error:", len(res), res)
+                                    log(
+                                        "Flushed from read buffer; ",
+                                        dev.read(timeout=0.5),
+                                    )
+                        # print(".")
+                        for consumer in consumers:
+                            consumer.poll()
+                        time.sleep(max(0, 1 - time.time() - now))
 
-        except btle.BTLEDisconnectError:
+                    # if STATUS.get('load_enabled'):
-            log("ERROR: Disconnected")
+                    #    write(wd, CMD_DISABLE_LOAD)
-            time.sleep(1)
+                    # else:
-        except KeyboardInterrupt:
+                    #    write(wd, CMD_ENABLE_LOAD)
-            break
+
+            except btle.BTLEDisconnectError:
+                log("ERROR: Disconnected")
+                time.sleep(1)
+
+    except (KeyboardInterrupt, SystemExit, Exception) as e:
+        for consumer in consumers:
+            consumer.exit()
+
+        if type(e) is not KeyboardInterrupt:
+            raise

test_config.py

@@ -2,7 +2,7 @@
 import importlib
 import os
 from time import sleep
-from typing import Any, Dict, Optional, Type
+from typing import Any, Dict, List, Optional, Type
 
 import yaml
 
@@ -14,8 +14,8 @@ def get_consumer(name: str) -> Optional[Type[BaseConsumer]]:
 
     mod = importlib.import_module(f"consumers.{mod_name}")
 
-    print(mod)
+    # print(mod)
-    print(dir(mod))
+    # print(dir(mod))
     res = getattr(mod, cls_name)
     assert issubclass(res, BaseConsumer)
 
@@ -36,29 +36,37 @@ def write_config(conf: Dict[str, Any]):
     os.rename(".config.yaml~writing", "config.yaml")
 
 
-conf = get_config()
+def get_consumers(conf: Optional[Dict[str, Any]] = None) -> List[BaseConsumer]:
+    if conf is None:
+        conf = get_config()
 
-consumers = []
+    consumers = []
-for name, consumer_config in conf["consumers"].items():
+    for name, consumer_config in conf["consumers"].items():
-    print(name, consumer_config)
+        # print(name, consumer_config)
-    mod = get_consumer(name)
+        mod = get_consumer(name)
-    if mod:
+        if mod:
-        print(mod)
+            # print(mod)
-        consumers.append(mod(consumer_config))
+            consumers.append(mod(consumer_config))
-
-write_config(conf)
-
-
-try:
-    while True:
-        for consumer in consumers:
-            consumer.poll()
-            sleep(1)
-except (KeyboardInterrupt, SystemExit, Exception) as e:
-    for consumer in consumers:
-        consumer.exit()
-
-    if type(e) is not KeyboardInterrupt:
-        raise
 
     write_config(conf)
+    return consumers
+
+
+if __name__ == "__main__":
+    conf = get_config()
+
+    consumers = get_consumers(conf)
+
+    try:
+        while True:
+            for consumer in consumers:
+                consumer.poll()
+                sleep(1)
+    except (KeyboardInterrupt, SystemExit, Exception) as e:
+        for consumer in consumers:
+            consumer.exit()
+
+        if type(e) is not KeyboardInterrupt:
+            raise
+
+        write_config(conf)