Convert endianness (#2506)

Co-authored-by: jan iversen <jancasacondor@gmail.com>

Author: daanwtb

examples/client_async_calls.py

@@ -146,6 +146,14 @@ async def async_handle_holding_registers(client):
     assert not rr.isError()  # test that call was OK
     assert rr.registers == [10]
 
+    value_int32 = 13211
+    registers = client.convert_to_registers(value_int32, client.DATATYPE.INT32)
+    await client.write_registers(1, registers, slave=SLAVE)
+    rr = await client.read_holding_registers(1, count=len(registers), slave=SLAVE)
+    assert not rr.isError()  # test that call was OK
+    value = client.convert_from_registers(rr.registers, client.DATATYPE.INT32)
+    assert value_int32 == value
+
     _logger.info("### write read holding registers")
     arguments = {
         "read_address": 1,

examples/client_calls.py

@@ -135,10 +135,13 @@ def handle_holding_registers(client):
     assert not rr.isError()  # test that call was OK
     assert rr.registers[0] == 10
 
-    client.write_registers(1, [10] * 8, slave=SLAVE)
-    rr = client.read_holding_registers(1, count=8, slave=SLAVE)
+    value_int32 = 13211
+    registers = client.convert_to_registers(value_int32, client.DATATYPE.INT32)
+    client.write_registers(1, registers, slave=SLAVE)
+    rr = client.read_holding_registers(1, count=len(registers), slave=SLAVE)
     assert not rr.isError()  # test that call was OK
-    assert rr.registers == [10] * 8
+    value = client.convert_from_registers(rr.registers, client.DATATYPE.INT32)
+    assert value_int32 == value
 
     _logger.info("### write read holding registers")
     arguments = {

pymodbus/client/mixin.py

@@ -4,7 +4,7 @@
 import struct
 from abc import abstractmethod
 from enum import Enum
-from typing import Generic, TypeVar
+from typing import Generic, Literal, TypeVar, cast
 
 import pymodbus.pdu.bit_message as pdu_bit
 import pymodbus.pdu.diag_message as pdu_diag
@@ -691,65 +691,83 @@ class DATATYPE(Enum):
         FLOAT32 = ("f", 2)
         FLOAT64 = ("d", 4)
         STRING = ("s", 0)
-        BITS = "bits"
+        BITS = ("bits", 0)
 
     @classmethod
     def convert_from_registers(
-        cls, registers: list[int], data_type: DATATYPE
-    ) -> int | float | str | list[bool]:
+        cls, registers: list[int], data_type: DATATYPE, word_order: Literal["big", "little"] = "big"
+    ) -> int | float | str | list[bool] | list[int] | list[float]:
         """Convert registers to int/float/str.
 
         :param registers: list of registers received from e.g. read_holding_registers()
         :param data_type: data type to convert to
-        :returns: int, float, str or list[bool] depending on "data_type"
-        :raises ModbusException: when size of registers is not 1, 2 or 4
-        """
-        byte_list = bytearray()
-        for x in registers:
-            byte_list.extend(int.to_bytes(x, 2, "big"))
-        if data_type == cls.DATATYPE.STRING:
-            # remove trailing null bytes
-            trailing_nulls_begin = len(byte_list)
-            while trailing_nulls_begin > 0 and not byte_list[trailing_nulls_begin - 1]:
-                trailing_nulls_begin -= 1
-            byte_list = byte_list[:trailing_nulls_begin]
-
-            return byte_list.decode("utf-8")
-        if data_type == cls.DATATYPE.BITS:
+        :param word_order: "big"/"little" order of words/registers
+        :returns: scalar or array of "data_type"
+        :raises ModbusException: when size of registers is not a multiple of data_type
+        """
+        if not (data_len := data_type.value[1]):
+            byte_list = bytearray()
+            if word_order == "little":
+                registers.reverse()
+            for x in registers:
+                byte_list.extend(int.to_bytes(x, 2, "big"))
+            if data_type == cls.DATATYPE.STRING:
+                trailing_nulls_begin = len(byte_list)
+                while trailing_nulls_begin > 0 and not byte_list[trailing_nulls_begin - 1]:
+                    trailing_nulls_begin -= 1
+                byte_list = byte_list[:trailing_nulls_begin]
+                return byte_list.decode("utf-8")
             return unpack_bitstring(byte_list)
-        if len(registers) != data_type.value[1]:
+        if (reg_len := len(registers)) % data_len:
             raise ModbusException(
-                f"Illegal size ({len(registers)}) of register array, cannot convert!"
+                f"Registers illegal size ({len(registers)}) expected multiple of {data_len}!"
             )
-        return struct.unpack(f">{data_type.value[0]}", byte_list)[0]
+
+        result = []
+        for i in range(0, reg_len, data_len):
+            regs = registers[i:i+data_len]
+            if word_order == "little":
+                regs.reverse()
+            byte_list = bytearray()
+            for x in regs:
+                byte_list.extend(int.to_bytes(x, 2, "big"))
+            result.append(struct.unpack(f">{data_type.value[0]}", byte_list)[0])
+        return result if len(result) != 1 else result[0]
 
     @classmethod
     def convert_to_registers(
-        cls, value: int | float | str | list[bool], data_type: DATATYPE
+        cls, value: int | float | str | list[bool] | list[int] | list[float] , data_type: DATATYPE, word_order: Literal["big", "little"] = "big"
     ) -> list[int]:
         """Convert int/float/str to registers (16/32/64 bit).
 
         :param value: value to be converted
-        :param data_type: data type to be encoded as registers
+        :param data_type: data type to convert from
+        :param word_order: "big"/"little" order of words/registers
         :returns: List of registers, can be used directly in e.g. write_registers()
         :raises TypeError: when there is a mismatch between data_type and value
         """
         if data_type == cls.DATATYPE.BITS:
             if not isinstance(value, list):
-                raise TypeError(f"Value should be string but is {type(value)}.")
+                raise TypeError(f"Value should be list of bool but is {type(value)}.")
             if (missing := len(value) % 16):
                 value = value + [False] * (16 - missing)
-            byte_list = pack_bitstring(value)
+            byte_list = pack_bitstring(cast(list[bool], value))
         elif data_type == cls.DATATYPE.STRING:
             if not isinstance(value, str):
                 raise TypeError(f"Value should be string but is {type(value)}.")
             byte_list = value.encode()
             if len(byte_list) % 2:
                 byte_list += b"\x00"
         else:
-            byte_list = struct.pack(f">{data_type.value[0]}", value)
+            if not isinstance(value, list):
+                value = cast(list[int], [value])
+            byte_list = bytearray()
+            for v in value:
+                byte_list.extend(struct.pack(f">{data_type.value[0]}", v))
         regs = [
             int.from_bytes(byte_list[x : x + 2], "big")
             for x in range(0, len(byte_list), 2)
         ]
+        if word_order == "little":
+            regs.reverse()
         return regs

test/client/test_client.py

@@ -95,6 +95,7 @@ def fake_execute(_self, _no_response_expected, request):
             getattr(ModbusClientMixin(), method)(**arglist[arg])
             assert isinstance(pdu_to_call, pdu_request)
 
+    @pytest.mark.parametrize(("word_order"), ["big", "little", None])
     @pytest.mark.parametrize(
         ("datatype", "value", "registers"),
         [
@@ -156,11 +157,16 @@ def fake_execute(_self, _no_response_expected, request):
             ),
         ],
     )
-    def test_client_mixin_convert(self, datatype, registers, value):
+    def test_client_mixin_convert(self, datatype, word_order, registers, value):
         """Test converter methods."""
-        regs = ModbusClientMixin.convert_to_registers(value, datatype)
+        if word_order == "little":
+            x = registers.copy()
+            x.reverse()
+            registers = x
+        kwargs = {"word_order": word_order} if word_order else {}
+        regs = ModbusClientMixin.convert_to_registers(value, datatype, **kwargs)
         assert regs == registers
-        result = ModbusClientMixin.convert_from_registers(registers, datatype)
+        result = ModbusClientMixin.convert_from_registers(registers, datatype, **kwargs)
         if datatype == ModbusClientMixin.DATATYPE.FLOAT32:
             result = round(result, 6)
         if datatype == ModbusClientMixin.DATATYPE.BITS:
@@ -174,6 +180,7 @@ def test_client_mixin_convert(self, datatype, registers, value):
             (ModbusClientMixin.DATATYPE.STRING, "0123", [b'\x30\x31', b'\x32\x33']),
             (ModbusClientMixin.DATATYPE.UINT16, 258, [b'\x01\x02']),
             (ModbusClientMixin.DATATYPE.INT16, -32510, [b'\x81\x02']),
+            (ModbusClientMixin.DATATYPE.INT16, [-32510, 258], [b'\x81\x02', b'\x01\x02']),
             (ModbusClientMixin.DATATYPE.UINT32, 16909060, [b'\x01\x02', b'\x03\x04']),
             (ModbusClientMixin.DATATYPE.INT32, -2130574588, [b'\x81\x02', b'\x03\x04']),
             (
@@ -209,6 +216,9 @@ def test_client_mixin_convert_fail(self):
         with pytest.raises(ModbusException):
             ModbusClientMixin.convert_from_registers([123], ModbusClientMixin.DATATYPE.FLOAT64)
 
+        with pytest.raises(TypeError):
+            ModbusClientMixin.convert_to_registers(bool, ModbusClientMixin.DATATYPE.BITS)
+
 
 class TestClientBase:
     """Test client code."""