Message RTU.

pymodbus/framer/binary_framer.py

@@ -5,7 +5,7 @@
 from pymodbus.exceptions import ModbusIOException
 from pymodbus.framer.base import BYTE_ORDER, FRAME_HEADER, ModbusFramer
 from pymodbus.logging import Log
-from pymodbus.utilities import checkCRC, computeCRC
+from pymodbus.message.rtu import MessageRTU
 
 
 BINARY_FRAME_HEADER = BYTE_ORDER + FRAME_HEADER
@@ -113,7 +113,7 @@ def buildPacket(self, message):
             struct.pack(BINARY_FRAME_HEADER, message.slave_id, message.function_code)
             + data
         )
-        packet += struct.pack(">H", computeCRC(packet))
+        packet += struct.pack(">H", MessageRTU.compute_CRC(packet))
         packet = self._start + packet + self._end
         return packet
 

pymodbus/framer/rtu_framer.py

@@ -8,7 +8,8 @@
 )
 from pymodbus.framer.base import BYTE_ORDER, FRAME_HEADER, ModbusFramer
 from pymodbus.logging import Log
-from pymodbus.utilities import ModbusTransactionState, checkCRC, computeCRC
+from pymodbus.message.rtu import MessageRTU
+from pymodbus.utilities import ModbusTransactionState
 
 
 RTU_FRAME_HEADER = BYTE_ORDER + FRAME_HEADER
@@ -175,7 +176,7 @@ def buildPacket(self, message):
             struct.pack(RTU_FRAME_HEADER, message.slave_id, message.function_code)
             + data
         )
-        packet += struct.pack(">H", computeCRC(packet))
+        packet += struct.pack(">H", MessageRTU.compute_CRC(packet))
         # Ensure that transaction is actually the slave id for serial comms
         if message.slave_id:
             message.transaction_id = message.slave_id

pymodbus/message/rtu.py

@@ -40,10 +40,62 @@ class MessageRTU(MessageBase):
     neither when receiving nor when sending.
     """
 
+    @classmethod
+    def generate_crc16_table(cls) -> list[int]:
+        """Generate a crc16 lookup table.
+
+        .. note:: This will only be generated once
+        """
+        result = []
+        for byte in range(256):
+            crc = 0x0000
+            for _ in range(8):
+                if (byte ^ crc) & 0x0001:
+                    crc = (crc >> 1) ^ 0xA001
+                else:
+                    crc >>= 1
+                byte >>= 1
+            result.append(crc)
+        return result
+    crc16_table: list[int] = [0]
+
     def decode(self, _data: bytes) -> tuple[int, int, int, bytes]:
         """Decode message."""
         return 0, 0, 0, b''
 
-    def encode(self, _data: bytes, _device_id: int, _tid: int) -> bytes:
+    def encode(self, data: bytes, device_id: int, _tid: int) -> bytes:
         """Decode message."""
-        return b''
+        packet = device_id.to_bytes(1,'big') + data
+        return packet + MessageRTU.compute_CRC(packet).to_bytes(2,'big')
+
+    @classmethod
+    def check_CRC(cls, data: bytes, check: int) -> bool:
+        """Check if the data matches the passed in CRC.
+
+        :param data: The data to create a crc16 of
+        :param check: The CRC to validate
+        :returns: True if matched, False otherwise
+        """
+        return cls.compute_CRC(data) == check
+
+    @classmethod
+    def compute_CRC(cls, data: bytes) -> int:
+        """Compute a crc16 on the passed in bytes.
+
+        For modbus, this is only used on the binary serial protocols (in this
+        case RTU).
+
+        The difference between modbus's crc16 and a normal crc16
+        is that modbus starts the crc value out at 0xffff.
+
+        :param data: The data to create a crc16 of
+        :returns: The calculated CRC
+        """
+        crc = 0xFFFF
+        for data_byte in data:
+            idx = cls.crc16_table[(crc ^ int(data_byte)) & 0xFF]
+            crc = ((crc >> 8) & 0xFF) ^ idx
+        swapped = ((crc << 8) & 0xFF00) | ((crc >> 8) & 0x00FF)
+        return swapped
+
+MessageRTU.crc16_table = MessageRTU.generate_crc16_table()

pymodbus/utilities.py

@@ -10,8 +10,6 @@
     "pack_bitstring",
     "unpack_bitstring",
     "default",
-    "computeCRC",
-    "checkCRC",
     "rtuFrameSize",
 ]
 
@@ -152,57 +150,6 @@ def unpack_bitstring(data: bytes) -> list[bool]:
 # --------------------------------------------------------------------------- #
 
 
-def __generate_crc16_table():
-    """Generate a crc16 lookup table.
-
-    .. note:: This will only be generated once
-    """
-    result = []
-    for byte in range(256):
-        crc = 0x0000
-        for _ in range(8):
-            if (byte ^ crc) & 0x0001:
-                crc = (crc >> 1) ^ 0xA001
-            else:
-                crc >>= 1
-            byte >>= 1
-        result.append(crc)
-    return result
-
-
-__crc16_table = __generate_crc16_table()
-
-
-def computeCRC(data):  # pylint: disable=invalid-name
-    """Compute a crc16 on the passed in string.
-
-    For modbus, this is only used on the binary serial protocols (in this
-    case RTU).
-
-    The difference between modbus's crc16 and a normal crc16
-    is that modbus starts the crc value out at 0xffff.
-
-    :param data: The data to create a crc16 of
-    :returns: The calculated CRC
-    """
-    crc = 0xFFFF
-    for data_byte in data:
-        idx = __crc16_table[(crc ^ int(data_byte)) & 0xFF]
-        crc = ((crc >> 8) & 0xFF) ^ idx
-    swapped = ((crc << 8) & 0xFF00) | ((crc >> 8) & 0x00FF)
-    return swapped
-
-
-def checkCRC(data, check):  # pylint: disable=invalid-name
-    """Check if the data matches the passed in CRC.
-
-    :param data: The data to create a crc16 of
-    :param check: The CRC to validate
-    :returns: True if matched, False otherwise
-    """
-    return computeCRC(data) == check
-
-
 def rtuFrameSize(data, byte_count_pos):  # pylint: disable=invalid-name
     """Calculate the size of the frame based on the byte count.
 

test/message/test_ascii.py

@@ -1,6 +1,4 @@
 """Test transport."""
-import struct
-
 import pytest
 
 from pymodbus.message.ascii import MessageAscii
@@ -18,22 +16,22 @@ def prepare_frame():
 
     def test_check_LRC(self):
         """Test check_LRC."""
-        data = struct.pack(">HHHH", 0x1234, 0x2345, 0x3456, 0x4567)
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
         assert MessageAscii.check_LRC(data, 0x1C)
 
     def test_check_noLRC(self):
         """Test check_LRC."""
-        data = struct.pack(">HHHH", 0x1234, 0x2345, 0x3456, 0x4567)
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
         assert not MessageAscii.check_LRC(data, 0x0C)
 
     def test_compute_LRC(self):
         """Test compute_LRC."""
-        data = struct.pack(">HHHH", 0x1234, 0x2345, 0x3456, 0x4567)
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
         assert MessageAscii.compute_LRC(data) == 0x1c
 
     def test_roundtrip_LRC(self):
         """Test combined compute/check LRC."""
-        data = struct.pack(">HHHH", 0x1234, 0x2345, 0x3456, 0x4567)
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
         assert MessageAscii.compute_LRC(data) == 0x1c
         assert MessageAscii.check_LRC(data, 0x1C)
 

test/message/test_rtu.py

@@ -0,0 +1,117 @@
+"""Test transport."""
+import pytest
+
+from pymodbus.message.rtu import MessageRTU
+
+
+class TestMessageRTU:
+    """Test message module."""
+
+    @staticmethod
+    @pytest.fixture(name="frame")
+    def prepare_frame():
+        """Return message object."""
+        return MessageRTU([1], False)
+
+
+    def test_crc16_table(self):
+        """Test the crc16 table is prefilled."""
+        assert len(MessageRTU.crc16_table) == 256
+        assert isinstance(MessageRTU.crc16_table[0], int)
+        assert isinstance(MessageRTU.crc16_table[255], int)
+
+    def test_check_CRC(self):
+        """Test check_CRC."""
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
+        assert MessageRTU.check_CRC(data, 0xE2DB)
+
+    def test_check_noCRC(self):
+        """Test check_CRC."""
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
+        assert not MessageRTU.check_CRC(data, 0xDBE2)
+
+    def test_compute_CRC(self):
+        """Test compute_CRC."""
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
+        assert MessageRTU.compute_CRC(data) == 0xE2DB
+
+    def test_roundtrip_CRC(self):
+        """Test combined compute/check CRC."""
+        data = b'\x12\x34\x23\x45\x34\x56\x45\x67'
+        assert MessageRTU.compute_CRC(data) == 0xE2DB
+        assert MessageRTU.check_CRC(data, 0xE2DB)
+
+    #   b"\x02\x01\x01\x00Q\xcc"
+    #   b"\x01\x01\x03\x01\x00\n\xed\x89"
+    #   b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x43"
+    #   b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAD"
+
+    #   b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAD\x11\x03" # good frame + part of next frame
+
+    #   b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAC"  # invalid frame CRC
+    #   b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAC"  # bad crc
+    #   b"\x61\x62\x00\x01\x00\n\xec\x1c"  # bad function code
+    #   b"\x01\x03\x03\x01\x00\n\x94\x49" # Not ok
+
+    # test frame ready
+    # (b"", False),
+    # (b"\x11", False),
+    # (b"\x11\x03", False),
+    # (b"\x11\x03\x06", False),
+    # (b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49", False),
+    # (b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAD", True),
+    # (b"\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAD\xAB\xCD", True),
+
+
+    @pytest.mark.parametrize(
+        ("packet", "used_len", "res_id", "res"),
+        [
+            (b':010100010001FC\r\n', 17, 1, b'\x01\x00\x01\x00\x01'),
+            (b':00010001000AF4\r\n', 17, 0, b'\x01\x00\x01\x00\x0a'),
+            (b':01010001000AF3\r\n', 17, 1, b'\x01\x00\x01\x00\x0a'),
+            (b':61620001000A32\r\n', 17, 97, b'\x62\x00\x01\x00\x0a'),
+            (b':01270001000ACD\r\n', 17, 1, b'\x27\x00\x01\x00\x0a'),
+            (b':010100', 0, 0, b''), # short frame
+            (b':00010001000AF4', 0, 0, b''),
+            (b'abc:00010001000AF4', 3, 0, b''), # garble before frame
+            (b'abc00010001000AF4', 17, 0, b''), # only garble
+            (b':01010001000A00\r\n', 17, 0, b''),
+        ],
+    )
+    def xtest_decode(self, frame, packet, used_len, res_id, res):
+        """Test decode."""
+        res_len, tid, dev_id, data = frame.decode(packet)
+        assert res_len == used_len
+        assert data == res
+        assert not tid
+        assert dev_id == res_id
+
+    @pytest.mark.parametrize(
+        ("data", "dev_id", "res_msg"),
+        [
+            (b'\x01\x01\x00', 2, b'\x02\x01\x01\x00\x51\xcc'),
+            (b'\x03\x06\xAE\x41\x56\x52\x43\x40', 17, b'\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAD'),
+            (b'\x01\x03\x01\x00\x0a', 1, b'\x01\x01\x03\x01\x00\x0a\xed\x89'),
+        ],
+    )
+    def test_encode(self, frame, data, dev_id, res_msg):
+        """Test encode."""
+        msg = frame.encode(data, dev_id, 0)
+        assert res_msg == msg
+        assert dev_id == int(msg[0])
+
+    @pytest.mark.parametrize(
+        ("data", "dev_id", "res_msg"),
+        [
+            (b'\x01\x01\x00', 2, b'\x02\x01\x01\x00\x51\xcc'),
+            (b'\x03\x06\xAE\x41\x56\x52\x43\x40', 17, b'\x11\x03\x06\xAE\x41\x56\x52\x43\x40\x49\xAD'),
+            (b'\x01\x03\x01\x00\x0a', 1, b'\x01\x01\x03\x01\x00\x0a\xed\x89'),
+        ],
+    )
+    def xtest_roundtrip(self, frame, data, dev_id, res_msg):
+        """Test encode."""
+        msg = frame.encode(data, dev_id, 0)
+        res_len, _, res_id, res_data = frame.decode(msg)
+        assert data == res_data
+        assert dev_id == res_id
+        assert res_len == len(res_msg)

test/test_utilities.py

@@ -2,7 +2,6 @@
 import struct
 
 from pymodbus.utilities import (
-    checkCRC,
     default,
     dict_property,
     pack_bitstring,
@@ -91,8 +90,3 @@ def test_bit_packing(self):
         """Test all string <=> bit packing functions."""
         assert unpack_bitstring(b"\x55") == self.bits
         assert pack_bitstring(self.bits) == b"\x55"
-
-    def test_cyclic_redundancy_check(self):
-        """Test the cyclic redundancy check code."""
-        assert checkCRC(self.data, 0xE2DB)
-        assert checkCRC(self.string, 0x889E)