Merge pull request #43 from HotNoob/v1.1.4

V1.1.4

Author: HotNoob

.github/workflows/python-package.yml

@@ -33,8 +33,10 @@ jobs:
       run: |
         # stop the build if there are Python syntax errors or undefined names
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+
+        # don't care about coding standards
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        # flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
         pytest

.gitignore

@@ -2,7 +2,13 @@
 *.pyc
 .vscode
 settings.json
+
+# Ignore Python virtual environment directories
 virtualenv/*
+venv/
+env/
+.venv/
+.env/
 
 growatt2mqtt.cfg
 growatt2mqtt.service

README.md

@@ -41,11 +41,13 @@ protocol_version = {{version}}
 v0.14 = growatt inverters 2020+
 sigineer_v0.11 = sigineer inverters
 growatt_2020_v1.24 = alt protocol for large growatt inverters - currently untested
-eg4_v58 = eg4 inverters ( EG4-6000XP ) - confirmed working
 srne_v3.9 = SRNE inverters - Untested
 victron_gx_3.3 = Victron GX Devices - Untested
 solark_v1.1 = SolarArk 8/12K Inverters - Untested
 hdhk_16ch_ac_module = some chinese current monitoring device :P
+
+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

classes/protocol_settings.py

@@ -1,6 +1,7 @@
 import csv
 from dataclasses import dataclass
 from enum import Enum
+from typing import Union
 from defs.common import strtoint
 import itertools
 import json
@@ -22,6 +23,8 @@ class Data_Type(Enum):
     '''32 bit signed int'''
     _16BIT_FLAGS = 7
     _8BIT_FLAGS = 8
+    _32BIT_FLAGS = 9
+
 
     ASCII = 84
     ''' 2 characters '''
@@ -104,7 +107,9 @@ def getSize(cls, data_type : 'Data_Type'):
                     Data_Type.UINT : 32,
                     Data_Type.SHORT : 16,
                     Data_Type.INT : 32,
-                    Data_Type._16BIT_FLAGS : 16
+                    Data_Type._8BIT_FLAGS : 8,
+                    Data_Type._16BIT_FLAGS : 16,
+                    Data_Type._32BIT_FLAGS : 32
                  }
 
         if data_type in sizes:
@@ -143,6 +148,7 @@ def fromString(cls, name : str):
             "READDISABLED"  : "READDISABLED",
             "DISABLED"      : "READDISABLED",
             "D"             : "READDISABLED",
+            "R/W"    : "WRITE",
             "RW"    : "WRITE",
             "W"     : "WRITE",
             "YES"   : "WRITE"
@@ -337,7 +343,7 @@ def determine_delimiter(first_row) -> str:
             if first_row.count(';') < first_row.count(','):
                 delimeter = ','
 
-            first_row = re.sub(r"\s+" + re.escape(delimeter) +"|" + re.escape(delimeter) +"\s+", delimeter, first_row) #trim values
+            first_row = re.sub(r"\s+" + re.escape(delimeter) +"|" + re.escape(delimeter) +r"\s+", delimeter, first_row) #trim values
 
             csvfile = itertools.chain([first_row], csvfile) #add clean header to begining of iterator 
 
@@ -357,7 +363,7 @@ def determine_delimiter(first_row) -> str:
                     character_part = row['unit']
                 else:
                     # Use regular expressions to extract numeric and character parts
-                    matches = re.findall(r'([0-9.]+)|(.*?)$', row['unit'])
+                    matches = re.findall(r'(\-?[0-9.]+)|(.*?)$', row['unit'])
 
                     # Iterate over the matches and assign them to appropriate variables
                     for match in matches:
@@ -373,7 +379,7 @@ def determine_delimiter(first_row) -> str:
                 variable_name = row['variable name'] if row['variable name'] else row['documented name']
                 variable_name = variable_name = variable_name.strip().lower().replace(' ', '_').replace('__', '_') #clean name
 
-                if re.search("[^a-zA-Z0-9\_]", variable_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))
 
                 #convert to float
@@ -659,15 +665,21 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
             value = int.from_bytes(register[:2], byteorder='big', signed=False)
         elif entry.data_type == Data_Type.SHORT:
             value = int.from_bytes(register[:2], byteorder='big', signed=True)
-        elif entry.data_type == Data_Type._16BIT_FLAGS or entry.data_type == Data_Type._8BIT_FLAGS:
+        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
-            if entry.data_type == Data_Type._8BIT_FLAGS:
-                start_bit = 8
+            end_bit : int = 16 #default 16 bit
+            flag_size : int = Data_Type.getSize(entry.data_type)
+
+            if entry.register_bit > 0: #handle custom bit offset
+                start_bit = entry.register_bit
+
+            #handle custom sizes, less than 1 register
+            end_bit = flag_size + start_bit
 
             if entry.documented_name+'_codes' in self.protocolSettings.codes:
                 flags : list[str] = []
-                for i in range(start_bit, 16):  # Iterate over each bit position (0 to 15)
+                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]
@@ -680,7 +692,7 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
                 value = ",".join(flags)
             else:
                 flags : list[str] = []
-                for i in range(start_bit, 16):  # Iterate over each bit position (0 to 15)
+                for i in range(start_bit, end_bit):  # Iterate over each bit position (0 to 15)
                     # Check if the i-th bit is set
                     if (val >> i) & 1:
                         flags.append("1")
@@ -762,32 +774,58 @@ def process_register_ushort(self, registry : dict[int, int], entry : registry_ma
             value = -value
             #value = struct.unpack('<h', bytes([min(max(registry[item.register], 0), 255), min(max(registry[item.register+1], 0), 255)]))[0]
             #value = int.from_bytes(bytes([registry[item.register], registry[item.register + 1]]), byteorder='little', signed=True)
-        elif entry.data_type == Data_Type._16BIT_FLAGS or entry.data_type == Data_Type._8BIT_FLAGS:
-            val = registry[entry.register]
+        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
-            if entry.data_type == Data_Type._8BIT_FLAGS:
-                start_bit = 8
+            end_bit : int = 16 #default 16 bit
+            flag_size : int = Data_Type.getSize(entry.data_type)
+
+            if entry.register_bit > 0: #handle custom bit offset
+                start_bit = entry.register_bit
+
+            #handle custom sizes, less than 1 register
+            end_bit = flag_size + start_bit
+
+            offset : int = 0
+            #calculate current offset for mutliregiter values, were assuming concatenate registers is in order, 0 being the first / lowest
+            #offset should always be >= 0
+            if entry.concatenate:
+                offset : int = entry.register - entry.concatenate_registers[0]
+
+            #compensate for current offset
+            end_bit = end_bit - (offset * 16)
+
+            val = registry[entry.register]
 
             if entry.documented_name+'_codes' in self.codes:
                 flags : list[str] = []
-                for i in range(start_bit, 16):  # Iterate over each bit position (0 to 15)
-                    # Check if the i-th bit is set
-                    if (val >> i) & 1:
-                        flag_index = "b"+str(i)
-                        if flag_index in self.codes[entry.documented_name+'_codes']:
-                            flags.append(self.codes[entry.documented_name+'_codes'][flag_index])
+                offset : int = 0
+
+                if end_bit > 0:
+                    end : int = 16 if end_bit >= 16 else end_bit
+                    for i in range(start_bit, end):  # Iterate over each bit position (0 to 15)
+                        # Check if the i-th bit is set
+                        if (val >> i) & 1:
+                            flag_index = "b"+str(i+offset)
+                            if flag_index in self.codes[entry.documented_name+'_codes']:
+                                flags.append(self.codes[entry.documented_name+'_codes'][flag_index])
+
 
                 value = ",".join(flags)
             else:
                 flags : list[str] = []
-                for i in range(start_bit, 16):  # Iterate over each bit position (0 to 15)
-                    # Check if the i-th bit is set
-                    if (val >> i) & 1:
-                        flags.append("1")
-                    else:
-                        flags.append("0")
+                if end_bit > 0:
+                    end : int = 16 if end_bit >= 16 else end_bit
+                    for i in range(start_bit, end):  # Iterate over each bit position (0 to 15)
+                        # Check if the i-th bit is set
+                        if (val >> i) & 1:
+                            flags.append("1")
+                        else:
+                            flags.append("0")
+
                 value = ''.join(flags)
+
         elif entry.data_type.value > 200 or entry.data_type == Data_Type.BYTE: #bit types
                 bit_size = Data_Type.getSize(entry.data_type)
                 bit_mask = (1 << bit_size) - 1  # Create a mask for extracting X bits
@@ -823,7 +861,7 @@ def process_register_ushort(self, registry : dict[int, int], entry : registry_ma
 
         return value
 
-    def process_registery(self, registry : dict[int,int] | dict[int,bytes] , map : list[registry_map_entry]) -> dict[str,str]:
+    def process_registery(self, registry : Union[dict[int, int], dict[int, bytes]] , map : list[registry_map_entry]) -> dict[str,str]:
         '''process registry into appropriate datatypes and names -- maybe add func for single entry later?'''
 
         concatenate_registry : dict = {}
@@ -856,6 +894,10 @@ def process_registery(self, registry : dict[int,int] | dict[int,bytes] , map : l
                         concatenated_value = concatenated_value + str(concatenate_registry[key])
                         del concatenate_registry[key]
 
+                    #replace null characters with spaces and trim
+                    if entry.data_type == Data_Type.ASCII:
+                        concatenated_value = concatenated_value.replace("\x00", " ").strip()
+
                     info[entry.variable_name] = concatenated_value
             else:
                 info[entry.variable_name] = value
@@ -871,7 +913,7 @@ def validate_registry_entry(self, entry : registry_map_entry, val) -> int:
                     return 0
 
             if entry.data_type == Data_Type.ASCII:
-                if val and not re.match('[^a-zA-Z0-9\_\-]', val): #validate ascii
+                if val and not re.match(r'[^a-zA-Z0-9\_\-]', val): #validate ascii
                     if entry.value_regex: #regex validation
                         if re.match(entry.value_regex, val):
                             if entry.concatenate:

classes/transports/transport_base.py

@@ -33,8 +33,12 @@ class transport_base:
 
     def __init__(self, settings : 'SectionProxy', protocolSettings : 'protocol_settings' = None) -> None:
 
+        #apply log level to logger
+        self._log_level = logging.getLevelName(settings.get('log_level', fallback='INFO'))
         self._log : logging.Logger = logging.getLogger(__name__)
-        logging.basicConfig(level=logging.DEBUG)
+
+        self._log.setLevel(self._log_level)
+        logging.basicConfig(level=self._log_level)
 
         self.transport_name = settings.name #section name
 

config.cfg.example

@@ -2,6 +2,9 @@
 log_level = DEBUG
 
 [transport.0] #name must be unique, ie: transport.modbus
+#logging level for transport
+log_level = DEBUG
+
 #rs485 / modbus device
 #protocol config files are located in protocols/
 protocol_version = v0.14

defs/common.py

@@ -16,6 +16,10 @@ def strtobool (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
+
     if val and val[0] == 'x':
         return int.from_bytes(bytes.fromhex(val[1:]), byteorder='big')
 

docs/EG4-3000-EHV - MODBUS Communication Protocol.pdf

@@ -96,7 +96,7 @@ class Protocol_Gateway:
     def __init__(self, config_file : str):
         self.__log = logging.getLogger('invertermodbustomqqt_log')
         handler = logging.StreamHandler(sys.stdout)
-        self.__log.setLevel(logging.DEBUG)
+        #self.__log.setLevel(logging.DEBUG)
         formatter = logging.Formatter('[%(asctime)s]  {%(filename)s:%(lineno)d}  %(levelname)s - %(message)s')
         handler.setFormatter(formatter)
         self.__log.addHandler(handler)