fix: use the real PI services _GARB and _MODU for compress and copy RAM to ROM

CHANGES.md

@@ -33,6 +33,9 @@ Unreleased
 * `set_plc_datetime` and the session-password calls now accept the null acknowledgement (return code 0x0A, no data) that
   real PLCs send for services that return no data, instead of failing with "Object does not exist". The bundled
   server acknowledges these services the same way (#940).
+* Compress now uses the PI service `_GARB` and copy RAM to ROM uses file system `EP` with `_MODU`, matching real
+  engineering tools and native Snap7, instead of the invented `_MSZL`. The bundled server recognises the new
+  requests (#941).
 
 3.2.1
 -----

snap7/client.py

@@ -1917,7 +1917,7 @@ def compress(self, timeout: int) -> int:
         """
         Compress PLC memory.
 
-        Sends real S7 PLC_CONTROL protocol with PI service "_MSZL".
+        Sends real S7 PLC_CONTROL protocol with PI service "_GARB".
 
         Args:
             timeout: Timeout in milliseconds (used for receive timeout)
@@ -1940,7 +1940,7 @@ def copy_ram_to_rom(self, timeout: int = 0) -> int:
         """
         Copy RAM to ROM.
 
-        Sends real S7 PLC_CONTROL protocol with PI service "_MSZL" and file ID "P".
+        Sends real S7 PLC_CONTROL protocol with PI service "_MODU" and file system "EP".
 
         Args:
             timeout: Timeout in milliseconds (used for receive timeout)

snap7/s7protocol.py

@@ -114,6 +114,10 @@ def get_return_code_description(return_code: int) -> str:
 _HOT_START_PARAMS = bytes.fromhex("28000000000000fd000009") + b"P_PROGRAM"
 # Cold start: same as hot start but with block "C " before PI service
 _COLD_START_PARAMS = bytes.fromhex("28000000000000fd0002432009") + b"P_PROGRAM"
+# Compress: function(0x28) + 7 reserved + block count(0) + PI length(5) + "_GARB" (as native Snap7 TReqFunCompress)
+_COMPRESS_PARAMS = bytes.fromhex("28000000000000fd000005") + b"_GARB"
+# Copy RAM to ROM: like compress but with file system "EP" before PI service "_MODU" (native TReqFunCopyRamToRom)
+_COPY_RAM_TO_ROM_PARAMS = bytes.fromhex("28000000000000fd0002455005") + b"_MODU"
 
 
 class S7Protocol:
@@ -402,7 +406,7 @@ def build_plc_control_request(self, operation: str) -> bytes:
         Build PLC control request using the S7 PI service PDU format.
 
         Args:
-            operation: Control operation ('stop', 'hot_start', 'cold_start')
+            operation: Control operation ('stop', 'hot_start', 'cold_start', 'compress', 'copy_ram_to_rom')
 
         Returns:
             Complete S7 PDU for PLC control
@@ -411,6 +415,8 @@ def build_plc_control_request(self, operation: str) -> bytes:
             "stop": _STOP_PARAMS,
             "hot_start": _HOT_START_PARAMS,
             "cold_start": _COLD_START_PARAMS,
+            "compress": _COMPRESS_PARAMS,
+            "copy_ram_to_rom": _COPY_RAM_TO_ROM_PARAMS,
         }
 
         if operation not in params:
@@ -449,87 +455,23 @@ def build_compress_request(self) -> bytes:
         """
         Build PLC control request for memory compression.
 
-        Uses PI service "_MSZL" (compress memory).
+        Uses PI service "_GARB" (compress memory), as native Snap7 does.
 
         Returns:
             Complete S7 PDU for compress request
         """
-        # PI service command for compress
-        pi_service = b"_MSZL"
-
-        # Parameter section: function code + PI service
-        # Format: func(1) + unknown(7) + pi_len(1) + pi_service
-        param_data = (
-            struct.pack(
-                ">BBBBBBBBB",
-                S7Function.PLC_CONTROL,  # 0x28
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                len(pi_service),  # PI service length
-            )
-            + pi_service
-        )
-
-        header = struct.pack(
-            ">BBHHHH",
-            0x32,  # Protocol ID
-            S7PDUType.REQUEST,  # PDU type
-            0x0000,  # Reserved
-            self._next_sequence(),  # Sequence
-            len(param_data),  # Parameter length
-            0x0000,  # Data length
-        )
-
-        return header + param_data
+        return self.build_plc_control_request("compress")
 
     def build_copy_ram_to_rom_request(self) -> bytes:
         """
         Build PLC control request for copying RAM to ROM.
 
-        Uses PI service "_MSZL" with file system parameters.
+        Uses PI service "_MODU" with the "EP" file system, as native Snap7 does.
 
         Returns:
             Complete S7 PDU for copy RAM to ROM request
         """
-        # PI service command for copy RAM to ROM
-        # Uses EP parameter for target file system
-        pi_service = b"_MSZL"
-        file_id = b"P"  # P = passive file system (ROM)
-
-        # Parameter section with file system identifier
-        param_data = (
-            struct.pack(
-                ">BBBBBBBBB",
-                S7Function.PLC_CONTROL,  # 0x28
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                0x00,  # Reserved
-                len(file_id),  # File ID length
-                len(pi_service),  # PI service length
-            )
-            + file_id
-            + pi_service
-        )
-
-        header = struct.pack(
-            ">BBHHHH",
-            0x32,  # Protocol ID
-            S7PDUType.REQUEST,  # PDU type
-            0x0000,  # Reserved
-            self._next_sequence(),  # Sequence
-            len(param_data),  # Parameter length
-            0x0000,  # Data length
-        )
-
-        return header + param_data
+        return self.build_plc_control_request("copy_ram_to_rom")
 
     # ========================================================================
     # Block Transfer PDU Builders (Upload/Download)

snap7/server/__init__.py

@@ -1129,14 +1129,10 @@ def _handle_plc_control(self, request: Dict[str, Any], client_address: Tuple[str
             pi_service = params.get("pi_service", b"")
 
             # Check for PI service operations
-            if pi_service == b"_MSZL":
-                file_id = params.get("file_id", b"")
-                if file_id == b"P":
-                    # Copy RAM to ROM
-                    logger.info(f"Copy RAM to ROM requested from {client_address}")
-                else:
-                    # Compress memory
-                    logger.info(f"Compress memory requested from {client_address}")
+            if pi_service == b"_MODU":
+                logger.info(f"Copy RAM to ROM requested from {client_address}")
+            elif pi_service == b"_GARB":
+                logger.info(f"Compress memory requested from {client_address}")
             elif len(params) >= 2:
                 # Has restart type parameter - start operation
                 restart_type = params.get("restart_type", 1)
@@ -1411,9 +1407,10 @@ def _parse_request_parameters(self, param_data: bytes) -> Dict[str, Any]:
             if b"P_PROGRAM" in param_data:
                 restart_type = 2 if b"C " in param_data else 1
                 return {"function_code": function_code, "restart_type": restart_type, "pi_service": b"P_PROGRAM"}
-            elif b"_MSZL" in param_data:
-                file_id = b"P" if b"P_MSZL" in param_data else b""
-                return {"function_code": function_code, "pi_service": b"_MSZL", "file_id": file_id}
+            elif b"_MODU" in param_data:
+                return {"function_code": function_code, "pi_service": b"_MODU", "file_id": b"EP"}
+            elif b"_GARB" in param_data:
+                return {"function_code": function_code, "pi_service": b"_GARB"}
             elif b"_DELE" in param_data:
                 return {"function_code": function_code, "pi_service": b"_DELE"}
             if len(param_data) >= 2 and param_data[1] in (1, 2):

tests/test_s7protocol.py

@@ -274,6 +274,16 @@ def test_set_clock_year_range(self) -> None:
             with pytest.raises(ValueError, match="1990-2089"):
                 self.proto.build_set_clock_request(datetime(year, 1, 1))
 
+    def test_compress_request_matches_real_job(self) -> None:
+        # Parameter block sent by an engineering tool (CISA icsnpp-s7comm snap7.pcap): _GARB
+        request = self.proto.build_compress_request()
+        assert request[10:] == bytes.fromhex("28000000000000fd0000055f47415242")
+
+    def test_copy_ram_to_rom_request_matches_real_job(self) -> None:
+        # Parameter block sent by an engineering tool (CISA icsnpp-s7comm snap7.pcap): file system EP, _MODU
+        request = self.proto.build_copy_ram_to_rom_request()
+        assert request[10:] == bytes.fromhex("28000000000000fd00024550055f4d4f4455")
+
     def test_request_db_download(self) -> None:
         assert self.proto.build_download_request(0x41, 1, bytes(64)).hex() == (
             "320100000001002000001a00010000000000095f30413030303031500d31303030303634303030303238"