Merge branch 'v1.1.10-pr92' into main

Author: HotNoob

.github/workflows/python-3.10.yml

@@ -3,6 +3,9 @@
 
 name: Python 3.10
 
+permissions:
+  contents: read
+  
 on:
   push:
     branches: [ "main" ]

.github/workflows/python-3.11.yml

@@ -3,6 +3,9 @@
 
 name: Python 3.11
 
+permissions:
+  contents: read
+
 on:
   push:
     branches: [ "main" ]

.github/workflows/python-3.12.yml

@@ -3,6 +3,9 @@
 
 name: Python 3.12
 
+permissions:
+  contents: read
+
 on:
   push:
     branches: [ "main" ]

.github/workflows/python-3.13.yml

@@ -3,6 +3,9 @@
 
 name: Python 3.13
 
+permissions:
+  contents: read
+
 on:
   push:
     branches: [ "main" ]

.github/workflows/python-3.9.yml

@@ -3,6 +3,9 @@
 
 name: Python 3.9
 
+permissions:
+  contents: read
+
 on:
   push:
     branches: [ "main" ]

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.9-alpine as base
+FROM python:3.13-alpine as base
 FROM base as builder
 RUN mkdir /install
 WORKDIR /install

README.md

@@ -58,7 +58,7 @@ solark_v1.1 = SolarArk 8/12K Inverters - Untested
 hdhk_16ch_ac_module = some chinese current monitoring device :P
 srne_2021_v1.96 = SRNE inverters 2021+ (tested at ASF48100S200-H, ok-ish for HF2430U60-100 )
 
-eg4_v58 = eg4 inverters ( EG4-6000XP ) - confirmed working
+eg4_v58 = eg4 inverters ( EG4-6000XP, EG4-18K ) - confirmed working
 eg4_3000ehv_v1 = eg4 inverters ( EG4_3000EHV )
 ```
 
@@ -154,11 +154,16 @@ if you installed this when it was called growatt2mqtt-hotnoob or InverterModBusT
 
 ### donate
 this took me a while to make; and i had to make it because there werent any working solutions. 
-donations would be appreciated.
-![BitCoin Donation](https://github.com/HotNoob/growatt2mqtt-hotnoob/blob/main/images/donate_to_hotnoob.png?raw=true)
+donations / sponsoring this repo would be appreciated.
 
-```(btc) bc1qh394vazcguedkw2rlklnuhapdq7qgpnnz9c3t0```
-
-### Use Docker - untested
+### Use Docker
 - ```docker build . -t protocol_gateway ```
 - ```docker run --device=/dev/ttyUSB0 protocol_gateway```
+
+### Use Docker Image
+- ``` docker pull hotn00b/pythonprotocolgateway ``` 
+- ```docker run -v $(pwd)/config.cfg:/app/config.cfg --device=/dev/ttyUSB0 hotn00b/pythonprotocolgateway```
+
+See [config.cfg.example](https://github.com/HotNoob/PythonProtocolGateway/blob/main/config.cfg.example)
+
+[Docker Image Repo](https://hub.docker.com/r/hotn00b/pythonprotocolgateway)

classes/protocol_settings.py

@@ -188,6 +188,7 @@ class registry_map_entry:
     register_bit : int
     register_byte : int
     ''' byte offset for canbus ect... '''
+
     variable_name : str
     documented_name : str
     unit : str
@@ -208,6 +209,9 @@ class registry_map_entry:
     data_type_size : int = -1
     ''' for non-fixed size types like ASCII'''
 
+    data_byteorder : str = ''
+    ''' entry specific byte order little | big | '' '''
+
     read_command : bytes = None
     ''' for transports/protocols that require sending a command ontop of "register" '''
 
@@ -512,16 +516,30 @@ def process_row(row):
 
             #region data type
             data_type = Data_Type.USHORT
-
             data_type_len : int = -1
+            data_byteorder : str = ''
             #optional row, only needed for non-default data types
             if "data type" in row and row["data type"]:
+                data_type_str : str = ''
+
                 matches = data_type_regex.search(row["data type"])
                 if matches:
                     data_type_len = int(matches.group("length"))
-                    data_type = Data_Type.fromString(matches.group("datatype"))
+                    data_type_str = matches.group("datatype")
                 else:
-                    data_type = Data_Type.fromString(row["data type"])
+                    data_type_str = row["data type"]
+
+                #check if datatype specifies byteorder 
+                if data_type_str.upper().endswith("_LE"):
+                    data_byteorder = "little"
+                    data_type_str = data_type_str[:-3]
+                elif data_type_str.upper().endswith("_BE"):
+                    data_byteorder = "big"
+                    data_type_str = data_type_str[:-3]
+
+
+                data_type = Data_Type.fromString(data_type_str)
+
 
 
             if "values" not in row:
@@ -658,6 +676,7 @@ def process_row(row):
                                             unit_mod= unit_multiplier,
                                             data_type= data_type,
                                             data_type_size = data_type_len,
+                                            data_byteorder = data_byteorder,
                                             concatenate = concatenate,
                                             concatenate_registers = concatenate_registers,
                                             values=values,
@@ -857,6 +876,10 @@ 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'''
 
+        byte_order : str = self.byteorder
+        if entry.data_byteorder: #allow map entry to override byteorder
+            byte_order = entry.data_byteorder
+
         if isinstance(registry[entry.register], tuple):
             register = registry[entry.register][0] #can bus uses tuple for timestamp
         else:
@@ -869,14 +892,15 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
             register = register[:entry.data_type_size]
 
         if entry.data_type == Data_Type.UINT:
-            value = int.from_bytes(register[:4], byteorder=self.byteorder, signed=False)
+            value = int.from_bytes(register[:4], byteorder=byte_order, signed=False)
         elif entry.data_type == Data_Type.INT:
-            value = int.from_bytes(register[:4], byteorder=self.byteorder, signed=True)
+            value = int.from_bytes(register[:4], byteorder=byte_order, signed=True)
         elif entry.data_type == Data_Type.USHORT:
-            value = int.from_bytes(register[:2], byteorder=self.byteorder, signed=False)
+            value = int.from_bytes(register[:2], byteorder=byte_order, signed=False)
         elif entry.data_type == Data_Type.SHORT:
-            value = int.from_bytes(register[:2], byteorder=self.byteorder, signed=True)
+            value = int.from_bytes(register[:2], byteorder=byte_order, 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:
+            val = int.from_bytes(register, byteorder=byte_order, signed=False)
             #16 bit flags
             start_bit : int = 0
             end_bit : int = 16 #default 16 bit
@@ -952,11 +976,20 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
                 # If positive, simply extract the value using the bit mask
                 value = (register >> bit_index) & bit_mask
 
-        elif entry.data_type.value > 200 or entry.data_type == Data_Type.BYTE: #bit types
+        elif entry.data_type == Data_Type.BYTE: #bit types
+            value = int.from_bytes(register[:1], byteorder=byte_order, signed=False)
+        elif entry.data_type.value > 200: #bit types
             bit_size = Data_Type.getSize(entry.data_type)
             bit_mask = (1 << bit_size) - 1  # Create a mask for extracting X bits
             bit_index = entry.register_bit
+
+
+            if isinstance(register, bytes):
+                register = int.from_bytes(register, byteorder=byte_order)
+                            
             value = (register >> bit_index) & bit_mask
+
+
         elif entry.data_type == Data_Type.HEX:
             value = register.hex() #convert bytes to hex
         elif entry.data_type == Data_Type.ASCII:
@@ -986,6 +1019,11 @@ def process_register_bytes(self, registry : dict[int,bytes], entry : registry_ma
 
     def process_register_ushort(self, registry : dict[int, int], entry : registry_map_entry ):
         ''' process ushort type registry into data'''
+
+        byte_order : str = self.byteorder
+        if entry.data_byteorder:
+            byte_order = entry.data_byteorder
+
         if entry.data_type == Data_Type.UINT: #read uint
             if entry.register + 1 not in registry:
                 return
@@ -1074,10 +1112,10 @@ 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.HEX:
-                value = registry[entry.register].to_bytes((16 + 7) // 8, byteorder=self.byteorder) #convert to ushort to bytes
+                value = registry[entry.register].to_bytes((16 + 7) // 8, byteorder=byte_order) #convert to ushort to bytes
                 value = value.hex() #convert bytes to hex
         elif entry.data_type == Data_Type.ASCII:
-            value = registry[entry.register].to_bytes((16 + 7) // 8, byteorder=self.byteorder) #convert to ushort to bytes
+            value = registry[entry.register].to_bytes((16 + 7) // 8, byteorder=byte_order) #convert to ushort to bytes
             try:
                 value = value.decode("utf-8") #convert bytes to ascii
             except UnicodeDecodeError as e:

classes/transports/transport_base.py

@@ -98,7 +98,7 @@ def __init__(self, settings : "SectionProxy") -> None:
             self.device_name = settings.get(["device_name", "name"], fallback=self.device_manufacturer+"_"+self.device_serial_number)
             self.bridge = settings.get("bridge", self.bridge)
             self.read_interval = settings.getfloat("read_interval", self.read_interval)
-            self.max_precision = settings.getint(["max_precision", "precision"], self.max_precision)
+            self.max_precision = settings.getint(["max_precision", "precision"], fallback=self.max_precision)
             if "write_enabled" in settings or "enable_write" in settings:
                 self.write_enabled = settings.getboolean(["write_enabled", "enable_write"], self.write_enabled)
 

defs/common.py

@@ -1,4 +1,5 @@
 import re
+import os
 
 import serial.tools.list_ports
 
@@ -46,13 +47,23 @@ def strtoint(val : str) -> int:
     return int(val)
 
 def get_usb_serial_port_info(port : str = "") -> str:
+
+    # If port is a symlink
+    if os.path.islink(port):
+        port = os.path.realpath(port)
+
     for p in serial.tools.list_ports.comports():
         if str(p.device).upper() == port.upper():
             return "["+hex(p.vid)+":"+hex(p.pid)+":"+str(p.serial_number)+":"+str(p.location)+"]"
 
     return ""
 
 def find_usb_serial_port(port : str =  "", vendor_id : str = "", product_id : str = "", serial_number : str = "", location : str = "") -> str:
+
+    # If port is a symlink
+    if os.path.islink(port):
+        port = os.path.realpath(port)
+
     if not port.startswith("["):
         return port