Merge pull request HotNoob#55 from HotNoob/v1.1.6

V1.1.6

Author: HotNoob

.gitignore

@@ -14,6 +14,7 @@ growatt2mqtt.cfg
 growatt2mqtt.service
 config.cfg
 *.cfg
+*.old
 invertermodbustomqtt.service
 log.txt
 variable_mask.txt

README.md

@@ -6,17 +6,18 @@
 
 [![CodeQL](https://github.com/HotNoob/PythonProtocolGateway/actions/workflows/github-code-scanning/codeql/badge.svg)](https://github.com/HotNoob/PythonProtocolGateway/actions/workflows/github-code-scanning/codeql)
 
-For advanced configuration help, please checkout the wiki :)
+For advanced configuration help, please checkout the documentation :)
+https://github.com/HotNoob/PythonProtocolGateway/tree/main/documentation
 
 # Python Protocol Gateway
 
 Python Protocol Gateway is a python-based service that reads data via Modbus RTU or other protocols and translates the data for MQTT. 
 Configuration is handled via a small config files. 
 In the long run, Python Protocol Gateway will become a general purpose protocol gateway to translate between more than just modbus and mqtt. 
 
-For specific device installation instructions please checkout the wiki:
+For specific device installation instructions please checkout the documentation:
 Growatt, EG4, Sigineer, SOK, PACE-BMS
-https://github.com/HotNoob/PythonProtocolGateway/wiki
+https://github.com/HotNoob/PythonProtocolGateway/tree/main/documentation
 
 # General Installation
 Connect the USB port on the inverter into your computer / device. This port is essentially modbus usb adapter.
@@ -56,7 +57,8 @@ eg4_v58 = eg4 inverters ( EG4-6000XP ) - confirmed working
 eg4_3000ehv_v1 = eg4 inverters ( EG4_3000EHV )
 ```
 
-more details on these protocols can be found in the wiki
+more details on these protocols can be found in the documentation:
+https://github.com/HotNoob/PythonProtocolGateway/tree/main/documentation
 
 ### run as script
 ```

classes/protocol_settings.py

@@ -2,6 +2,7 @@
 from dataclasses import dataclass
 from enum import Enum
 import glob
+import logging
 from typing import Union
 from defs.common import strtoint
 import itertools
@@ -233,7 +234,18 @@ class protocol_settings:
     settings : dict[str, str]
     ''' default settings provided by protocol json '''
 
+    byteorder : str = "big"
+
+    _log : logging.Logger = None
+
+
     def __init__(self, protocol : str, settings_dir : str = 'protocols'):
+
+        #apply log level to logger
+        self._log_level = getattr(logging, logging.getLevelName(logging.getLogger().getEffectiveLevel()), logging.INFO)
+        self._log : logging.Logger = logging.getLogger(__name__)
+        self._log.setLevel(self._log_level)
+
         self.protocol = protocol
         self.settings_dir = settings_dir
 
@@ -266,6 +278,8 @@ def __init__(self, protocol : str, settings_dir : str = 'protocols'):
         else:
             self.transport = "modbus_rtu"
 
+        if "byteorder" in self.settings: #handle byte order for ints n stuff
+            self.byteorder = self.settings["byteorder"]
 
         for registry_type in Registry_Type:
             self.load_registry_map(registry_type)
@@ -305,7 +319,7 @@ def load__json(self, file : str = '', settings_dir : str = ''):
 
         #if path does not exist; nothing to load. skip.
         if not path:
-            print("ERROR: '"+file+"' not found")
+            self._log.error("ERROR: '"+file+"' not found")
             return
 
         with open(path) as f:
@@ -321,13 +335,13 @@ def load__json(self, file : str = '', settings_dir : str = ''):
 
     def load__registry(self, path, registry_type : Registry_Type = Registry_Type.INPUT) -> list[registry_map_entry]: 
         registry_map : list[registry_map_entry] = []
-        register_regex = re.compile(r'(?P<register>(?:0?x[\dA-Z]+|[\d]+))\.(b(?P<bit>x?\d{1,2})|(?P<byte>x?\d{1,2}))')
+        register_regex = re.compile(r'(?P<register>(?:0?x[\da-z]+|[\d]+))\.(b(?P<bit>x?\d{1,2})|(?P<byte>x?\d{1,2}))')
 
         data_type_regex = re.compile(r'(?P<datatype>\w+)\.(?P<length>\d+)')
 
-        range_regex = re.compile(r'(?P<reverse>r|)(?P<start>(?:0?x[\dA-Z]+|[\d]+))[\-~](?P<end>(?:0?x[\dA-Z]+|[\d]+))')
+        range_regex = re.compile(r'(?P<reverse>r|)(?P<start>(?:0?x[\da-z]+|[\d]+))[\-~](?P<end>(?:0?x[\da-z]+|[\d]+))')
         ascii_value_regex = re.compile(r'(?P<regex>^\[.+\]$)')
-        list_regex = re.compile(r'\s*(?:(?P<range_start>(?:0?x[\dA-Z]+|[\d]+))-(?P<range_end>(?:0?x[\dA-Z]+|[\d]+))|(?P<element>[^,\s][^,]*?))\s*(?:,|$)')
+        list_regex = re.compile(r'\s*(?:(?P<range_start>(?:0?x[\da-z]+|[\d]+))-(?P<range_end>(?:0?x[\da-z]+|[\d]+))|(?P<element>[^,\s][^,]*?))\s*(?:,|$)')
 
 
         if not os.path.exists(path): #return empty is file doesnt exist.
@@ -386,7 +400,7 @@ def determine_delimiter(first_row) -> str:
                 variable_name = variable_name = variable_name.strip().lower().replace(' ', '_').replace('__', '_') #clean name
 
                 if re.search(r"[^a-zA-Z0-9\_]", variable_name) :
-                    print("WARNING Invalid Name : " + str(variable_name) + " reg: " + str(row['register']) + " doc name: " + str(row['documented name']) + " path: " + str(path))
+                    self._log.warning("Invalid Name : " + str(variable_name) + " reg: " + str(row['register']) + " doc name: " + str(row['documented name']) + " path: " + str(path))
 
                 #convert to float
                 try:
@@ -402,7 +416,7 @@ def determine_delimiter(first_row) -> str:
 
                 if 'values' not in row:
                     row['values'] = ""
-                    print("WARNING No Value Column : path: " + str(path)) 
+                    self._log.warning("No Value Column : path: " + str(path)) 
 
                 data_type_len : int = -1
                 #optional row, only needed for non-default data types
@@ -471,6 +485,7 @@ def determine_delimiter(first_row) -> str:
                 register : int = -1
                 register_bit : int = 0
                 register_byte : int = -1
+                row['register'] = row['register'].lower() #ensure is all lower case
                 match = register_regex.search(row['register'])
                 if match:
                     register = strtoint(match.group('register'))
@@ -673,21 +688,25 @@ def load_registry_map(self, registry_type : Registry_Type, file : str = '', sett
     def process_register_bytes(self, registry : dict[int,bytes], entry : registry_map_entry):
         ''' process bytes into data'''
 
-        register = registry[entry.register]
+        if isinstance(registry[entry.register], tuple):
+            register = registry[entry.register][0] #can bus uses tuple for timestamp
+        else:
+            register = registry[entry.register]
+            
         if entry.register_byte > 0:
             register = register[entry.register_byte:]
 
         if entry.data_type_size > 0:
             register = register[:entry.data_type_size]
 
         if entry.data_type == Data_Type.UINT:
-            value = int.from_bytes(register[:4], byteorder='big', signed=False)
+            value = int.from_bytes(register[:4], byteorder=self.byteorder, signed=False)
         elif entry.data_type == Data_Type.INT:
-            value = int.from_bytes(register[:4], byteorder='big', signed=True)
+            value = int.from_bytes(register[:4], byteorder=self.byteorder, signed=True)
         elif entry.data_type == Data_Type.USHORT:
-            value = int.from_bytes(register[:2], byteorder='big', signed=False)
+            value = int.from_bytes(register[:2], byteorder=self.byteorder, signed=False)
         elif entry.data_type == Data_Type.SHORT:
-            value = int.from_bytes(register[:2], byteorder='big', signed=True)
+            value = int.from_bytes(register[:2], byteorder=self.byteorder, signed=True)
         elif entry.data_type == Data_Type._16BIT_FLAGS or entry.data_type == Data_Type._8BIT_FLAGS or entry.data_type == Data_Type._32BIT_FLAGS:
             #16 bit flags
             start_bit : int = 0
@@ -700,18 +719,31 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
             #handle custom sizes, less than 1 register
             end_bit = flag_size + start_bit
 
-            if entry.documented_name+'_codes' in self.protocolSettings.codes:
+            if entry.documented_name+'_codes' in self.codes:
+                code_key : str = entry.documented_name+'_codes'
                 flags : list[str] = []
+                flag_indexes : list[str] = []
                 for i in range(start_bit, end_bit):  # Iterate over each bit position (0 to 15)
                     byte = i // 8 
                     bit = i % 8
                     val = register[byte]
                     # Check if the i-th bit is set
                     if (val >> bit) & 1:
                         flag_index = "b"+str(i)
-                        if flag_index in self.protocolSettings.codes[entry.documented_name+'_codes']:
-                            flags.append(self.protocolSettings.codes[entry.documented_name+'_codes'][flag_index])
-                        
+                        flag_indexes.append(flag_index)
+                        if flag_index in self.codes[code_key]:
+                            flags.append(self.codes[code_key][flag_index])
+
+                #check multibit flags
+                multibit_flags = [key for key in self.codes if '&' in key]
+
+                if multibit_flags: #if multibit flags are found
+                    flag_indexes_set : set[str] = set(flag_indexes)
+                    for multibit_flag in multibit_flags:
+                        bits = multibit_flag.split('&')  # Split key into 'bits'
+                        if all(bit in flag_indexes_set for bit in bits): # Check if all bits are present in the flag_indexes_set
+                            flags.append(self.codes[code_key][multibit_flag])
+                    
                 value = ",".join(flags)
             else:
                 flags : list[str] = []
@@ -760,7 +792,23 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
             try:
                 value = register.decode("utf-8") #convert bytes to ascii
             except UnicodeDecodeError as e:
-                print("UnicodeDecodeError:", e)
+                self._log.error("UnicodeDecodeError:", e)
+
+        #apply unit mod
+        if entry.unit_mod != float(1):
+            value = value * entry.unit_mod
+
+        #apply codes
+        if (entry.data_type != Data_Type._16BIT_FLAGS and
+            entry.documented_name+'_codes' in self.codes):
+            try:
+                cleanval = str(int(value))
+        
+                if cleanval in self.codes[entry.documented_name+'_codes']:
+                    value = self.codes[entry.documented_name+'_codes'][cleanval]
+            except:
+                #do nothing; try is for intval
+                value = value
 
         return value
 
@@ -855,11 +903,11 @@ def process_register_ushort(self, registry : dict[int, int], entry : registry_ma
                 bit_index = entry.register_bit
                 value = (registry[entry.register] >> bit_index) & bit_mask
         elif entry.data_type == Data_Type.ASCII:
-            value = registry[entry.register].to_bytes((16 + 7) // 8, byteorder='big') #convert to ushort to bytes
+            value = registry[entry.register].to_bytes((16 + 7) // 8, byteorder=self.byteorder) #convert to ushort to bytes
             try:
                 value = value.decode("utf-8") #convert bytes to ascii
             except UnicodeDecodeError as e:
-                print("UnicodeDecodeError:", e)
+                self._log.error("UnicodeDecodeError:", e)
 
         else: #default, Data_Type.USHORT
             value = float(registry[entry.register])