From 62f5954b27e3ab8bca62a59f2920f717641cc500 Mon Sep 17 00:00:00 2001 From: seaspotter Date: Sun, 6 Sep 2026 09:43:20 +0200 Subject: [PATCH 1/2] Fix EVSE readout retry gap and unsynchronized Modbus access --- packages/modules/common/hardware_check.py | 3 +- .../modules/common/hardware_check_test.py | 26 +++ packages/modules/common/modbus.py | 212 ++++++++++-------- packages/modules/common/modbus_test.py | 55 +++++ 4 files changed, 195 insertions(+), 101 deletions(-) create mode 100644 packages/modules/common/modbus_test.py diff --git a/packages/modules/common/hardware_check.py b/packages/modules/common/hardware_check.py index ca7cf79fd5..59c0a8c616 100644 --- a/packages/modules/common/hardware_check.py +++ b/packages/modules/common/hardware_check.py @@ -89,8 +89,7 @@ def request_and_check_hardware(self: ClientHandlerProtocol, evse_state = self.evse_client.get_evse_state() evse_check_passed = True break - except (pymodbus.exceptions.ModbusIOException, - pymodbus.exceptions.ConnectionException) as e: + except Exception as e: evse_check_passed = self.handle_exception(e) # nur warten, wenn danach noch ein Versuch folgt if attempt < MAX_ATTEMPTS - 1 and evse_check_passed is False: diff --git a/packages/modules/common/hardware_check_test.py b/packages/modules/common/hardware_check_test.py index 931db36d8d..6da1ae58b1 100644 --- a/packages/modules/common/hardware_check_test.py +++ b/packages/modules/common/hardware_check_test.py @@ -66,6 +66,32 @@ def test_hardware_check_fails(evse_side_effect, ClientHandler(0, client, [1], Mock()) +@pytest.mark.parametrize( + "transient_exception", + [pytest.param(Exception("Modbus"), id="response.isError() in modbus.py"), + pytest.param(ValueError("Unbekannter Zustand der EVSE"), id="EvseStatusCode.FAILURE in evse.py")] +) +def test_hardware_check_retries_transient_exception(transient_exception, monkeypatch): + # Neither modbus.py's response.isError() case nor evse.py's EvseStatusCode.FAILURE case raise + # ModbusIOException/ConnectionException. The retry loop must catch those too, or a single + # transient glitch fails the whole readout with no retry. + mock_evse_client = Mock(spec=Evse, version=18, + get_evse_state=Mock(side_effect=[transient_exception, Mock(spec=EvseState)])) + monkeypatch.setattr(ClientHandler, "_evse_factory", Mock(return_value=mock_evse_client)) + + counter_state_mock = Mock(spec=CounterState, voltages=[230]*3, currents=[0, 0, 0], powers=[0, 0, 0], + power=0, serial_number="1234") + mock_meter_client = Mock(spec=sdm.Sdm630_72, get_counter_state=Mock(return_value=counter_state_mock)) + monkeypatch.setattr(ClientHandler, "find_meter_client", Mock(return_value=mock_meter_client)) + + client = Mock(spec=ModbusSerialClient_, __enter__=Mock(return_value=None), __exit__=Mock(return_value=None)) + + # execution and evaluation + # keine Exception: der zweite Versuch muss noch innerhalb von request_and_check_hardware laufen + ClientHandler(0, client, [1], Mock()) + assert mock_evse_client.get_evse_state.call_count == 2 + + def test_hardware_check_succeeds(monkeypatch): # setup mock_evse_client = Mock(spec=Evse, get_evse_state=Mock(return_value=Mock(spec=EvseState)), version=17) diff --git a/packages/modules/common/modbus.py b/packages/modules/common/modbus.py index e3602a8c6a..a90526f7db 100644 --- a/packages/modules/common/modbus.py +++ b/packages/modules/common/modbus.py @@ -6,6 +6,7 @@ """ import logging import struct +import threading from enum import Enum import time from types import TracebackType @@ -58,32 +59,39 @@ def __init__(self, self.address = address self.port = port self.sleep_after_connect = sleep_after_connect + # Serialisiert alle Zugriffe auf die (ggf. von mehreren Threads geteilte) Verbindung, zB. + # Haupt-Poll-Thread und Phasenumschaltungs-/CP-Unterbrechungs-Thread bei internem Ladepunkt. + self._lock = threading.RLock() def __enter__(self): - try: - self._delegate.__enter__() - time.sleep(self.sleep_after_connect) - except pymodbus.exceptions.ConnectionException as e: - e.args += (NO_CONNECTION.format(self.address, self.port),) - raise e + with self._lock: + try: + self._delegate.__enter__() + time.sleep(self.sleep_after_connect) + except pymodbus.exceptions.ConnectionException as e: + e.args += (NO_CONNECTION.format(self.address, self.port),) + raise e return self def __exit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], exc_traceback: Optional[TracebackType]): - self._delegate.__exit__(exc_type, exc_value, exc_traceback) + with self._lock: + self._delegate.__exit__(exc_type, exc_value, exc_traceback) def connect(self) -> None: - self._delegate.connect() - time.sleep(self.sleep_after_connect) + with self._lock: + self._delegate.connect() + time.sleep(self.sleep_after_connect) def close(self) -> None: - try: - log.debug("Close Modbus TCP connection") - self._delegate.close() - except Exception as e: - raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e + with self._lock: + try: + log.debug("Close Modbus TCP connection") + self._delegate.close() + except Exception as e: + raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e def is_socket_open(self) -> bool: return self._delegate.is_socket_open() @@ -94,37 +102,38 @@ def __read_registers(self, read_register_method: Callable[[int, int], Any], byteorder: str = Endian.Big, wordorder: str = Endian.Big, **kwargs: Any): - if self.is_socket_open() is False: - self.connect() - try: - multi_request = isinstance(types, Iterable) - if not multi_request: - types = [types] - - def divide_rounding_up(numerator: int, denominator: int): - return -(-numerator // denominator) - - number_of_addresses = sum(divide_rounding_up( - t.bits, _MODBUS_HOLDING_REGISTER_SIZE) for t in types) - response = read_register_method( - address, number_of_addresses, **kwargs) - if response.isError(): - raise Exception(__name__+" "+str(response)) - decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder, wordorder) - result = [struct.unpack(">e", struct.pack(">H", decoder.decode_16bit_uint())) if t == - ModbusDataType.FLOAT_16 else getattr(decoder, t.decoding_method)() for t in types] - return result if multi_request else result[0] - except pymodbus.exceptions.ConnectionException as e: - self.close() - e.args += (NO_CONNECTION.format(self.address, self.port),) - raise e - except pymodbus.exceptions.ModbusIOException as e: - self.close() - e.args += (NO_VALUES.format(self.address, self.port),) - raise e - except Exception as e: - self.close() - raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e + with self._lock: + if self.is_socket_open() is False: + self.connect() + try: + multi_request = isinstance(types, Iterable) + if not multi_request: + types = [types] + + def divide_rounding_up(numerator: int, denominator: int): + return -(-numerator // denominator) + + number_of_addresses = sum(divide_rounding_up( + t.bits, _MODBUS_HOLDING_REGISTER_SIZE) for t in types) + response = read_register_method( + address, number_of_addresses, **kwargs) + if response.isError(): + raise Exception(__name__+" "+str(response)) + decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder, wordorder) + result = [struct.unpack(">e", struct.pack(">H", decoder.decode_16bit_uint())) if t == + ModbusDataType.FLOAT_16 else getattr(decoder, t.decoding_method)() for t in types] + return result if multi_request else result[0] + except pymodbus.exceptions.ConnectionException as e: + self.close() + e.args += (NO_CONNECTION.format(self.address, self.port),) + raise e + except pymodbus.exceptions.ModbusIOException as e: + self.close() + e.args += (NO_VALUES.format(self.address, self.port),) + raise e + except Exception as e: + self.close() + raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e @overload def read_holding_registers(self, address: int, types: Iterable[ModbusDataType], byteorder: str = Endian.Big, @@ -177,17 +186,18 @@ def read_coils(self, address: int, count: int, **kwargs: Any) -> Union[bool, Lis pass def read_coils(self, address: int, count: int = 1, **kwargs: Any) -> Union[bool, List[bool]]: - try: - response = self._delegate.read_coils(address, count, **kwargs) - if response.isError(): - raise Exception(__name__+" "+str(response)) - return response.bits[0] if count == 1 else response.bits[:count] - except pymodbus.exceptions.ConnectionException as e: - e.args += (NO_CONNECTION.format(self.address, self.port),) - raise e - except pymodbus.exceptions.ModbusIOException as e: - e.args += (NO_VALUES.format(self.address, self.port),) - raise e + with self._lock: + try: + response = self._delegate.read_coils(address, count, **kwargs) + if response.isError(): + raise Exception(__name__+" "+str(response)) + return response.bits[0] if count == 1 else response.bits[:count] + except pymodbus.exceptions.ConnectionException as e: + e.args += (NO_CONNECTION.format(self.address, self.port),) + raise e + except pymodbus.exceptions.ModbusIOException as e: + e.args += (NO_VALUES.format(self.address, self.port),) + raise e def _build_binary_payload(self, value: Union[int, float], @@ -208,24 +218,27 @@ def _build_binary_payload(self, def write_register(self, address: int, value: Union[int, float], data_type: Optional[ModbusDataType] = None, byteorder: str = Endian.Big, wordorder: str = Endian.Big, **kwargs: Any): - 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) + with self._lock: + 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: - # 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) + # Fallback für bestehenden Code ohne data_type + self._delegate.write_registers(address, value, **kwargs) def write_single_coil(self, address: int, value: bool, **kwargs: Any): - self._delegate.write_coil(address, value, **kwargs) + with self._lock: + self._delegate.write_coil(address, value, **kwargs) def write_coils(self, address: int, value: List[bool], **kwargs: Any): - self._delegate.write_coils(address, value, **kwargs) + with self._lock: + self._delegate.write_coils(address, value, **kwargs) def __read_bulk(self, read_register_method: Callable[[int, int], Any], @@ -239,36 +252,37 @@ def __read_bulk(self, Liest einen Registerbereich und gibt ein dict mit reg als Key und dekodiertem Wert als Value zurück. mapping: Liste von Tupeln (reg, ModbusDataType) """ - if self.is_socket_open() is False: - self.connect() - try: - response = read_register_method(start_address, count, **kwargs) - if response.isError(): - raise Exception(__name__+" "+str(response)) - decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder, wordorder) - results = {} - for register_address, data_type in mapping: - multiple_register_requested = isinstance(data_type, Iterable) - if not multiple_register_requested: - data_type = [data_type] - offset = register_address - start_address - decoder.reset() - decoder.skip_bytes(offset * 2) - val = [struct.unpack(">e", struct.pack(">H", decoder.decode_16bit_uint())) if t == - ModbusDataType.FLOAT_16 else getattr(decoder, t.decoding_method)() for t in data_type] - results[register_address] = val if multiple_register_requested else val[0] - return results - except pymodbus.exceptions.ConnectionException as e: - self.close() - e.args += (NO_CONNECTION.format(self.address, self.port),) - raise e - except pymodbus.exceptions.ModbusIOException as e: - self.close() - e.args += (NO_VALUES.format(self.address, self.port),) - raise e - except Exception as e: - self.close() - raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e + with self._lock: + if self.is_socket_open() is False: + self.connect() + try: + response = read_register_method(start_address, count, **kwargs) + if response.isError(): + raise Exception(__name__+" "+str(response)) + decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder, wordorder) + results = {} + for register_address, data_type in mapping: + multiple_register_requested = isinstance(data_type, Iterable) + if not multiple_register_requested: + data_type = [data_type] + offset = register_address - start_address + decoder.reset() + decoder.skip_bytes(offset * 2) + val = [struct.unpack(">e", struct.pack(">H", decoder.decode_16bit_uint())) if t == + ModbusDataType.FLOAT_16 else getattr(decoder, t.decoding_method)() for t in data_type] + results[register_address] = val if multiple_register_requested else val[0] + return results + except pymodbus.exceptions.ConnectionException as e: + self.close() + e.args += (NO_CONNECTION.format(self.address, self.port),) + raise e + except pymodbus.exceptions.ModbusIOException as e: + self.close() + e.args += (NO_VALUES.format(self.address, self.port),) + raise e + except Exception as e: + self.close() + raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e def read_input_registers_bulk(self, start_address: int, diff --git a/packages/modules/common/modbus_test.py b/packages/modules/common/modbus_test.py new file mode 100644 index 0000000000..06605b2920 --- /dev/null +++ b/packages/modules/common/modbus_test.py @@ -0,0 +1,55 @@ +import threading +import time +from unittest.mock import Mock + +from modules.common.modbus import ModbusClient, ModbusDataType + + +def _fake_response(): + response = Mock() + response.isError.return_value = False + response.registers = [0] + return response + + +def test_concurrent_reads_and_writes_are_serialized_and_none_dropped(): + # Simulates the real-world case (internal_chargepoint_handler): the main poll loop reading the + # EVSE state concurrently with a phase-switch/cp-interruption thread writing set_current, both on + # the same shared serial connection. Every call must still run to completion (nothing dropped/ + # skipped), and no two calls may ever be "on the wire" at the same time (no interleaving). + active = 0 + max_concurrent = 0 + guard = threading.Lock() # protects the counters below, not part of the code under test + + def slow_delegate_call(*args, **kwargs): + nonlocal active, max_concurrent + with guard: + active += 1 + max_concurrent = max(max_concurrent, active) + time.sleep(0.02) + with guard: + active -= 1 + return _fake_response() + + delegate = Mock() + delegate.is_socket_open.return_value = True + delegate.read_holding_registers.side_effect = slow_delegate_call + delegate.write_registers.side_effect = slow_delegate_call + + client = ModbusClient(delegate, "test", 502) + + reads = 15 + writes = 15 + threads = ( + [threading.Thread(target=client.read_holding_registers, args=(1000, ModbusDataType.UINT_16)) + for _ in range(reads)] + + [threading.Thread(target=client.write_register, args=(1000, 0)) for _ in range(writes)] + ) + for t in threads: + t.start() + for t in threads: + t.join() + + assert delegate.read_holding_registers.call_count == reads + assert delegate.write_registers.call_count == writes + assert max_concurrent == 1 From b5c64b6b0a35eb9ea94012aa139871d9ff3a575d Mon Sep 17 00:00:00 2001 From: seaspotter Date: Tue, 15 Sep 2026 11:54:16 +0200 Subject: [PATCH 2/2] Modbus: verbleibende ungesperrte Zugriffe auf den rohen Delegate schliessen Follow-up zu #3895 - Copilot-Review deckte auf, dass die neue Verbindungs- sperre in ModbusClient nur die eigenen Wrapper-Methoden schuetzt, waehrend drei Stellen direkt an self._delegate/.delegate vorbeigreifen und die Sperre damit umgehen: - sungrow_sh/bat.py: unnoetiger Roh-Zugriff fuer ein Bitfeld, das per ModbusDataType.UINT_16 identisch dekodiert werden kann - jetzt ueber den normalen, gesperrten Lesepfad. - marstek/venus_c_e/bat.py und openwb_series2_satellit/chargepoint_module.py (Phasenumschaltung) brauchten echtes FC06 (Write Single Register), das write_register() (FC16) nicht abbildet - dafuer gab es bisher keinen gesperrten Zugriffspfad. Neue Methode write_single_register_raw() schliesst diese Luecke. --- .../openwb_series2_satellit/chargepoint_module.py | 8 ++++---- packages/modules/common/modbus.py | 6 ++++++ packages/modules/common/modbus_test.py | 15 +++++++++++++++ packages/modules/devices/marstek/venus_c_e/bat.py | 2 +- .../modules/devices/sungrow/sungrow_sh/bat.py | 3 +-- 5 files changed, 27 insertions(+), 7 deletions(-) diff --git a/packages/modules/chargepoints/openwb_series2_satellit/chargepoint_module.py b/packages/modules/chargepoints/openwb_series2_satellit/chargepoint_module.py index 3a6b63979d..822172aa69 100644 --- a/packages/modules/chargepoints/openwb_series2_satellit/chargepoint_module.py +++ b/packages/modules/chargepoints/openwb_series2_satellit/chargepoint_module.py @@ -147,16 +147,16 @@ def switch_phases(self, phases_to_use: int) -> None: self._client.evse_client.set_current(0) time.sleep(5) if phases_to_use == 1: - self._client.client.delegate.write_register( + self._client.client.write_single_register_raw( 0x0001, 256, unit=self.ID_PHASE_SWITCH_UNIT) time.sleep(1) - self._client.client.delegate.write_register( + self._client.client.write_single_register_raw( 0x0001, 512, unit=self.ID_PHASE_SWITCH_UNIT) else: - self._client.client.delegate.write_register( + self._client.client.write_single_register_raw( 0x0002, 512, unit=self.ID_PHASE_SWITCH_UNIT) time.sleep(1) - self._client.client.delegate.write_register( + self._client.client.write_single_register_raw( 0x0002, 256, unit=self.ID_PHASE_SWITCH_UNIT) except AttributeError: self._create_client() diff --git a/packages/modules/common/modbus.py b/packages/modules/common/modbus.py index a90526f7db..3db58c8d81 100644 --- a/packages/modules/common/modbus.py +++ b/packages/modules/common/modbus.py @@ -232,6 +232,12 @@ def write_register(self, address: int, value: Union[int, float], data_type: Opti # Fallback für bestehenden Code ohne data_type self._delegate.write_registers(address, value, **kwargs) + def write_single_register_raw(self, address: int, value: int, **kwargs: Any): + # FC06 (Write Single Register) statt FC16 wie write_register() oben - manche Geräte akzeptieren + # FC16 nicht zuverlässig für ein einzelnes Register. + with self._lock: + self._delegate.write_register(address, value, **kwargs) + def write_single_coil(self, address: int, value: bool, **kwargs: Any): with self._lock: self._delegate.write_coil(address, value, **kwargs) diff --git a/packages/modules/common/modbus_test.py b/packages/modules/common/modbus_test.py index 06605b2920..edc75882f2 100644 --- a/packages/modules/common/modbus_test.py +++ b/packages/modules/common/modbus_test.py @@ -35,15 +35,18 @@ def slow_delegate_call(*args, **kwargs): delegate.is_socket_open.return_value = True delegate.read_holding_registers.side_effect = slow_delegate_call delegate.write_registers.side_effect = slow_delegate_call + delegate.write_register.side_effect = slow_delegate_call client = ModbusClient(delegate, "test", 502) reads = 15 writes = 15 + raw_writes = 15 threads = ( [threading.Thread(target=client.read_holding_registers, args=(1000, ModbusDataType.UINT_16)) for _ in range(reads)] + [threading.Thread(target=client.write_register, args=(1000, 0)) for _ in range(writes)] + + [threading.Thread(target=client.write_single_register_raw, args=(1000, 0)) for _ in range(raw_writes)] ) for t in threads: t.start() @@ -52,4 +55,16 @@ def slow_delegate_call(*args, **kwargs): assert delegate.read_holding_registers.call_count == reads assert delegate.write_registers.call_count == writes + assert delegate.write_register.call_count == raw_writes assert max_concurrent == 1 + + +def test_write_single_register_raw_uses_fc06_not_fc16(): + delegate = Mock() + delegate.is_socket_open.return_value = True + + client = ModbusClient(delegate, "test", 502) + client.write_single_register_raw(1000, 42, unit=1) + + delegate.write_register.assert_called_once_with(1000, 42, unit=1) + delegate.write_registers.assert_not_called() diff --git a/packages/modules/devices/marstek/venus_c_e/bat.py b/packages/modules/devices/marstek/venus_c_e/bat.py index 0e22812173..c022117f21 100644 --- a/packages/modules/devices/marstek/venus_c_e/bat.py +++ b/packages/modules/devices/marstek/venus_c_e/bat.py @@ -35,7 +35,7 @@ def _read_reg(self, addr: int, type_: ModbusDataType) -> Union[int, float]: def _write_reg(self, addr: int, val: int) -> None: # Marstek Venus does not work with write_registers! - self.client._delegate.write_register(addr, val, unit=self.component_config.configuration.modbus_id) + self.client.write_single_register_raw(addr, val, unit=self.component_config.configuration.modbus_id) def update(self) -> None: power = -self._read_reg(32202, ModbusDataType.INT_32) diff --git a/packages/modules/devices/sungrow/sungrow_sh/bat.py b/packages/modules/devices/sungrow/sungrow_sh/bat.py index cb2f64d3f5..ab3f3d0815 100644 --- a/packages/modules/devices/sungrow/sungrow_sh/bat.py +++ b/packages/modules/devices/sungrow/sungrow_sh/bat.py @@ -80,8 +80,7 @@ def update(self) -> None: bat_current = self.__tcp_client.read_input_registers(13020, ModbusDataType.INT_16, unit=unit) * -0.1 bat_power = self.__tcp_client.read_input_registers(13021, ModbusDataType.UINT_16, unit=unit) - resp = self.__tcp_client._delegate.read_input_registers(13000, 1, unit=unit) - running_state = resp.registers[0] + running_state = self.__tcp_client.read_input_registers(13000, ModbusDataType.UINT_16, unit=unit) is_charging = (running_state & 0x02) != 0 is_discharging = (running_state & 0x04) != 0