From 04260cd2cb6a856ab40cbfb83bc61e491647d5dc Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Tue, 3 Feb 2026 12:30:14 +0100 Subject: [PATCH 1/2] binary payload builder --- .../sample_modbus/sample_modbus/bat.py | 2 +- packages/modules/common/evse.py | 6 +- packages/modules/common/modbus.py | 57 +++++++++++++------ .../devices/alpha_ess/alpha_ess/bat.py | 10 ++-- .../modules/devices/good_we/good_we/bat.py | 16 +++--- .../devices/kostal/kostal_plenticore/bat.py | 17 ++---- .../sma/sma_sunny_boy/bat_smart_energy.py | 25 +------- .../devices/solaredge/solaredge/bat.py | 26 +-------- .../modules/devices/sungrow/sungrow/bat.py | 14 ++--- .../modules/devices/victron/victron/bat.py | 24 ++++---- runs/evse_write_modbus.py | 2 +- 11 files changed, 86 insertions(+), 113 deletions(-) diff --git a/docs/samples/sample_modbus/sample_modbus/bat.py b/docs/samples/sample_modbus/sample_modbus/bat.py index 33dfbe574e..180fb74ef7 100644 --- a/docs/samples/sample_modbus/sample_modbus/bat.py +++ b/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 diff --git a/packages/modules/common/evse.py b/packages/modules/common/evse.py index 86d0d6efcd..11246df1aa 100644 --- a/packages/modules/common/evse.py +++ b/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) diff --git a/packages/modules/common/modbus.py b/packages/modules/common/modbus.py index 3ef4bdd7bb..5c9c3fbe82 100644 --- a/packages/modules/common/modbus.py +++ b/packages/modules/common/modbus.py @@ -13,7 +13,7 @@ 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 @@ -21,20 +21,21 @@ 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_8bit_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_8bit_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,32 @@ 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 in [ModbusDataType.FLOAT_16, 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) diff --git a/packages/modules/devices/alpha_ess/alpha_ess/bat.py b/packages/modules/devices/alpha_ess/alpha_ess/bat.py index 461f6b1f56..c7c062ba7c 100644 --- a/packages/modules/devices/alpha_ess/alpha_ess/bat.py +++ b/packages/modules/devices/alpha_ess/alpha_ess/bat.py @@ -65,7 +65,7 @@ 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) @@ -73,15 +73,15 @@ def set_power_limit(self, power_limit: Optional[int]) -> None: # 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: diff --git a/packages/modules/devices/good_we/good_we/bat.py b/packages/modules/devices/good_we/good_we/bat.py index dc5feb91e1..b23c16d6d7 100644 --- a/packages/modules/devices/good_we/good_we/bat.py +++ b/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 diff --git a/packages/modules/devices/kostal/kostal_plenticore/bat.py b/packages/modules/devices/kostal/kostal_plenticore/bat.py index 1f3207796c..3e6180e3e9 100644 --- a/packages/modules/devices/kostal/kostal_plenticore/bat.py +++ b/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 diff --git a/packages/modules/devices/sma/sma_sunny_boy/bat_smart_energy.py b/packages/modules/devices/sma/sma_sunny_boy/bat_smart_energy.py index f5e5ab7394..ee871c019a 100644 --- a/packages/modules/devices/sma/sma_sunny_boy/bat_smart_energy.py +++ b/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 diff --git a/packages/modules/devices/solaredge/solaredge/bat.py b/packages/modules/devices/solaredge/solaredge/bat.py index cb81757c2a..af10a840b5 100644 --- a/packages/modules/devices/solaredge/solaredge/bat.py +++ b/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 diff --git a/packages/modules/devices/sungrow/sungrow/bat.py b/packages/modules/devices/sungrow/sungrow/bat.py index ef99e62237..bf567720b6 100644 --- a/packages/modules/devices/sungrow/sungrow/bat.py +++ b/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 diff --git a/packages/modules/devices/victron/victron/bat.py b/packages/modules/devices/victron/victron/bat.py index 3cfdf58048..2465081473 100644 --- a/packages/modules/devices/victron/victron/bat.py +++ b/packages/modules/devices/victron/victron/bat.py @@ -66,21 +66,21 @@ def set_power_limit(self, power_limit: Optional[int]) -> None: log.debug("Keine Batteriesteuerung, Selbstregelung durch Wechselrichter") if self.last_mode is not None: # ESS Mode 1 für Selbstregelung mit Phasenkompensation setzen - self.__tcp_client.write_registers(2902, [1], data_type=ModbusDataType.UINT_16, unit=modbus_id) - self.__tcp_client.write_registers(39, [0], data_type=ModbusDataType.UINT_16, unit=vebus_id) + self.__tcp_client.write_register(2902, 1, data_type=ModbusDataType.UINT_16, unit=modbus_id) + self.__tcp_client.write_register(39, 0, data_type=ModbusDataType.UINT_16, unit=vebus_id) 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': # ESS Mode 3 für externe Steuerung und keine Entladung - self.__tcp_client.write_registers(2902, [3], data_type=ModbusDataType.UINT_16, unit=modbus_id) - self.__tcp_client.write_registers(39, [1], data_type=ModbusDataType.UINT_16, unit=vebus_id) + self.__tcp_client.write_register(2902, 3, data_type=ModbusDataType.UINT_16, unit=modbus_id) + self.__tcp_client.write_register(39, 1, data_type=ModbusDataType.UINT_16, unit=vebus_id) self.last_mode = 'stop' elif power_limit < 0: if self.last_mode != 'discharge': # ESS Mode 3 für externe Steuerung und auf L1 wird entladen - self.__tcp_client.write_registers(2902, [3], data_type=ModbusDataType.UINT_16, unit=modbus_id) - self.__tcp_client.write_registers(39, [0], data_type=ModbusDataType.UINT_16, unit=vebus_id) + self.__tcp_client.write_register(2902, 3, data_type=ModbusDataType.UINT_16, unit=modbus_id) + self.__tcp_client.write_register(39, 0, data_type=ModbusDataType.UINT_16, unit=vebus_id) self.last_mode = 'discharge' # Die maximale Entladeleistung begrenzen auf 5000W if phases == 3: @@ -88,13 +88,13 @@ def set_power_limit(self, power_limit: Optional[int]) -> None: power_value = int(min(power_limit, 5000)) log.debug(f"Aktive Batteriesteuerung. Victron mit {phases} Phase(n). " f"Batterie wird mit {power_value} W pro Phase entladen.") - self.__tcp_client.write_registers( - 37, [power_value & 0xFFFF], data_type=ModbusDataType.INT_16, unit=vebus_id) + self.__tcp_client.write_register( + 37, power_value & 0xFFFF, data_type=ModbusDataType.INT_16, unit=vebus_id) if phases == 3: - self.__tcp_client.write_registers( - 40, [power_value & 0xFFFF], data_type=ModbusDataType.INT_16, unit=vebus_id) - self.__tcp_client.write_registers( - 41, [power_value & 0xFFFF], data_type=ModbusDataType.INT_16, unit=vebus_id) + self.__tcp_client.write_register( + 40, power_value & 0xFFFF, data_type=ModbusDataType.INT_16, unit=vebus_id) + self.__tcp_client.write_register( + 41, power_value & 0xFFFF, data_type=ModbusDataType.INT_16, unit=vebus_id) def power_limit_controllable(self) -> bool: return True diff --git a/runs/evse_write_modbus.py b/runs/evse_write_modbus.py index e05bd9d5cc..5e1eed4333 100644 --- a/runs/evse_write_modbus.py +++ b/runs/evse_write_modbus.py @@ -12,4 +12,4 @@ value = int(sys.argv[3]) client, evse_ids = get_modbus_client(local_chargepoint_num) -client.write_registers(register, value, unit=evse_ids[0]) +client.write_register(register, value, unit=evse_ids[0]) From 7fd2e6759f8766d4825b4c99b154ed14890955b6 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Tue, 3 Feb 2026 12:34:33 +0100 Subject: [PATCH 2/2] fix --- packages/modules/common/modbus.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/modules/common/modbus.py b/packages/modules/common/modbus.py index 5c9c3fbe82..9bdf3e73dd 100644 --- a/packages/modules/common/modbus.py +++ b/packages/modules/common/modbus.py @@ -21,11 +21,11 @@ class ModbusDataType(Enum): - UINT_8 = 8, "add_8bit_uint", "decode_8bit_uint" + 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_8bit_int", "decode_8bit_int" + 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" @@ -193,7 +193,12 @@ def _build_binary_payload(self, byteorder: Endian = Endian.Big, wordorder: Endian = Endian.Big) -> list: builder = BinaryPayloadBuilder(byteorder=byteorder, wordorder=wordorder) - if data_type in [ModbusDataType.FLOAT_16, ModbusDataType.FLOAT_32, ModbusDataType.FLOAT_64]: + 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))