Write 32-bit modbus registers high-word-first on S series pumps

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),

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()

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

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(