V1.1.11 Home Assistant Discovery, EG4 Protocol Definition, and other small misc Improvements

.gitignore

@@ -38,4 +38,6 @@ python_protocol_gateway.egg-info/*
 inflxudb_backlog/*
 
 #ignore dumps for dev
-tools/dumps/*
+tools/dumps/*
+/.vs/PythonProtocolGateway.slnx/FileContentIndex
+/.vs

classes/protocol_settings.py

@@ -11,7 +11,7 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Union
 
-from defs.common import strtoint
+from defs.common import strtoint, strtobool_or_og
 
 if TYPE_CHECKING:
     from configparser import SectionProxy
@@ -224,6 +224,8 @@ class registry_map_entry:
     write_mode : WriteMode = WriteMode.READ
     ''' enable disable reading/writing '''
 
+    ha_disc : dict[str, any] = None
+
     def __str__(self):
         return self.variable_name
 
@@ -593,7 +595,7 @@ def process_row(row):
                             end = strtoint(groups["range_end"])
                             values.extend(range(start, end + 1))
                         else:
-                            values.append(groups["element"])
+                            values.append(strtoint(groups["element"]))
                 else:
                     matched : bool = False
                     val_match = range_regex.search(row["values"])
@@ -677,6 +679,10 @@ def process_row(row):
             if "write" in row:
                 writeMode = WriteMode.fromString(row["write"])
 
+            ha_discovery = {}
+            if "ha discovery" in row and row["ha discovery"]:
+                ha_discovery = {key.strip().lower(): strtobool_or_og(value.strip().lower()) for key, value in dict(disc.split(":") for disc in row["ha discovery"].split(",")).items()}
+
             for i in r:
                 item = registry_map_entry(
                                             registry_type = registry_type,
@@ -698,7 +704,8 @@ def process_row(row):
                                             value_regex=value_regex,
                                             read_command = read_command,
                                             read_interval=read_interval,
-                                            write_mode=writeMode
+                                            write_mode=writeMode,
+                                            ha_disc = ha_discovery
                                         )
                 registry_map.append(item)
 
@@ -762,7 +769,6 @@ def process_row(row):
                             else:
                                 combined_item.data_type = Data_Type.UINT
 
-
                         if combined_item.documented_name == combined_item.variable_name:
                             combined_item.variable_name = combined_item.variable_name[:-2].strip()
 
@@ -1216,8 +1222,14 @@ def validate_registry_entry(self, entry : registry_map_entry, val) -> int:
                                 return len(entry.concatenate_registers)
 
             else: #default type
-                intval = int(val)
-                if intval >= entry.value_min and intval <= entry.value_max:
+                #apply unit mod before comparison to min/maxes
+                if entry.unit_mod != 1:
+                    intval = int(float(val) / entry.unit_mod)
+                else:
+                    intval = int(val)
+                if intval in entry.values:
+                    return 1
+                elif intval >= entry.value_min and intval <= entry.value_max:
                     return 1
 
                 self._log.error(f"validate_registry_entry '{entry.variable_name}' fail (INT) {intval} != {entry.value_min}~{entry.value_max}")

classes/transports/modbus_base.py

@@ -166,8 +166,6 @@ def enable_write(self):
             else:
                 self._log.error("enable write FAILED - WRITE DISABLED")
 
-
-
     def write_data(self, data : dict[str, str], from_transport : transport_base) -> None:
         if not self.write_enabled:
             return
@@ -216,7 +214,6 @@ def validate_protocol(self, protocolSettings : "protocol_settings") -> float:
         score_percent = self.validate_registry(Registry_Type.HOLDING)
         return score_percent
 
-
     def validate_registry(self, registry_type : Registry_Type = Registry_Type.INPUT) -> float:
         score : float = 0
         info = {}
@@ -393,7 +390,6 @@ def evaluate_score(entry : registry_map_entry, val):
             self._log.debug("input register score: " + str(input_register_score[name]) + "; valid registers: "+str(input_valid_count[name])+" of " + str(len(protocols[name].get_registry_map(Registry_Type.INPUT))))
             self._log.debug("holding register score : " + str(holding_register_score[name]) + "; valid registers: "+str(holding_valid_count[name])+" of " + str(len(protocols[name].get_registry_map(Registry_Type.HOLDING))))
 
-
     def write_variable(self, entry : registry_map_entry, value : str, registry_type : Registry_Type = Registry_Type.HOLDING):
         """ writes a value to a ModBus register; todo: registry_type to handle other write functions"""
 
@@ -493,7 +489,6 @@ def write_variable(self, entry : registry_map_entry, value : str, registry_type
         self.write_register(entry.register, ushortValue)
         #entry.next_read_timestamp = 0 #ensure is read next interval
 
-
     def read_variable(self, variable_name : str, registry_type : Registry_Type, entry : registry_map_entry = None):
         ##clean for convinecne
         if variable_name:

classes/transports/mqtt.py

@@ -3,6 +3,7 @@
 import random
 import time
 import warnings
+import copy
 from configparser import SectionProxy
 
 import paho.mqtt.client
@@ -11,11 +12,24 @@
 from paho.mqtt.client import MQTT_ERR_NO_CONN
 from paho.mqtt.client import Client as MQTTClient
 
-from defs.common import strtobool
+from defs.common import project_details, strtobool
 
 from ..protocol_settings import Registry_Type, WriteMode, registry_map_entry
 from .transport_base import transport_base
 
+unit_to_discovery = {
+    'kwh'   :   dict({'device_class':'energy','state_class':'total','mdi':'meter-electric','enabled_by_default':True}),
+    'wh'    :   dict({'device_class':'energy','state_class':'total','mdi':'meter-electric','enabled_by_default':True}),
+    'v'     :   dict({'device_class':'voltage','state_class':'measurement','mdi':'lightning-bolt','enabled_by_default':True}),
+    'va'    :   dict({'device_class':'apparent_power','state_class':'measurement','enabled_by_default':True}),
+    'a'     :   dict({'device_class':'current','state_class':'measurement','mdi':'current-ac','enabled_by_default':True}),
+    'w'     :   dict({'device_class':'power','state_class':'measurement','mdi':'power','enabled_by_default':True}),
+    's'     :   dict({'device_class':'duration','state_class':'measurement','mdi':'timer','enabled_by_default':True}),
+    'ms'    :   dict({'device_class':'duration','state_class':'measurement','mdi':'timer','enabled_by_default':True}),
+    '°c'     :   dict({'device_class':'temperature','state_class':'measurement','mdi':'thermometer','enabled_by_default':True}),
+    'hz'    :   dict({'device_class':'frequency','state_class':'measurement','enabled_by_default':True})
+    }
+
 
 class mqtt(transport_base):
     ''' for future; this will hold mqtt transport'''
@@ -163,7 +177,9 @@ def write_data(self, data : dict[str, str], from_transport : transport_base):
         self._log.info(f"write data from [{from_transport.transport_name}] to mqtt transport")
         self._log.info(data)
         #have to send this every loop, because mqtt doesnt disconnect when HA restarts. HA bug.
+
         info = self.client.publish(self.base_topic + "/" + from_transport.device_identifier + "/availability","online", qos=0,retain=True)
+
         if info.rc == MQTT_ERR_NO_CONN:
             self.connected = False
 
@@ -215,10 +231,25 @@ def mqtt_discovery(self, from_transport : transport_base):
         device["identifiers"] = "hotnoob_" + from_transport.device_model + "_" + from_transport.device_serial_number
         device["name"] = from_transport.device_name
 
+        data = project_details()
+        origin = {}
+        origin["name"] = data['project']['name']
+        origin["sw"] = data['project']['version']
+        origin["url"] = data['project']['urls']['Homepage']
+
         registry_map : list[registry_map_entry] = []
         for entries in from_transport.protocolSettings.registry_map.values():
             registry_map.extend(entries)
 
+        disc_payload_base = {}
+        disc_payload_base["availability_topic"] = self.base_topic + "/" + from_transport.device_identifier + "/availability"
+        disc_payload_base["device"] = device
+        disc_payload_base["origin"] = origin
+        disc_payload_base["cmps"] = {}
+        disc_payload = copy.deepcopy(disc_payload_base)
+
+        write_only = {}
+
         length = len(registry_map)
         count = 0
         for item in registry_map:
@@ -230,7 +261,6 @@ def mqtt_discovery(self, from_transport : transport_base):
             if item.write_mode == WriteMode.READDISABLED: #disabled
                 continue
 
-
             clean_name = item.variable_name.lower().replace(" ", "_").strip()
             if not clean_name: #if name is empty, skip
                 continue
@@ -242,35 +272,80 @@ def mqtt_discovery(self, from_transport : transport_base):
                 if self.__holding_register_prefix and item.registry_type == Registry_Type.HOLDING:
                     clean_name = self.__holding_register_prefix + clean_name
 
-
             print(("#Publishing Topic "+str(count)+" of " + str(length) + ' "'+str(clean_name)+'"').ljust(100)+"#", end="\r", flush=True)
 
             #device['sw_version'] = bms_version
-            disc_payload = {}
-            disc_payload["availability_topic"] = self.base_topic + "/" + from_transport.device_identifier + "/availability"
-            disc_payload["device"] = device
-            disc_payload["name"] = clean_name
-            disc_payload["unique_id"] = "hotnoob_" + from_transport.device_serial_number + "_"+clean_name
+            unique_id = "hotnoob_" + from_transport.device_serial_number + "_" + clean_name
+            disc_payload["cmps"][unique_id] = {}
+            disc_payload["cmps"][unique_id]["name"] = clean_name
+            disc_payload["cmps"][unique_id]["unique_id"] = unique_id
+            disc_payload["cmps"][unique_id].update(dict({'enabled_by_default': 'true'}))
+
+            if item.unit:
+                disc_payload["cmps"][unique_id]["unit_of_measurement"] = item.unit
+
+                for key, value in unit_to_discovery.items():
+                    if str(item.unit).lower() == key:
+                        disc_payload["cmps"][unique_id].update(value)
+                        break
+
+            disc_payload["cmps"][unique_id].update(dict({'p': 'sensor'}))
+
+            if from_transport.write_enabled:
+                if item.write_mode == WriteMode.WRITE or WriteMode.WRITEONLY:
+                    #determine appropriate command topic
+                    for key, value in self._mqtt__write_topics.items():
+                        if value == item:
+                            disc_payload["cmps"][unique_id].update(dict({'command_topic': key}))
+
+                            if item.variable_name+"_codes" in from_transport.protocolSettings.codes:
+                                codes = from_transport.protocolSettings.codes[item.variable_name+"_codes"]
+                                disc_payload["cmps"][unique_id].update(dict({'options': list(codes.values()) }))
+                                disc_payload["cmps"][unique_id].update(dict({'p': 'select'}))
+                            else:
+                                #if the min/max is 0/1 then its probably a switch rather than a number entry
+                                if item.value_min == 0 and item.value_max == 1:
+                                    disc_payload["cmps"][unique_id].update(dict({'p': 'switch'}))
+                                    disc_payload["cmps"][unique_id].update(dict({'payload_on': 1}))
+                                    disc_payload["cmps"][unique_id].update(dict({'payload_off': 0}))
+                                #if there is only one value its probably a button rather than a number
+                                elif item.value_min == 1 and item.value_max == 1:
+                                    disc_payload["cmps"][unique_id].update(dict({'p': 'button'}))
+                                    disc_payload["cmps"][unique_id].update(dict({'payload_press': 1})) 
+                                else:
+                                    disc_payload["cmps"][unique_id].update(dict({'p': 'number'}))
+                                disc_payload["cmps"][unique_id].update(dict({'min': item.value_min * item.unit_mod}))
+                                disc_payload["cmps"][unique_id].update(dict({'max': item.value_max * item.unit_mod}))
+                            break
+
+            disc_payload["cmps"][unique_id].update(item.ha_disc)
 
             writePrefix = ""
             if from_transport.write_enabled and ( item.write_mode == WriteMode.WRITE or item.write_mode == WriteMode.WRITEONLY ):
                 writePrefix = "" #home assistant doesnt like write prefix
 
-            disc_payload["state_topic"] = self.base_topic + "/" +from_transport.device_identifier + writePrefix+ "/"+clean_name
-
-            if item.unit:
-                disc_payload["unit_of_measurement"] = item.unit
+            disc_payload["cmps"][unique_id]["state_topic"] = self.base_topic + "/" + from_transport.device_identifier + writePrefix+ "/" + clean_name
 
+            discovery_topic = self.discovery_topic+"/device/HN-" + from_transport.device_serial_number  + writePrefix + "/config"
 
-            discovery_topic = self.discovery_topic+"/sensor/HN-" + from_transport.device_serial_number  + writePrefix + "/" + disc_payload["name"].replace(" ", "_") + "/config"
-
-            self.client.publish(discovery_topic,
-                                       json.dumps(disc_payload),qos=1, retain=True)
-
-            #send WO message to indicate topic is write only
+            #add WO message to be sent later to indicate topic is write only
             if item.write_mode == WriteMode.WRITEONLY:
-                self.client.publish(disc_payload["state_topic"], "WRITEONLY")
+                write_only[disc_payload["cmps"][unique_id]["state_topic"]] = "WRITEONLY"
+
+            #break up items into batches to make the messages smaller (10 was arbitrarily picked but it seems to work well enough)
+            if len(disc_payload["cmps"]) > 10:
+                self.client.publish(discovery_topic, json.dumps(disc_payload),qos=1, retain=True)
+                #reset component list for next batch
+                disc_payload = copy.deepcopy(disc_payload_base)
+                time.sleep(0.07)
+
+        #publish whatever is left
+        if len(disc_payload["cmps"]) > 0:
+            self.client.publish(discovery_topic, json.dumps(disc_payload),qos=1, retain=True)
+            time.sleep(0.07) #slow down for better reliability
 
+        for t, val in write_only.items():
+            self.client.publish(t, val)
             time.sleep(0.07) #slow down for better reliability
 
         self.client.publish(disc_payload["availability_topic"],"online",qos=0, retain=True)

defs/common.py

@@ -3,6 +3,20 @@
 
 from serial.tools import list_ports
 
+import tomli
+
+def project_details():
+    try:
+        par_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
+        file_path = os.path.join(par_dir, 'pyproject.toml')
+        with open(file_path, 'rb') as f:
+            data = tomli.load(f)
+            return data
+    except FileNotFoundError:
+        print(f"Error: The file '{file_path}' was not found.")
+    except tomli.TOMLDecodeError:
+        print("Error: Could not decode the TOML file. Check for syntax errors.")
+
 
 def strtobool (val):
     """Convert a string representation of truth to true (1) or false (0).
@@ -17,13 +31,27 @@ def strtobool (val):
 
     return 0
 
+def strtobool_or_og(val):
+    """Convert a string representation of truth to boolean if it represents a boolean otherwise returns the original value.
+    True values are 'y', 'yes', 't', 'true', 'on', and '1'
+    False values are 'n', 'no', 'f', 'false', 'off', and '0'
+    """
+    if isinstance(val, str):
+        clean_val = val.strip().lower()
+        if clean_val in ("y", "yes", "t", "true", "on", "1"):
+            return True
+        elif clean_val in ("n", "no", "f", "false", "off", "0"):
+            return False
+    return val
+
 def strtoint(val : str) -> int:
     ''' converts str to int, but allows for hex string input, identified by x prefix'''
 
     if isinstance(val, int): #is already int.
         return val
 
     val = val.lower().strip()
+    val = re.sub(r'^\W+|\W+$', '', val)
 
     if val and val[0] == "x":
         val = val[1:]