From 902df3ee46ed1bb390fd99418809c51f9f23dbe4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 2 Sep 2026 06:46:36 +0000 Subject: [PATCH 1/2] Initial plan From 6f29b9bd1d9ffa542330a9c78eb401d329fb4490 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 2 Sep 2026 06:50:56 +0000 Subject: [PATCH 2/2] Write 32-bit modbus registers high-word-first on S series pumps Co-authored-by: yozik04 <2420038+yozik04@users.noreply.github.com> --- nibe/connection/encoders.py | 20 ++++++++++++-- nibe/connection/modbus.py | 20 ++++++++++++-- tests/connection/test_encoders.py | 16 +++++++++++ tests/connection/test_modbus.py | 45 ++++++++++++++++++++++++++++--- 4 files changed, 94 insertions(+), 7 deletions(-) diff --git a/nibe/connection/encoders.py b/nibe/connection/encoders.py index 49c129dc..16382f03 100644 --- a/nibe/connection/encoders.py +++ b/nibe/connection/encoders.py @@ -135,17 +135,33 @@ def _pad(self, parser: Construct, value: int) -> bytes: class CoilDataEncoderModbus(CoilDataEncoder[List[SupportsInt]]): + """Encode and decode coil data for modbus. + + Some heat pumps use a different word order for reading and writing 32 bit + registers. `word_swap_write` overrides the word order used for encoding, + when it is not set `word_swap` is used for both directions. + """ + word_swap: Optional[bool] = None + word_swap_write: Optional[bool] = None - def __init__(self, word_swap: Optional[bool] = None): + def __init__( + self, + word_swap: Optional[bool] = None, + word_swap_write: Optional[bool] = None, + ): self.word_swap = word_swap + self.word_swap_write = word_swap_write def encode_raw_value(self, size: str, raw_value: int) -> List[SupportsInt]: signed = size in ("s32", "s16", "s8") + word_swap = ( + self.word_swap if self.word_swap_write is None else self.word_swap_write + ) raw_bytes = raw_value.to_bytes(8, "little", signed=signed) if size in ("s32", "u32"): - if self.word_swap: + if word_swap: return [ int.from_bytes(raw_bytes[0:2], "little", signed=False), int.from_bytes(raw_bytes[2:4], "little", signed=False), diff --git a/nibe/connection/modbus.py b/nibe/connection/modbus.py index 2e8f54bb..7053c3a5 100644 --- a/nibe/connection/modbus.py +++ b/nibe/connection/modbus.py @@ -1,5 +1,6 @@ import asyncio import logging +from typing import Optional from async_modbus import modbus_for_url import async_timeout @@ -20,7 +21,7 @@ WriteIOException, WriteTimeoutException, ) -from nibe.heatpump import HeatPump +from nibe.heatpump import HeatPump, Series from . import verify_connectivity_read_write_alarm @@ -72,7 +73,22 @@ def __init__( except ValueError as exc: raise ModbusUrlException(str(exc)) from exc - self.coil_encoder = CoilDataEncoderModbus(heatpump.word_swap) + self.coil_encoder = CoilDataEncoderModbus( + heatpump.word_swap, word_swap_write=self._get_word_swap_write(heatpump) + ) + + @staticmethod + def _get_word_swap_write(heatpump: HeatPump) -> Optional[bool]: + """Get word order to use when writing 32 bit registers. + + S series heat pumps return 32 bit registers with the low word first, + but expect the high word first when they are written. Other series use + the same word order in both directions. + """ + model = heatpump.model + if model is not None and model.series is Series.S: + return False + return None async def stop(self) -> None: await self._client.stream.close() diff --git a/tests/connection/test_encoders.py b/tests/connection/test_encoders.py index d1e9401c..01c0ffa7 100644 --- a/tests/connection/test_encoders.py +++ b/tests/connection/test_encoders.py @@ -154,3 +154,19 @@ def test_modbus_encode_raw_value( assert CoilDataEncoderModbus(True).encode_raw_value(size, raw_value) == raw if word_swap in (False, None): assert CoilDataEncoderModbus(False).encode_raw_value(size, raw_value) == raw + + +@pytest.mark.parametrize( + "size, raw, raw_value", + [ + ("u8", [0x0001], 1), + ("s16", [0xFFFF], -1), + ("s32", [0x0000, 0x5432], 0x5432), + ("s32", [0xFFFF, 0xF9D8], -0x628), + ], +) +def test_modbus_encode_raw_value_word_swap_write_override(size, raw, raw_value): + """Word order for writing can differ from the one used for reading.""" + encoder = CoilDataEncoderModbus(True, word_swap_write=False) + assert encoder.encode_raw_value(size, raw_value) == raw + assert encoder.decode_raw_value(size, list(reversed(raw))) == raw_value diff --git a/tests/connection/test_modbus.py b/tests/connection/test_modbus.py index 6480034e..d00ab1c8 100644 --- a/tests/connection/test_modbus.py +++ b/tests/connection/test_modbus.py @@ -32,6 +32,19 @@ def fixture_connection(heatpump: HeatPump): yield Modbus(heatpump, "tcp://127.0.0.1", 0) +@pytest.fixture(name="heatpump_f_series") +async def fixture_heatpump_f_series(): + heatpump = HeatPump(Model.F1255) + heatpump.word_swap = True + await heatpump.initialize() + yield heatpump + + +@pytest.fixture(name="connection_f_series") +def fixture_connection_f_series(heatpump_f_series: HeatPump): + yield Modbus(heatpump_f_series, "tcp://127.0.0.1", 0) + + @pytest.mark.parametrize( ("size", "raw", "value"), [ @@ -60,8 +73,8 @@ async def test_read_holding_register_coil( @pytest.mark.parametrize( ("size", "raw", "value"), [ - ("u32", [1, 0], 0x00000001), - ("u32", [0, 32768], 0x80000000), + ("u32", [0, 1], 0x00000001), + ("u32", [32768, 0], 0x80000000), ("u16", [1], 0x0001), ("u16", [32768], 0x8000), ("u8", [1], 0x01), @@ -75,6 +88,7 @@ async def test_write_holding_register( raw: List[bytes], value: Union[int, float, str], ): + """S series pumps expect the high word first when writing 32 bit registers.""" coil = Coil(40002, "test", "test", size, 1, write=True) coil_data = CoilData(coil, value) await connection.write_coil(coil_data) @@ -83,6 +97,31 @@ async def test_write_holding_register( ) +@pytest.mark.parametrize( + ("size", "raw", "value"), + [ + ("u32", [1, 0], 0x00000001), + ("u32", [0, 32768], 0x80000000), + ("u16", [1], 0x0001), + ("u8", [1], 0x01), + ], +) +async def test_write_holding_register_f_series( + connection_f_series: Modbus, + modbus_client: AsyncMock, + size: str, + raw: List[bytes], + value: Union[int, float, str], +): + """F series pumps use the same word order for reading and writing.""" + coil = Coil(40002, "test", "test", size, 1, write=True) + coil_data = CoilData(coil, value) + await connection_f_series.write_coil(coil_data) + modbus_client.write_registers.assert_called_with( + slave_id=0, starting_address=1, values=raw + ) + + @pytest.mark.parametrize( ("size", "raw", "value"), [ @@ -180,7 +219,7 @@ async def test_read_coils_failed_read( ("u8", [0], 0x00), ("s8", [0xFFF6], -0xA), ("s16", [0xFFF6], -0xA), - ("s32", [0xFFF6, 0xFFFF], -0xA), + ("s32", [0xFFFF, 0xFFF6], -0xA), ], ) async def test_write_coil_coil(