poller: home assistant pushes

Author: pshirshov

pyproject.toml

@@ -27,10 +27,14 @@ classifiers = [
 dependencies = [
     "pyserial",
     "construct",
+
     "standard-telnetlib",
     "Exscript",
+
     "rich",
     "pymongo",
+    "requests",
+
 ]
 url = "http://github.com/Frankkkkk/python-pylontech"
 

src/pylontechpoller/poller.py

@@ -1,31 +1,37 @@
 import argparse
-import datetime
 import json
-import time
-import sys
 import logging
-import itertools
+import sys
+import time
 
-from rich import print_json
 from pylontech import *
-from pymongo import MongoClient
+from pylontechpoller.reporter import MongoReporter, HassReporter
 
 logger = logging.getLogger(__name__)
 
 
 def find_min_max_modules(modules):
     all_voltages = []
+    all_disbalances = []
+
     for module in modules:
-        for voltage in module["CellVoltages"]:
-            all_voltages.append((module["NumberOfModule"], voltage))
+        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
+    return min_pair, max_pair, max_disbalance
 
 
 
@@ -48,22 +54,18 @@ def minimize_module(m: json) -> json:
     modules = b["modules"]
     find_min_max_modules(modules)
 
-    (min_pair, max_pair) = find_min_max_modules(modules)
-    # allcv = list(itertools.chain.from_iterable(map(lambda m: m["CellVoltages"], modules)))
-    # vmin = min(allcv)
-    # vmax = max(allcv)
+    (min_pair, max_pair, max_disbalance) = find_min_max_modules(modules)
 
     return {
         "ts": b["timestamp"],
         "cvmin": min_pair,
         "cvmax": max_pair,
         "stack_disbalance": min_pair[1] - max_pair[1],
+        "max_module_disbalance": max_disbalance,
         "modules": list(map(minimize_module, modules)),
     }
 
-def mongo_cleanup(collection, retention_days):
-    threshold = datetime.datetime.now() - datetime.timedelta(days=retention_days)
-    collection.delete_many({"ts": {"$lt": threshold}})
+
 
 def run(argv: list[str]):
     parser = argparse.ArgumentParser(description="Pylontech RS485 poller")
@@ -75,11 +77,19 @@ def run(argv: list[str]):
     parser.add_argument("--interval", type=int, help="polling interval in msec", default=1000)
     parser.add_argument("--retention-days", type=int, help="how long to retain history data", default=90)
     parser.add_argument("--debug", type=bool, help="verbose output", default=False)
-    parser.add_argument("--mongo-url", type=str, help="mongodb url", default=False)
+
+    parser.add_argument("--mongo-url", type=str, help="mongodb url", default=None)
     parser.add_argument("--mongo-db", type=str, help="target mongo database", default="pylontech")
     parser.add_argument("--mongo-collection-history", type=str, help="target mongo collection_hist for stack history", default="history")
     parser.add_argument("--mongo-collection-meta", type=str, help="target mongo collection_hist for stack data", default="meta")
 
+    parser.add_argument("--hass-url", type=str, help="hass url", default=None)
+    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")
+
+
     args = parser.parse_args(argv[1:])
 
     level = logging.DEBUG if args.debug else logging.INFO
@@ -88,25 +98,42 @@ def run(argv: list[str]):
     cc = 0
     spinner = ['|', '/', '-', '\\']
 
+    reporters = []
+
     while True:
         try:
             logging.debug("Preparing client...")
             p = Pylontech(ExscriptTelnetTransport(host=args.source_host, port=args.source_port, timeout=args.timeout))
-            
-            mongo = MongoClient(args.mongo_url)
-            db = mongo[args.mongo_db]
-
-            collection_meta = db[args.mongo_collection_meta]
 
-            collection_hist = db[args.mongo_collection_history]
-            collection_hist.create_index("ts", expireAfterSeconds=3600*24*90)
+            mongo_url = args.mongo_url
+
+            if mongo_url:
+                reporters.append(MongoReporter(
+                    mongo_url,
+                    args.mongo_db,
+                    args.mongo_collection_meta,
+                    args.mongo_collection_history,
+                    args.retention_days
+                ))
+
+            hass_url = args.hass_url
+            print(hass_url)
+            if hass_url:
+                reporters.append(HassReporter(
+                    hass_url,
+                    args.hass_stack_disbalance,
+                    args.hass_max_battery_disbalance,
+                    args.hass_max_battery_disbalance_id,
+                    args.hass_token_file
+                ))
 
             logging.info("About to start polling...")
             bats = p.scan_for_batteries(2, 10)
 
             logging.info("Have battery stack data")
-            collection_meta.insert_one({'ts':  datetime.datetime.now().isoformat(), "stack": to_json_serializable(bats)})
 
+            for reporter in reporters:
+                reporter.report_meta(bats)
 
             for b in p.poll_parameters(bats.range()):
                 cc += 1
@@ -115,11 +142,14 @@ def run(argv: list[str]):
                     sys.stdout.write('\r' + spinner[cc % len(spinner)])
                     sys.stdout.flush()
 
+                mb = minimize(b)
                 # print(print_json(json.dumps(minimize(b))))
-                collection_hist.insert_one(minimize(b))
+                for reporter in reporters:
+                    reporter.report_state(mb)
 
                 if cc % 86400 == 0:
-                    mongo_cleanup(collection_hist, args.retention_days)
+                    for reporter in reporters:
+                        reporter.cleanup()
 
                 time.sleep(args.interval / 1000.0)
         except (KeyboardInterrupt, SystemExit):

src/pylontechpoller/reporter.py

@@ -0,0 +1,71 @@
+import datetime
+import json
+import logging
+
+import requests
+from pymongo import MongoClient
+
+from pylontech import to_json_serializable
+
+logger = logging.getLogger(__name__)
+
+class Reporter:
+    def report_meta(self, meta):
+        pass
+
+    def report_state(self, state):
+        pass
+
+    def cleanup(self):
+        pass
+
+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):
+        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}})
+
+class HassReporter(Reporter):
+    def __init__(self, hass_url, hass_stack_disbalance, hass_max_battery_disbalance, hass_max_battery_disbalance_id, hass_token_file):
+        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
+        with open(hass_token_file, 'r') as file:
+            self.hass_token = file.read().strip()
+
+
+    def report_state(self, state):
+        md = state["max_module_disbalance"]
+        self.update_hass_state(self.hass_stack_disbalance, state["stack_disbalance"])
+        self.update_hass_state(self.hass_max_battery_disbalance, md[1])
+        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}")