mqtt reporter

Author: pshirshov

pyproject.toml

@@ -35,6 +35,9 @@ dependencies = [
     "pymongo",
     "requests",
 
+    "paho-mqtt",
+    "ha-mqtt-discoverable",
+
 ]
 url = "http://github.com/Frankkkkk/python-pylontech"
 

src/pylontechpoller/hass_basic_reporter.py

@@ -0,0 +1,39 @@
+import json
+import os
+
+import requests
+
+from pylontechpoller.reporter import Reporter, logger
+
+
+class HassReporter(Reporter):
+    def __init__(self, hass_url, hass_stack_disbalance, hass_max_battery_disbalance, hass_max_battery_disbalance_id, hass_token):
+        self.hass_url = hass_url
+        self.hass_stack_disbalance = hass_stack_disbalance
+        self.hass_max_battery_disbalance = hass_max_battery_disbalance
+        self.hass_max_battery_disbalance_id = hass_max_battery_disbalance_id
+        if os.path.exists(hass_token):
+            with open(hass_token, 'r') as file:
+                hass_token = file.read().strip()
+        self.hass_token = hass_token
+
+
+    def report_state(self, state):
+        md = state["max_module_disbalance"]
+        self.update_hass_state(self.hass_stack_disbalance, int(state["stack_disbalance"] * 10000) / 10000.0)
+        self.update_hass_state(self.hass_max_battery_disbalance, int(md[1] * 10000) / 10000.0)
+        self.update_hass_state(self.hass_max_battery_disbalance_id, md[0])
+
+    def update_hass_state(self, id, value):
+        tpe = id.split('.')[0]
+        update = {
+            "entity_id": id,
+            "value": value
+        }
+
+        url = f'{self.hass_url}/api/services/{tpe}/set_value'
+
+        response = requests.post(url, data=json.dumps(update), headers={"Authorization": f"Bearer {self.hass_token}"})
+
+        if response.status_code != 200:
+            logger.error(f"hass state update failed for {id}: {response.status_code} {response.text}")

src/pylontechpoller/mongo_reporter.py

@@ -0,0 +1,27 @@
+import datetime
+
+from pymongo import MongoClient
+
+from pylontech import to_json_serializable, Pylontech
+from pylontech.pylontech import PylontechStackData
+from pylontechpoller.reporter import Reporter
+
+
+class MongoReporter(Reporter):
+    def __init__(self, mongo_url, mongo_db, mongo_collection_meta, mongo_collection_history, retention_days):
+        mongo = MongoClient(mongo_url)
+        db = mongo[mongo_db]
+        self.retention_days = retention_days
+        self.collection_meta = db[mongo_collection_meta]
+        self.collection_hist = db[mongo_collection_history]
+        self.collection_hist.create_index("ts", expireAfterSeconds=3600 * 24 * 90)
+
+    def report_meta(self, meta: PylontechStackData, p: Pylontech):
+        self.collection_meta.insert_one({'ts':  datetime.datetime.now().isoformat(), "stack": to_json_serializable(meta)})
+
+    def report_state(self, state):
+        self.collection_hist.insert_one(state)
+
+    def cleanup(self):
+        threshold = datetime.datetime.now() - datetime.timedelta(days= self.retention_days)
+        self.collection_hist.delete_many({"ts": {"$lt": threshold}})

src/pylontechpoller/mqtt_reporter.py

@@ -0,0 +1,169 @@
+import os.path
+
+from ha_mqtt_discoverable import Settings, DeviceInfo
+from ha_mqtt_discoverable.sensors import SensorInfo, Sensor
+
+from pylontech.pylontech import PylontechModule, Pylontech, PylontechStackData
+from pylontechpoller.tools import minimize
+from pylontechpoller.reporter import Reporter
+
+import paho.mqtt.client as mqtt
+
+
+class MqttReporter(Reporter):
+    def __init__(self, mqtt_host, mqtt_port, mqtt_login, mqtt_password):
+        if os.path.exists(mqtt_password):
+            with open(mqtt_password, 'r') as file:
+                mqtt_password = file.read().strip()
+
+        client = mqtt.Client(client_id="pylontech-poller")
+        client.username_pw_set(mqtt_login, mqtt_password)
+        client.connect(mqtt_host, mqtt_port)
+        client.loop_start()
+        self.mqtt_settings = Settings.MQTT(client=client)
+        # client.enable_logger(logger)
+
+        # self.mqtt_settings = Settings.MQTT(host=mqtt_host, port=mqtt_port, username=mqtt_login, password=mqtt_password,
+        #                                    client_name="pylontech-poller")
+
+        self.device_info = DeviceInfo(name="Pylontech Battery Stack", identifiers="pylontech_battery_stack")
+
+        self.hass_stack_disbalance_info = SensorInfo(
+            name="Stack Disbalance",
+            device_class="voltage",
+            unique_id="stack_disbalance",
+            unit_of_measurement="V",
+            suggested_display_precision=3,
+            device=self.device_info,
+            icon="mdi:scale-unbalanced",
+
+        )
+        self.hass_stack_disbalance_settings = Settings(mqtt=self.mqtt_settings, entity=self.hass_stack_disbalance_info)
+        self.hass_stack_disbalance = Sensor(self.hass_stack_disbalance_settings)
+
+        self.hass_max_battery_disbalance_info = SensorInfo(
+            name="Max Battery Disbalance",
+            device_class="voltage",
+            unique_id="max_battery_disbalance",
+            unit_of_measurement="V",
+            suggested_display_precision=3,
+            device=self.device_info,
+            icon="mdi:scale-unbalanced",
+        )
+        self.hass_max_battery_disbalance_settings = Settings(mqtt=self.mqtt_settings,
+                                                             entity=self.hass_max_battery_disbalance_info)
+        self.hass_max_battery_disbalance = Sensor(self.hass_max_battery_disbalance_settings)
+
+        self.hass_max_disbalance_id = Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+            name="Max disbalance ID",
+            unique_id=f"max_battery_disbalance_id",
+            device=self.device_info,
+            icon="mdi:battery-alert",
+
+        )))
+        self.bats = {}
+
+    def report_meta(self, meta: PylontechStackData, p: Pylontech):
+        moduledata = { m["n"] : m for m in minimize( next(p.poll_parameters(meta.range())) )["modules"]}
+        cells = {}
+
+        for id in meta.ids:
+            m = meta.modules[id]
+            device_info = DeviceInfo(
+                name=f"Pylontech Battery {id}",
+                identifiers=[f"pylontech_battery_{m.serial}", f"pylontech_battery_{id}", ],
+                manufacturer=m.manufacturer_info,
+                sw_version=".".join([str(x) for x in m.fw_version]),
+                model=m.device_name
+            )
+            mdata = moduledata[id]
+            for cn, c in enumerate(mdata["cv"]):
+                cells[f"cell_{cn}_voltage"] = Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                        name=f"Cell {cn} Voltage",
+                        device_class="voltage",
+                        unique_id=f"cell_voltage_{id}_{cn}",
+                        unit_of_measurement="V",
+                        suggested_display_precision=3,
+                        device=device_info,
+                        entity_category="diagnostic",
+                        icon="mdi:gauge",
+                    )))
+
+            self.bats[id] = {
+                "bat_soc": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="SoC",
+                    device_class="battery",
+                    unique_id=f"battery_soc_{id}",
+                    unit_of_measurement="%",
+                    suggested_display_precision=1,
+                    device=device_info
+                ))),
+                "bat_disbalance": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="Cell Disbalance",
+                    device_class="voltage",
+                    unique_id=f"battery_disbalance_{id}",
+                    unit_of_measurement="V",
+                    suggested_display_precision=3,
+                    device=device_info,
+                    icon="mdi:scale-unbalanced",
+                ))),
+                "bat_voltage": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="Voltage",
+                    device_class="voltage",
+                    unique_id=f"battery_voltage_{id}",
+                    unit_of_measurement="V",
+                    suggested_display_precision=3,
+                    device=device_info,
+                    icon="mdi:gauge",
+                ))),
+                "bat_current": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="Current",
+                    device_class="current",
+                    unique_id=f"battery_current_{id}",
+                    unit_of_measurement="A",
+                    suggested_display_precision=3,
+                    device=device_info,
+                    icon="mdi:current-dc",
+                ))),
+                "bat_power": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="Power",
+                    device_class="power",
+                    unique_id=f"battery_power_{id}",
+                    unit_of_measurement="W",
+                    suggested_display_precision=2,
+                    device=device_info,
+                    icon="mdi:battery-charging",
+                ))),
+                "bat_cycle": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="Cycle",
+                    unique_id=f"battery_cycle_{id}",
+                    device=device_info,
+                    icon="mdi:battery-sync",
+                ))),
+                "bat_temp": Sensor(Settings(mqtt=self.mqtt_settings, entity=SensorInfo(
+                    name="Temperature",
+                    device_class="temperature",
+                    unique_id=f"battery_temperature_{id}",
+                    unit_of_measurement="C",
+                    suggested_display_precision=1,
+                    device=device_info,
+                ))),
+            } | cells
+
+    def report_state(self, state):
+        md = state["max_module_disbalance"]
+        self.hass_stack_disbalance.set_state(state["stack_disbalance"])
+        self.hass_max_battery_disbalance.set_state(md[1])
+        self.hass_max_disbalance_id.set_state(md[0])
+
+        for b in state["modules"]:
+            s = self.bats[b["n"]]
+            s["bat_disbalance"].set_state(b["disbalance"])
+            s["bat_voltage"].set_state(b["v"])
+            s["bat_current"].set_state(b["current"])
+            s["bat_soc"].set_state(int(b["soc"] * 1000) / 10.0)
+            s["bat_power"].set_state(b["pw"])
+            s["bat_cycle"].set_state(b["cycle"])
+            s["bat_temp"].set_state(b["tempavg"])
+            for cn, c in enumerate(b["cv"]):
+                s[f"cell_{cn}_voltage"].set_state(c)

src/pylontechpoller/poller.py

@@ -1,70 +1,17 @@
 import argparse
-import json
 import logging
 import sys
 import time
 
 from pylontech import *
-from pylontechpoller.reporter import MongoReporter, HassReporter
+from pylontechpoller.mqtt_reporter import MqttReporter
+from pylontechpoller.hass_basic_reporter import HassReporter
+from pylontechpoller.mongo_reporter import MongoReporter
+from pylontechpoller.tools import minimize
 
 logger = logging.getLogger(__name__)
 
 
-def find_min_max_modules(modules):
-    all_voltages = []
-    all_disbalances = []
-
-    for module in modules:
-        mid = module["NumberOfModule"]
-        cvs = module["CellVoltages"]
-        for voltage in cvs:
-            all_voltages.append((mid, voltage))
-        vmax = max(cvs)
-        vmin = min(cvs)
-        d = vmax - vmin
-        all_disbalances.append((mid, d))
-
-    if not all_voltages:
-        return None, None
-
-    min_pair = min(all_voltages, key=lambda x: x[1])
-    max_pair = max(all_voltages, key=lambda x: x[1])
-    max_disbalance = max(all_disbalances, key=lambda x: abs(x[1]))
-
-    return min_pair, max_pair, max_disbalance
-
-
-
-def minimize(b: json) -> json:
-    def minimize_module(m: json) -> json:
-        return {
-            "n": m["NumberOfModule"],
-            "v": m["Voltage"],
-            "cv": m["CellVoltages"],
-            "current": m["Current"],
-            "pw": m["Power"],
-            "cycle": m["CycleNumber"],
-            "soc": m["StateOfCharge"],
-            "tempavg": m["AverageBMSTemperature"],
-            "temps": m["GroupedCellsTemperatures"],
-            "remaining": m["RemainingCapacity"],
-            "disbalance": max(m["CellVoltages"]) - min(m["CellVoltages"])
-        }
-
-    modules = b["modules"]
-    find_min_max_modules(modules)
-
-    (min_pair, max_pair, max_disbalance) = find_min_max_modules(modules)
-
-    return {
-        "ts": b["timestamp"],
-        "cvmin": min_pair,
-        "cvmax": max_pair,
-        "stack_disbalance": max_pair[1] - min_pair[1],
-        "max_module_disbalance": max_disbalance,
-        "modules": list(map(minimize_module, modules)),
-    }
-
 
 
 def run(argv: list[str]):
@@ -87,7 +34,14 @@ def run(argv: list[str]):
     parser.add_argument("--hass-stack-disbalance", type=str, help="state id", default="input_number.stack_disbalance")
     parser.add_argument("--hass-max-battery-disbalance", type=str, help="state id", default="input_number.max_bat_disbalance")
     parser.add_argument("--hass-max-battery-disbalance-id", type=str, help="state id", default="input_text.max_disbalance_id")
-    parser.add_argument("--hass-token-file", type=str, help="hass token file", default="/var/run/agenix/hass-token")
+    parser.add_argument("--hass-token", type=str, help="hass token or token file", default="/var/run/agenix/hass-token")
+
+
+    parser.add_argument("--mqtt-host", type=str, help="mqtt host", default=None)
+    parser.add_argument("--mqtt-port", type=int, help="mqtt url", default=1883)
+    parser.add_argument("--mqtt-user", type=str, help="mqtt login", default="mqtt")
+    parser.add_argument("--mqtt-password", type=str, help="mqtt password or password file", default="/var/run/agenix/mqtt-user")
+
 
 
     args = parser.parse_args(argv[1:])
@@ -128,13 +82,23 @@ def run(argv: list[str]):
                     args.hass_token_file
                 ))
 
+            mqtt_host = args.mqtt_host
+
+            if mqtt_host:
+                reporters.append(MqttReporter(
+                    mqtt_host,
+                    args.mqtt_port,
+                    args.mqtt_user,
+                    args.mqtt_password,
+                ))
+
             logging.info("About to start polling...")
             bats = p.scan_for_batteries(2, 10)
 
             logging.info("Have battery stack data")
 
             for reporter in reporters:
-                reporter.report_meta(bats)
+                reporter.report_meta(bats, p)
 
             for b in p.poll_parameters(bats.range()):
                 cc += 1