add _legacy_decoder to message rtu (#2119)

Author: janiversen

pymodbus/framer/rtu_framer.py

@@ -3,9 +3,7 @@
 import struct
 import time
 
-from pymodbus.exceptions import (
-    ModbusIOException,
-)
+from pymodbus.exceptions import ModbusIOException
 from pymodbus.framer.base import BYTE_ORDER, FRAME_HEADER, ModbusFramer
 from pymodbus.logging import Log
 from pymodbus.message.rtu import MessageRTU

pymodbus/message/base.py

@@ -16,7 +16,7 @@ class MessageBase:
 
     def __init__(
         self,
-        device_ids: list[int] | None,
+        device_ids: list[int],
         is_server: bool,
     ) -> None:
         """Initialize a message instance.
@@ -25,10 +25,12 @@ def __init__(
         """
         self.device_ids = device_ids
         self.is_server = is_server
+        self.broadcast: bool = (0 in device_ids)
+
 
     def validate_device_id(self, dev_id: int) -> bool:
         """Check if device id is expected."""
-        return not (self.device_ids and dev_id and dev_id not in self.device_ids)
+        return self.broadcast or (dev_id in self.device_ids)
 
     @abstractmethod
     def decode(self, _data: bytes) -> tuple[int, int, int, bytes]:

pymodbus/message/message.py

@@ -54,14 +54,14 @@ def __init__(self,
             message_type: MessageType,
             params: CommParams,
             is_server: bool,
-            device_ids: list[int] | None,
+            device_ids: list[int],
         ):
         """Initialize a message instance.
 
         :param message_type: Modbus message type
         :param params: parameter dataclass
         :param is_server: true if object act as a server (listen/connect)
-        :param device_ids: list of device id to accept (server only), None for all.
+        :param device_ids: list of device id to accept, 0 in list means broadcast.
         """
         super().__init__(params, is_server)
         self.device_ids = device_ids

pymodbus/message/rtu.py

@@ -6,6 +6,11 @@
 """
 from __future__ import annotations
 
+import struct
+
+from pymodbus.exceptions import ModbusIOException
+from pymodbus.factory import ClientDecoder
+from pymodbus.logging import Log
 from pymodbus.message.base import MessageBase
 
 
@@ -40,6 +45,13 @@ class MessageRTU(MessageBase):
     neither when receiving nor when sending.
     """
 
+    function_codes: list[int] = []
+
+    @classmethod
+    def set_legal_function_codes(cls, function_codes: list[int]):
+        """Set legal function codes."""
+        cls.function_codes = function_codes
+
     @classmethod
     def generate_crc16_table(cls) -> list[int]:
         """Generate a crc16 lookup table.
@@ -59,10 +71,116 @@ def generate_crc16_table(cls) -> list[int]:
         return result
     crc16_table: list[int] = [0]
 
-    def decode(self, _data: bytes) -> tuple[int, int, int, bytes]:
+    def _legacy_decode(self, callback, slave):  # noqa: C901
+        """Process new packet pattern."""
+
+        def is_frame_ready(self):
+            """Check if we should continue decode logic."""
+            size = self._header.get("len", 0)
+            if not size and len(self._buffer) > self._hsize:
+                try:
+                    self._header["uid"] = int(self._buffer[0])
+                    self._header["tid"] = int(self._buffer[0])
+                    func_code = int(self._buffer[1])
+                    pdu_class = self.decoder.lookupPduClass(func_code)
+                    size = pdu_class.calculateRtuFrameSize(self._buffer)
+                    self._header["len"] = size
+
+                    if len(self._buffer) < size:
+                        raise IndexError
+                    self._header["crc"] = self._buffer[size - 2 : size]
+                except IndexError:
+                    return False
+            return len(self._buffer) >= size if size > 0 else False
+
+        def get_frame_start(self, slaves, broadcast, skip_cur_frame):
+            """Scan buffer for a relevant frame start."""
+            start = 1 if skip_cur_frame else 0
+            if (buf_len := len(self._buffer)) < 4:
+                return False
+            for i in range(start, buf_len - 3):  # <slave id><function code><crc 2 bytes>
+                if not broadcast and self._buffer[i] not in slaves:
+                    continue
+                if (
+                    self._buffer[i + 1] not in self.function_codes
+                    and (self._buffer[i + 1] - 0x80) not in self.function_codes
+                ):
+                    continue
+                if i:
+                    self._buffer = self._buffer[i:]  # remove preceding trash.
+                return True
+            if buf_len > 3:
+                self._buffer = self._buffer[-3:]
+            return False
+
+        def check_frame(self):
+            """Check if the next frame is available."""
+            try:
+                self._header["uid"] = int(self._buffer[0])
+                self._header["tid"] = int(self._buffer[0])
+                func_code = int(self._buffer[1])
+                pdu_class = self.decoder.lookupPduClass(func_code)
+                size = pdu_class.calculateRtuFrameSize(self._buffer)
+                self._header["len"] = size
+
+                if len(self._buffer) < size:
+                    raise IndexError
+                self._header["crc"] = self._buffer[size - 2 : size]
+                frame_size = self._header["len"]
+                data = self._buffer[: frame_size - 2]
+                crc = self._header["crc"]
+                crc_val = (int(crc[0]) << 8) + int(crc[1])
+                return MessageRTU.check_CRC(data, crc_val)
+            except (IndexError, KeyError, struct.error):
+                return False
+
+        self._buffer = b''  # pylint: disable=attribute-defined-outside-init
+        broadcast = not slave[0]
+        skip_cur_frame = False
+        while get_frame_start(self, slave, broadcast, skip_cur_frame):
+            self._header: dict = {"uid": 0x00, "len": 0, "crc": b"\x00\x00"}  # pylint: disable=attribute-defined-outside-init
+            if not is_frame_ready(self):
+                Log.debug("Frame - not ready")
+                break
+            if not check_frame(self):
+                Log.debug("Frame check failed, ignoring!!")
+                # x = self._buffer
+                # self.resetFrame()
+                # self._buffer = x
+                skip_cur_frame = True
+                continue
+            start = 0x01  # self._hsize
+            end = self._header["len"] - 2
+            buffer = self._buffer[start:end]
+            if end > 0:
+                Log.debug("Getting Frame - {}", buffer, ":hex")
+                data = buffer
+            else:
+                data = b""
+            if (result := ClientDecoder().decode(data)) is None:
+                raise ModbusIOException("Unable to decode request")
+            result.slave_id = self._header["uid"]
+            result.transaction_id = self._header["tid"]
+            self._buffer = self._buffer[self._header["len"] :]  # pylint: disable=attribute-defined-outside-init
+            Log.debug("Frame advanced, resetting header!!")
+            callback(result)  # defer or push to a thread?
+
+
+    def decode(self, data: bytes) -> tuple[int, int, int, bytes]:
         """Decode message."""
+        resp = None
+        if len(data) < 4:
+            return 0, 0, 0, b''
+
+        def callback(result):
+            """Set result."""
+            nonlocal resp
+            resp = result
+
+        self._legacy_decode(callback, self.device_ids)
         return 0, 0, 0, b''
 
+
     def encode(self, data: bytes, device_id: int, _tid: int) -> bytes:
         """Decode message."""
         packet = device_id.to_bytes(1,'big') + data

test/message/test_message.py

@@ -68,8 +68,7 @@ async def test_message_build_send(self, msg):
 
     @pytest.mark.parametrize(
         ("dev_id", "res"), [
-        (None, True),
-        (0, True),
+        (0, False),
         (1, True),
         (2, False),
         ])
@@ -220,7 +219,7 @@ def test_encode(self, frame, frame_expected, data, dev_id, tid, inx1, inx2, inx3
             pytest.skip("Not supported")
         if frame == MessageTLS and (tid or dev_id):
             pytest.skip("Not supported")
-        frame_obj = frame(None, True)
+        frame_obj = frame([0], True)
         expected = frame_expected[inx1 + inx2 + inx3]
         encoded_data = frame_obj.encode(data, dev_id, tid)
         assert encoded_data == expected
@@ -324,7 +323,7 @@ async def test_decode_bad_crc(self, frame, data, exp_len):
         """Test encode method."""
         if frame == MessageRTU:
             pytest.skip("Waiting for implementation.")
-        frame_obj = frame(None, True)
+        frame_obj = frame([0], True)
         used_len, _, _, data = frame_obj.decode(data)
         assert used_len == exp_len
         assert not data