add write support and validation

needs more work and testing, but works for basic values. less than 1 register in size.

Author: HotNoob

inverter.py

@@ -115,19 +115,81 @@ def print_info(self):
         self.__log.info('\tUnit: %s\n', str(self.unit))
         self.__log.info('\tModbus Version: %s\n', str(self.modbus_version))
 
-    def read_variable(self, variable_name : str, registry_type : Registry_Type):
+    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"""
+
+        #read current value
+        current_registers = self.read_registers(start=entry.register, end=entry.register, registry_type=registry_type)
+        results = self.process_registery(current_registers, self.protocolSettings.get_registry_map(registry_type))
+        current_value = current_registers[entry.register]
+
+      
+        if not self.protocolSettings.validate_registry_entry(entry, current_value):
+            raise ValueError("Invalid value in register. unsafe to write")
+        
+        if not self.protocolSettings.validate_registry_entry(entry, value):
+            raise ValueError("Invalid new value. unsafe to write")
+        
+        #handle codes
+        if entry.variable_name+"_codes" in self.protocolSettings.codes:
+            codes = self.protocolSettings.codes[entry.variable_name+"_codes"]
+            for key, val in codes.items():
+                if val == value: #convert "string" to key value
+                    value = key
+                    break
+
+        #results[entry.variable_name]
+        ushortValue : int = None #ushort
+        if entry.data_type == Data_Type.USHORT:
+            ushortValue = int(value)
+            if ushortValue < 0 or ushortValue > 65535:
+                 raise ValueError("Invalid value")
+        elif entry.data_type.value > 200 or entry.data_type == Data_Type.BYTE: #bit types
+            bit_size = Data_Type.getSize(entry.data_type)
+
+            new_val = int(value)
+            if 0 > new_val or new_val > 2**bit_size:
+                raise ValueError("Invalid value")
+
+            bit_index = entry.register_bit
+            bit_mask = ((1 << bit_size) - 1) << bit_index  # Create a mask for extracting X bits starting from bit_index
+            clear_mask = ~(bit_mask)  # Mask for clearing the bits to be updated
+
+            # Clear the bits to be updated in the current_value
+            ushortValue = current_value & clear_mask
+
+            # Set the bits according to the new_value at the specified bit position
+            ushortValue |= (new_val << bit_index) & bit_mask
+
+            bit_mask = (1 << bit_size) - 1  # Create a mask for extracting X bits
+            check_value = (ushortValue >> bit_index) & bit_mask
+
+            if check_value != new_val:
+                raise ValueError("something went wrong bitwise")
+        else:
+            raise TypeError("Unsupported data type")
+            
+       
+
+        
+        if ushortValue == None:
+            raise ValueError("Invalid value - None")
+
+        self.reader.write_register(entry.register, ushortValue, registry_type=registry_type)
+
+
+    def read_variable(self, variable_name : str, registry_type : Registry_Type, entry : registry_map_entry = None):
         ##clean for convinecne  
-        variable_name = variable_name.strip().lower().replace(' ', '_')
-        if registry_type == Registry_Type.INPUT:
-            registry_map = self.protocolSettings.input_registry_map
-        elif registry_type == Registry_Type.HOLDING:
-            registry_map = self.protocolSettings.holding_registry_map
-
-        entry : registry_map_entry = None 
-        for e in registry_map:
-            if e.variable_name == variable_name:
-                entry = e
-                break
+        if variable_name:
+            variable_name = variable_name.strip().lower().replace(' ', '_')
+
+        registry_map = self.protocolSettings.get_registry_map(registry_type)
+
+        if entry == None:
+            for e in registry_map:
+                if e.variable_name == variable_name:
+                    entry = e
+                    break
 
         if entry:
             start : int = 0
@@ -148,6 +210,7 @@ def read_registers(self, ranges : list[tuple] = None, start : int = 0, end : int
 
 
         if not ranges: #ranges is empty, use min max
+            end = end + 1
             ranges = []
             start = start - batch_size
             while( start := start + batch_size ) < end:
@@ -218,7 +281,6 @@ def process_registery(self, registry : dict, map : list[registry_map_entry]) ->
 
             if item.register not in registry:
                 continue
-
             value = ''    
 
             if item.data_type == Data_Type.UINT: #read uint
@@ -320,6 +382,16 @@ def process_registery(self, registry : dict, map : list[registry_map_entry]) ->
                 info[item.variable_name] = value
 
         return info
+    
+    def read_registry(self, registry_type : Registry_Type = Registry_Type.INPUT) -> dict[str,str]:
+        map = self.protocolSettings.get_registry_map(registry_type)
+        if not map:
+            return {}
+        
+        registry = self.read_registers(self.protocolSettings.get_registry_ranges(registry_type), registry_type=registry_type)
+        info = self.process_registery(registry, map)
+        return info
+
 
     def read_input_registry(self) -> dict[str,str]:
         ''' reads input registers and returns as clean dict object inverters '''

invertermodbustomqtt.py

@@ -33,7 +33,7 @@
 
 from inverter import Inverter
 
-from protocol_settings import protocol_settings,Data_Type,registry_map_entry,Registry_Type
+from protocol_settings import protocol_settings,Data_Type,registry_map_entry,Registry_Type,WriteMode
 
 __logo = """
    ____                        _   _   ____  __  __  ___ _____ _____ 
@@ -102,6 +102,9 @@ class InverterModBusToMQTT:
 
     __max_precision : int = -1
 
+    __write : bool = False
+    ''' enable / disable write mode - setting'''
+
     __analyze_protocol : bool = False
     ''' enable / disable analyze mode'''
 
@@ -199,6 +202,7 @@ def init_invertermodbustomqtt(self):
             protocol_version = str(self.__settings.get(section, 'protocol_version'))
 
             self.__analyze_protocol = self.__settings.getboolean(section, 'analyze_protocol', fallback=False)
+            self.__write = self.__settings.getboolean(section, 'write', fallback=False)
             self.__analyze_protocol_save_load = self.__settings.getboolean(section, 'analyze_protocol_save_load', fallback=False)
 
 
@@ -317,9 +321,17 @@ def on_connect(self, client, userdata, flags, rc):
         self.__mqtt_connected = True
 
 
+    __write_topics : dict[str, registry_map_entry] = {}
+
     def on_message(self, client, userdata, msg):
         """ The callback for when a PUBLISH message is received from the server. """
-        self.__log.info(msg.topic+" "+str(msg.payload))
+        self.__log.info(msg.topic+" "+str(msg.payload.decode('utf-8')))
+
+        #self.inverter.protocolSettings.validate_registry_entry
+        if msg.topic in self.__write_topics:
+            entry = self.__write_topics[msg.topic]
+            self.inverter.write_variable(entry, value=str(msg.payload.decode('utf-8')))
+
 
     def run(self):
         """
@@ -340,6 +352,9 @@ def run(self):
 
         print("using serial number: " + self.__device_serial_number)
 
+        if self.__write:
+            self.enable_write()
+
         if self.__mqtt_discovery_enabled:
             self.mqtt_discovery()
 
@@ -424,6 +439,55 @@ def run(self):
                 # If all the inverters are not online because no power is being generated then we sleep for 1 min
                 time.sleep(self.__offline_interval)
 
+
+    def enable_write(self):
+        """
+        enable write to modbus; must pass tests.
+        """
+        print("Validating Protocol for Writing")
+        self.__write = False
+        score_percent = self.validate_protocol(Registry_Type.HOLDING)
+        if(score_percent > 90):
+            self.__write = True
+            print("enable write - validation passed")
+            
+            self.__write_topics = {}
+            #subscribe to write topics
+            for entry in self.inverter.protocolSettings.holding_registry_map:
+                if entry.write_mode == WriteMode.WRITE:
+                    #__write_topics
+                    topic : str = self.__mqtt_topic + "/write/" + entry.variable_name.lower().replace(' ', '_')
+                    self.__write_topics[topic] = entry
+                    self.__mqtt_client.subscribe(topic)
+    
+    def validate_protocol(self, registry_type : Registry_Type = Registry_Type.INPUT) -> float:
+        """
+        validate protocol
+        """
+
+        score : float = 0
+        info = {}
+        registry_map : list[registry_map_entry] = self.inverter.protocolSettings.get_registry_map(registry_type)
+        info = self.inverter.read_registry(registry_type)
+
+        for value in registry_map:
+            if value.variable_name in info:
+                evaluate = True
+
+                if value.concatenate and value.register != value.concatenate_registers[0]: #only eval concated values once
+                    evaluate = False
+                  
+                if evaluate:
+                    score = score + self.inverter.protocolSettings.validate_registry_entry(value, info[value.variable_name])
+
+        maxScore = len(registry_map)
+        percent = score*100/maxScore
+        print("validation score: " + str(score) + " of " + str(maxScore) + " : " + str(round(percent)) + "%")
+        return percent
+
+
+    
+
     def analyze_protocol(self, settings_dir : str = 'protocols'):
         print("=== PROTOCOL ANALYZER ===")
         protocol_names : list[str] = []
@@ -602,6 +666,8 @@ def mqtt_discovery(self):
             if item.concatenate and item.register != item.concatenate_registers[0]:
                 continue #skip all except the first register so no duplicates
 
+            if item.write_mode == WriteMode.READDISABLED: #disabled
+                continue
 
             clean_name = item.variable_name.lower().replace(' ', '_')
 
@@ -620,13 +686,18 @@ def mqtt_discovery(self):
             disc_payload['device'] = device
             disc_payload['name'] = clean_name
             disc_payload['unique_id'] = "hotnoob_" + self.__device_serial_number + "_"+clean_name
-            disc_payload['state_topic'] = self.__mqtt_topic + "/"+clean_name
+
+            writePrefix = ""
+            if self.__write and item.write_mode == WriteMode.WRITE:
+                writePrefix = "" #home assistant doesnt like write prefix
+
+            disc_payload['state_topic'] = self.__mqtt_topic +writePrefix+ "/"+clean_name
 
             if item.unit:
                 disc_payload['unit_of_measurement'] = item.unit
 
 
-            discovery_topic = self.__mqtt_discovery_topic+"/sensor/inverter-" + self.__device_serial_number  + "/" + disc_payload['name'].replace(' ', '_') + "/config"
+            discovery_topic = self.__mqtt_discovery_topic+"/sensor/inverter-" + self.__device_serial_number  + writePrefix + "/" + disc_payload['name'].replace(' ', '_') + "/config"
 
             self.__mqtt_client.publish(discovery_topic,
                                        json.dumps(disc_payload),qos=1, retain=True)