binary payload builder

docs/samples/sample_modbus/sample_modbus/bat.py

@@ -76,7 +76,7 @@ def set_power_limit(self, power_limit: Optional[int]) -> None:
         # Wenn der Speicher die Steuerung der Ladeleistung unterstützt, muss bei Übergabe einer Zahl auf aktive
         # Speichersteurung umgeschaltet werden, sodass der Speicher mit der übergebenen Leistung lädt/entlädt. Wird
         # None übergeben, muss der Speicher die Null-Punkt-Ausregelung selbst übernehmen.
-        self.client.write_registers(reg, power_limit)
+        self.client.write_register(reg, power_limit)
         # Wenn der Speicher keine Steuerung der Ladeleistung unterstützt
         pass
 

packages/modules/common/evse.py

@@ -95,7 +95,7 @@ def activate_precise_current(self) -> None:
         else:
             with ModifyLoglevelContext(log, logging.DEBUG):
                 log.debug("Bit zur Angabe der Ströme in 0,1A-Schritten wird gesetzt.")
-            self.client.write_registers(2005, value ^ self.PRECISE_CURRENT_BIT, unit=self.id)
+            self.client.write_register(2005, value ^ self.PRECISE_CURRENT_BIT, unit=self.id)
             # Zeit zum Verarbeiten geben
             time.sleep(1)
 
@@ -105,7 +105,7 @@ def deactivate_precise_current(self) -> None:
         if value & self.PRECISE_CURRENT_BIT:
             with ModifyLoglevelContext(log, logging.DEBUG):
                 log.debug("Bit zur Angabe der Ströme in 0,1A-Schritten wird zurueckgesetzt.")
-            self.client.write_registers(2005, value ^ self.PRECISE_CURRENT_BIT, unit=self.id)
+            self.client.write_register(2005, value ^ self.PRECISE_CURRENT_BIT, unit=self.id)
         else:
             return
 
@@ -118,4 +118,4 @@ def set_current(self, current: int, phases_in_use: Optional[int] = None) -> None
             if formatted_current > 16 and phases_in_use > 1:
                 formatted_current = 16
         if self.evse_current != formatted_current:
-            self.client.write_registers(1000, formatted_current, unit=self.id)
+            self.client.write_register(1000, formatted_current, unit=self.id)

packages/modules/common/modbus.py

@@ -13,28 +13,29 @@
 import pymodbus
 from pymodbus.client.sync import ModbusTcpClient, ModbusSerialClient
 from pymodbus.constants import Endian
-from pymodbus.payload import BinaryPayloadDecoder
+from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
 from pymodbus.transaction import ModbusSocketFramer
 from urllib3.util import parse_url
 
 log = logging.getLogger(__name__)
 
 
 class ModbusDataType(Enum):
-    UINT_8 = 8, "decode_8bit_uint"
-    UINT_16 = 16, "decode_16bit_uint"
-    UINT_32 = 32, "decode_32bit_uint"
-    UINT_64 = 64, "decode_64bit_uint"
-    INT_8 = 8, "decode_8bit_int"
-    INT_16 = 16, "decode_16bit_int"
-    INT_32 = 32, "decode_32bit_int"
-    INT_64 = 64, "decode_64bit_int"
-    FLOAT_16 = 16, "decode_16bit_float"
-    FLOAT_32 = 32, "decode_32bit_float"
-    FLOAT_64 = 64, "decode_64bit_float"
-
-    def __init__(self, bits: int, decoding_method: str):
+    UINT_8 = 8, "add_16bit_uint", "decode_8bit_uint"
+    UINT_16 = 16, "add_16bit_uint", "decode_16bit_uint"
+    UINT_32 = 32, "add_32bit_uint", "decode_32bit_uint"
+    UINT_64 = 64, "add_64bit_uint", "decode_64bit_uint"
+    INT_8 = 8, "add_16bit_int", "decode_8bit_int"
+    INT_16 = 16, "add_16bit_int", "decode_16bit_int"
+    INT_32 = 32, "add_32bit_int", "decode_32bit_int"
+    INT_64 = 64, "add_64bit_int", "decode_64bit_int"
+    FLOAT_16 = 16, "add_16bit_float", "decode_16bit_float"
+    FLOAT_32 = 32, "add_32bit_float", "decode_32bit_float"
+    FLOAT_64 = 64, "add_64bit_float", "decode_64bit_float"
+
+    def __init__(self, bits: int, encoding_method: str, decoding_method: str):
         self.bits = bits
+        self.encoding_method = encoding_method
         self.decoding_method = decoding_method
 
 
@@ -186,8 +187,37 @@ def read_coils(self, address: int, count: int, **kwargs):
             e.args += (NO_VALUES.format(self.address, self.port),)
             raise e
 
-    def write_registers(self, address: int, value: Any, **kwargs):
-        self._delegate.write_registers(address, value, **kwargs)
+    def _build_binary_payload(self,
+                              value: Union[int, float],
+                              data_type: ModbusDataType,
+                              byteorder: Endian = Endian.Big,
+                              wordorder: Endian = Endian.Big) -> list:
+        builder = BinaryPayloadBuilder(byteorder=byteorder, wordorder=wordorder)
+        if data_type == ModbusDataType.FLOAT_16:
+            # FLOAT_16 (IEEE 754 Half-Precision) manuelle Konvertierung
+            packed = struct.pack(">e", float(value))
+            uint16_value = struct.unpack(">H", packed)[0]
+            builder.add_16bit_uint(uint16_value)
+        elif data_type in [ModbusDataType.FLOAT_32, ModbusDataType.FLOAT_64]:
+            getattr(builder, data_type.encoding_method)(float(value))
+        else:
+            getattr(builder, data_type.encoding_method)(int(value))
+        return builder.to_registers()
+
+    def write_register(self, address: int, value: Union[int, float], data_type: Optional[ModbusDataType] = None,
+                       byteorder: Endian = Endian.Big, wordorder: Endian = Endian.Big, **kwargs):
+        if data_type is not None:
+            if data_type.bits > 16 or data_type in [ModbusDataType.FLOAT_16,
+                                                    ModbusDataType.FLOAT_32,
+                                                    ModbusDataType.FLOAT_64]:
+                registers = self._build_binary_payload(value, data_type, byteorder, wordorder)
+                self._delegate.write_registers(address, registers, **kwargs)
+            else:
+                # Einfache 16-bit oder kleinere Werte können direkt geschrieben werden
+                self._delegate.write_registers(address, [value], **kwargs)
+        else:
+            # Fallback für bestehenden Code ohne data_type
+            self._delegate.write_registers(address, value, **kwargs)
 
     def write_single_coil(self, address: int, value: Any, **kwargs):
         self._delegate.write_coil(address, value, **kwargs)

packages/modules/devices/alpha_ess/alpha_ess/bat.py

@@ -65,23 +65,23 @@ def set_power_limit(self, power_limit: Optional[int]) -> None:
             # Kein Powerlimit gefordert, externe Steuerung deaktivieren
             log.debug("Keine Batteriesteuerung gefordert, deaktiviere externe Steuerung.")
             if self.last_mode is not None:
-                self.__tcp_client.write_registers(2127, [0], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(2127, 0, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = None
         elif power_limit <= 0:
             # AlphaESS kann die Entladung nur über den SoC verhindern (komplette Entladesperre)
             # Netzladung mit geringen Ziel SoC verhindert auch Entladung (Default 10%)
             # Zeiten für Netzladung müssen im Wechselrichter aktiviert werden
             log.debug("Aktive Batteriesteuerung angestoßen. Setze Entladesperre.")
             if self.last_mode != 'stop':
-                self.__tcp_client.write_registers(2127, [1], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(2133, [10], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(2127, 1, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(2133, 10, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'stop'
         else:
             # Aktive Ladung
             log.debug("Aktive Batteriesteuerung angestoßen. Setze aktive Ladung.")
             if self.last_mode != 'charge':
-                self.__tcp_client.write_registers(2127, [1], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(2133, [100], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(2127, 1, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(2133, 100, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'charge'
 
     def power_limit_controllable(self) -> bool:

packages/modules/devices/good_we/good_we/bat.py

@@ -75,31 +75,31 @@ def set_power_limit(self, power_limit: Optional[int]) -> None:
         if power_limit is None:
             log.debug("Keine Batteriesteuerung, Selbstregelung durch Wechselrichter")
             if self.last_mode is not None:
-                self.__tcp_client.write_registers(47511, [1], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(47512, [0], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(47511, 1, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(47512, 0, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = None
         elif power_limit == 0:
             log.debug("Aktive Batteriesteuerung. Batterie wird auf Stop gesetzt und nicht entladen")
             if self.last_mode != 'stop':
-                self.__tcp_client.write_registers(47511, [2], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(47512, [0], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(47511, 2, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(47512, 0, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'stop'
         elif power_limit < 0:
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_limit} W entladen für den Hausverbrauch")
             if self.last_mode != 'discharge':
-                self.__tcp_client.write_registers(47511, [3], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(47511, 3, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'discharge'
             # Die maximale Entladeleistung begrenzen auf 5000W, maximaler Wertebereich Modbusregister.
             power_value = int(min(abs(power_limit), 10000))
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_value} W entladen für den Hausverbrauch")
-            self.__tcp_client.write_registers(47512, [power_value], data_type=ModbusDataType.UINT_16, unit=unit)
+            self.__tcp_client.write_register(47512, power_value, data_type=ModbusDataType.UINT_16, unit=unit)
         elif power_limit > 0:
             if self.last_mode != 'charge':
-                self.__tcp_client.write_registers(47511, [2], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(47511, 2, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'charge'
             power_value = int(min(abs(power_limit), 10000))
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_value} W entladen für den Hausverbrauch")
-            self.__tcp_client.write_registers(47512, [power_value], data_type=ModbusDataType.UINT_16, unit=unit)
+            self.__tcp_client.write_register(47512, power_value, data_type=ModbusDataType.UINT_16, unit=unit)
 
     def power_limit_controllable(self) -> bool:
         return True

packages/modules/devices/kostal/kostal_plenticore/bat.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 import logging
-import struct
 from typing import TypedDict, Any, Optional
 from pymodbus.constants import Endian
 
@@ -69,23 +68,15 @@ def set_power_limit(self, power_limit: Optional[int]) -> None:
         elif power_limit == 0:
             # wiederholt auf Stop setzen damit sich Register nicht zurücksetzt
             log.debug("Aktive Batteriesteuerung. Batterie wird auf Stop gesetzt und nicht entladen")
-            packed = struct.pack('>f', 0.0)
-            if self.endianess == Endian.Little:
-                registers = [int.from_bytes(packed[2:4], 'little'), int.from_bytes(packed[0:2], 'little')]
-            else:
-                registers = [int.from_bytes(packed[0:2], 'big'), int.from_bytes(packed[2:4], 'big')]
-            self.client.write_registers(1034, registers, data_type=ModbusDataType.FLOAT_32, unit=unit)
+            self.client.write_register(1034, 0.0, data_type=ModbusDataType.FLOAT_32,
+                                       wordorder=self.endianess, unit=unit)
         elif power_limit < 0:
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_limit} W entladen für den Hausverbrauch")
             # Die maximale Entladeleistung begrenzen auf 7000W
             power_value = float(min(abs(power_limit), 7000)) * -1
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_value} W entladen für den Hausverbrauch")
-            packed = struct.pack('>f', power_value)
-            if self.endianess == Endian.Little:
-                registers = [int.from_bytes(packed[2:4], 'little'), int.from_bytes(packed[0:2], 'little')]
-            else:
-                registers = [int.from_bytes(packed[0:2], 'big'), int.from_bytes(packed[2:4], 'big')]
-            self.client.write_registers(1034, registers, data_type=ModbusDataType.FLOAT_32, unit=unit)
+            self.client.write_register(1034, power_value, data_type=ModbusDataType.FLOAT_32,
+                                       wordorder=self.endianess, unit=unit)
 
     def power_limit_controllable(self) -> bool:
         return True

packages/modules/devices/sma/sma_sunny_boy/bat_smart_energy.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-import pymodbus
 from typing import TypedDict, Any, Dict, Union, Optional
 import logging
 
@@ -130,28 +129,8 @@ def _read_registers(self, register_names: list, unit: int) -> Dict[str, Union[in
     def _write_registers(self, values_to_write: Dict[str, Union[int, float]], unit: int) -> None:
         for key, value in values_to_write.items():
             address, data_type = self.REGISTERS[key]
-            encoded_value = self._encode_value(value, data_type)
-            self.__tcp_client.write_registers(address, encoded_value, unit=unit)
-            log.debug(f"Neuer Wert {encoded_value} in Register {address} geschrieben.")
-
-    def _encode_value(self, value: Union[int, float], data_type: ModbusDataType) -> list:
-        builder = pymodbus.payload.BinaryPayloadBuilder(
-            byteorder=pymodbus.constants.Endian.Big,
-            wordorder=pymodbus.constants.Endian.Big
-        )
-        encode_methods = {
-            ModbusDataType.UINT_32: builder.add_32bit_uint,
-            ModbusDataType.INT_32: builder.add_32bit_int,
-            ModbusDataType.UINT_16: builder.add_16bit_uint,
-            ModbusDataType.INT_16: builder.add_16bit_int,
-        }
-
-        if data_type in encode_methods:
-            encode_methods[data_type](int(value))
-        else:
-            raise ValueError(f"Unsupported data type: {data_type}")
-
-        return builder.to_registers()
+            self.__tcp_client.write_register(address, value, data_type, unit=unit)
+            log.debug(f"Neuer Wert {value} in Register {address} geschrieben.")
 
     def power_limit_controllable(self) -> bool:
         return True

packages/modules/devices/solaredge/solaredge/bat.py

@@ -235,35 +235,13 @@ def _read_registers(self, register_names: list, unit: int) -> Dict[str, Union[in
     def _write_registers(self, values_to_write: Dict[str, Union[int, float]], unit: int) -> None:
         for key, value in values_to_write.items():
             address, data_type = self.REGISTERS[key]
-            encoded_value = self._encode_value(value, data_type)
             try:
-                self.__tcp_client.write_registers(address, encoded_value, unit=unit)
-                log.debug(f"Neuer Wert {encoded_value} in Register {address} geschrieben.")
+                self.__tcp_client.write_register(address, value, data_type, wordorder=Endian.Little, unit=unit)
+                log.debug(f"Neuer Wert {value} in Register {address} geschrieben.")
             except pymodbus.exceptions.ModbusException as e:
                 log.error(f"Failed to write register {key} at address {address}: {e}")
                 self.fault_state.error(f"Modbus write error: {e}")
 
-    def _encode_value(self, value: Union[int, float], data_type: ModbusDataType) -> list:
-        builder = pymodbus.payload.BinaryPayloadBuilder(
-            byteorder=pymodbus.constants.Endian.Big,
-            wordorder=pymodbus.constants.Endian.Little
-        )
-        encode_methods = {
-            ModbusDataType.UINT_32: builder.add_32bit_uint,
-            ModbusDataType.INT_32: builder.add_32bit_int,
-            ModbusDataType.UINT_16: builder.add_16bit_uint,
-            ModbusDataType.INT_16: builder.add_16bit_int,
-            ModbusDataType.FLOAT_32: builder.add_32bit_float,
-        }
-        if data_type in encode_methods:
-            if data_type == ModbusDataType.FLOAT_32:
-                encode_methods[data_type](float(value))
-            else:
-                encode_methods[data_type](int(value))
-        else:
-            raise ValueError(f"Unsupported data type: {data_type}")
-        return builder.to_registers()
-
     def power_limit_controllable(self) -> bool:
         return True
 

packages/modules/devices/sungrow/sungrow/bat.py

@@ -121,25 +121,25 @@ def set_power_limit(self, power_limit: Optional[int]) -> None:
         if power_limit is None:
             log.debug("Keine Batteriesteuerung, Selbstregelung durch Wechselrichter")
             if self.last_mode is not None:
-                self.__tcp_client.write_registers(13049, [0], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(13050, [0xCC], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(13049, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(13050, 0xCC, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = None
         elif power_limit == 0:
             log.debug("Aktive Batteriesteuerung. Batterie wird auf Stop gesetzt und nicht entladen")
             if self.last_mode != 'stop':
-                self.__tcp_client.write_registers(13049, [2], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(13050, [0xCC], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(13049, 2, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(13050, 0xCC, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'stop'
         elif power_limit < 0:
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_limit} W entladen für den Hausverbrauch")
             if self.last_mode != 'discharge':
-                self.__tcp_client.write_registers(13049, [2], data_type=ModbusDataType.UINT_16, unit=unit)
-                self.__tcp_client.write_registers(13050, [0xBB], data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(13049, 2, data_type=ModbusDataType.UINT_16, unit=unit)
+                self.__tcp_client.write_register(13050, 0xBB, data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = 'discharge'
             # Die maximale Entladeleistung begrenzen auf 5000W, maximaler Wertebereich Modbusregister.
             power_value = int(min(abs(power_limit), 5000))
             log.debug(f"Aktive Batteriesteuerung. Batterie wird mit {power_value} W entladen für den Hausverbrauch")
-            self.__tcp_client.write_registers(13051, [power_value], data_type=ModbusDataType.UINT_16, unit=unit)
+            self.__tcp_client.write_register(13051, power_value, data_type=ModbusDataType.UINT_16, unit=unit)
 
     def power_limit_controllable(self) -> bool:
         return True