Allow sub_function_code is custom PDU. (#2753)

Author: janiversen

pymodbus/pdu/decoders.py

@@ -26,12 +26,7 @@ def lookupPduClass(self, data: bytes) -> type[ModbusPDU] | None:
             return ExceptionResponse
         if not (pdu := self.pdu_table.get(func_code, (None, None))[self.pdu_inx]):
             return None
-
-        if func_code == 0x2B:  # mei message, sub_function_code is 1 byte
-            sub_func_code = int(data[2])
-        elif func_code == 0x08:  # diag message,  sub_function_code is 2 bytes
-            sub_func_code = int.from_bytes(data[2:4], "big")
-        else:
+        if (sub_func_code := pdu.decode_sub_function_code(data)) < 0:
             return pdu
         return self.pdu_sub_table[func_code].get(sub_func_code, (None, None))[self.pdu_inx]
 

pymodbus/pdu/diag_message.py

@@ -44,6 +44,11 @@ def encode(self) -> bytes:
             raise TypeError(f"UNKNOWN DIAG message type: {type(self.message)}")
         return packet
 
+    @classmethod
+    def decode_sub_function_code(cls, data: bytes) -> int:
+        """Decode sub function code (2 bytes)."""
+        return int.from_bytes(data[2:4], "big")
+
     def decode(self, data: bytes) -> None:
         """Decode a diagnostic request."""
         (self.sub_function_code, ) = struct.unpack(">H", data[:2])

pymodbus/pdu/mei_message.py

@@ -50,6 +50,11 @@ def encode(self) -> bytes:
         )
         return packet
 
+    @classmethod
+    def decode_sub_function_code(cls, data: bytes) -> int:
+        """Decode sub function code (1 byte)."""
+        return int(data[2])
+
     def decode(self, data: bytes) -> None:
         """Decode data part of the message."""
         self.sub_function_code, self.read_code, self.object_id = struct.unpack(">BBB", data[:3])
@@ -134,6 +139,11 @@ def encode(self) -> bytes:
         packet += objects
         return packet
 
+    @classmethod
+    def decode_sub_function_code(cls, data: bytes) -> int:
+        """Decode sub function code (1 byte)."""
+        return int(data[2])
+
     def decode(self, data: bytes) -> None:
         """Decode a the response."""
         params = struct.unpack(">BBBBBB", data[0:6])

pymodbus/pdu/pdu.py

@@ -58,6 +58,12 @@ def verifyAddress(self, address: int = -1) -> None:
         if not 0 <= address <= 65535:
             raise ValueError(f"0 < address {address} < 65535 !")
 
+    @classmethod
+    def decode_sub_function_code(cls, data: bytes) -> int:
+        """Decode sub function code."""
+        _ = data
+        return -1
+
     def __str__(self) -> str:
         """Build a representation of a Modbus response."""
         return (