Simplify some BinaryPayload pack operations

Author: jameshilliard

pymodbus/payload.py

@@ -11,6 +11,8 @@
     "BinaryPayloadDecoder",
 ]
 
+from array import array
+
 # pylint: disable=missing-type-doc
 from struct import pack, unpack
 
@@ -19,13 +21,11 @@
 from pymodbus.logging import Log
 from pymodbus.utilities import (
     pack_bitstring,
+    system_endian,
     unpack_bitstring,
 )
 
 
-WC = {"b": 1, "h": 2, "e": 2, "i": 4, "l": 4, "q": 8, "f": 4, "d": 8}
-
-
 class BinaryPayloadBuilder:
     """A utility that helps build payload messages to be written with the various modbus messages.
 
@@ -69,15 +69,14 @@ def _pack_words(self, fstring: str, value) -> bytes:
         :return:
         """
         value = pack(f"!{fstring}", value)
-        wordorder = WC.get(fstring.lower()) // 2  # type: ignore[operator]
-        upperbyte = f"!{wordorder}H"
-        payload = unpack(upperbyte, value)
-
-        if self._wordorder == Endian.LITTLE:
-            payload = payload[::-1]
-
-        fstring = self._byteorder + "H"
-        return b"".join(pack(fstring, word) for word in payload)
+        if Endian.LITTLE in {self._byteorder, self._wordorder}:
+            value = array("H", value)
+            if self._byteorder == Endian.LITTLE:
+                value.byteswap()
+            if self._wordorder == Endian.LITTLE:
+                value.reverse()
+            value = value.tobytes()
+        return value
 
     def encode(self) -> bytes:
         """Get the payload buffer encoded in bytes."""
@@ -99,15 +98,14 @@ def to_registers(self):
 
         :returns: The register layout to use as a block
         """
-        # fstring = self._byteorder+"H"
-        fstring = "!H"
-        payload = self.build()
-        if self._repack:
-            payload = [unpack(self._byteorder + "H", value)[0] for value in payload]
-        else:
-            payload = [unpack(fstring, value)[0] for value in payload]
-        Log.debug("{}", payload)
-        return payload
+        payload = array("H", b''.join(self.build()))
+        swap = system_endian() != Endian.BIG
+        if self._repack and self._byteorder == system_endian():
+            swap = False
+        if swap:
+            payload.byteswap()
+        Log.debug("{}", payload.tolist())
+        return payload.tolist()
 
     def to_coils(self) -> list[bool]:
         """Convert the payload buffer into a coil layout that can be used as a context block.
@@ -295,8 +293,10 @@ def fromRegisters(
         """
         Log.debug("{}", registers)
         if isinstance(registers, list):  # repack into flat binary
-            payload = b"".join(pack("!H", x) for x in registers)
-            return cls(payload, byteorder, wordorder)
+            payload = array("H", registers)
+            if system_endian() != Endian.BIG:
+                payload.byteswap()
+            return cls(payload.tobytes(), byteorder, wordorder)
         raise ParameterException("Invalid collection of registers supplied")
 
     @classmethod
@@ -324,7 +324,7 @@ def fromCoils(
             return cls(payload, byteorder)
         raise ParameterException("Invalid collection of coils supplied")
 
-    def _unpack_words(self, fstring: str, handle) -> bytes:
+    def _unpack_words(self, handle) -> bytes:
         """Unpack words based on the word order and byte order.
 
         # ---------------------------------------------- #
@@ -336,15 +336,14 @@ def _unpack_words(self, fstring: str, handle) -> bytes:
         :param handle: Value to be unpacked
         :return:
         """
-        wc_value = WC.get(fstring.lower()) // 2  # type: ignore[operator]
-        handle = unpack(f"!{wc_value}H", handle)
-        if self._wordorder == Endian.LITTLE:
-            handle = list(reversed(handle))
-
-        # Repack as unsigned Integer
-        handle = [pack(self._byteorder + "H", p) for p in handle]
+        if Endian.LITTLE in {self._byteorder, self._wordorder}:
+            handle = array("H", handle)
+            if self._byteorder == Endian.LITTLE:
+                handle.byteswap()
+            if self._wordorder == Endian.LITTLE:
+                handle.reverse()
+            handle = handle.tobytes()
         Log.debug("handle: {}", handle)
-        handle = b"".join(handle)
         return handle
 
     def reset(self):
@@ -377,15 +376,15 @@ def decode_32bit_uint(self):
         self._pointer += 4
         fstring = "I"
         handle = self._payload[self._pointer - 4 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_64bit_uint(self):
         """Decode a 64 bit unsigned int from the buffer."""
         self._pointer += 8
         fstring = "Q"
         handle = self._payload[self._pointer - 8 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_8bit_int(self):
@@ -407,39 +406,39 @@ def decode_32bit_int(self):
         self._pointer += 4
         fstring = "i"
         handle = self._payload[self._pointer - 4 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_64bit_int(self):
         """Decode a 64 bit signed int from the buffer."""
         self._pointer += 8
         fstring = "q"
         handle = self._payload[self._pointer - 8 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_16bit_float(self):
         """Decode a 16 bit float from the buffer."""
         self._pointer += 2
         fstring = "e"
         handle = self._payload[self._pointer - 2 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_32bit_float(self):
         """Decode a 32 bit float from the buffer."""
         self._pointer += 4
         fstring = "f"
         handle = self._payload[self._pointer - 4 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_64bit_float(self):
         """Decode a 64 bit float(double) from the buffer."""
         self._pointer += 8
         fstring = "d"
         handle = self._payload[self._pointer - 8 : self._pointer]
-        handle = self._unpack_words(fstring, handle)
+        handle = self._unpack_words(handle)
         return unpack("!" + fstring, handle)[0]
 
     def decode_string(self, size=1):

pymodbus/utilities.py

@@ -11,10 +11,14 @@
     "unpack_bitstring",
     "default",
     "rtuFrameSize",
+    "system_endian",
 ]
 
 # pylint: disable=missing-type-doc
 import struct
+import sys
+
+from pymodbus.constants import Endian
 
 
 class ModbusTransactionState:  # pylint: disable=too-few-public-methods
@@ -182,3 +186,11 @@ def hexlify_packets(packet):
     if not packet:
         return ""
     return " ".join([hex(int(x)) for x in packet])
+
+
+def system_endian():
+    """Get system endianness.
+
+    :return: Endian of system
+    """
+    return Endian.LITTLE if sys.byteorder == "little" else Endian.BIG