Read the real CPU state from SZL 0x0424

CHANGES.md

@@ -1,6 +1,18 @@
 CHANGES
 ========
 
+Unreleased
+----------
+
+### Bug fixes
+
+* `get_cpu_state()` now reads the CPU's operating state from SZL 0x0424, in both clients. Previously it sent a read
+  request with no address and returned `"S7CpuStatusRun"` whatever the PLC answered. Startup, HOLD and DEFECT are
+  reported as `"S7CpuStatusStop"`, as native Snap7 does. A PLC that refuses SZL 0x0424 now raises `S7ProtocolError`;
+  the heartbeat of the synchronous client counts such a refusal as a live connection. Not yet tested on hardware.
+* The server answers SZL 0x0424 from its CPU state, and refuses a read request that has no address specification.
+  Previously it answered such a request with the first four bytes of DB1.
+
 3.4.0
 -----
 

snap7/async_client.py

@@ -23,7 +23,7 @@
 from .client_base import ClientMixin, _instrumented
 from .tags import Tag, decode_tag, encode_tag
 from .client import Client
-from .szl import parse_cp_info_szl, parse_cpu_info_szl, parse_order_code_szl, parse_protection_szl
+from .szl import parse_cp_info_szl, parse_cpu_info_szl, parse_cpu_state_szl, parse_order_code_szl, parse_protection_szl
 from .client import _parse_force_szl
 from .rate_limiter import RateLimitAlgorithm, RateLimitBehavior, RequestRateLimiter
 from .type import (
@@ -1115,10 +1115,16 @@ async def get_cpu_info(self) -> S7CpuInfo:
         return parse_cpu_info_szl(await self.read_szl(0x001C, 0))
 
     async def get_cpu_state(self) -> str:
-        """Get CPU state (running/stopped)."""
-        request = self.protocol.build_cpu_state_request()
-        response = await self._send_receive(request)
-        return self.protocol.extract_cpu_state(response)
+        """Get the CPU operating state by reading SZL 0x0424.
+
+        Returns ``"S7CpuStatusRun"``, ``"S7CpuStatusStop"`` or ``"S7CpuStatusUnknown"``.
+        Startup, HOLD and DEFECT are reported as ``"S7CpuStatusStop"``, as native Snap7 does.
+
+        Raises:
+            S7ConnectionError: If the client is not connected.
+            S7ProtocolError: If the PLC refuses SZL 0x0424 or its reply is too short.
+        """
+        return parse_cpu_state_szl(await self.read_szl(0x0424, 0))
 
     # ---------------------------------------------------------------
     # Upload / Download / Delete

snap7/client.py

@@ -24,14 +24,14 @@
 from .connection import ISOTCPConnection
 from .s7protocol import S7Protocol, S7UserDataGroup, S7UserDataSubfunction, get_return_code_description
 from .datatypes import S7DataTypes, S7WordLen
-from .error import S7Error, S7ConnectionError, S7ProtocolError, S7StalePacketError, S7TimeoutError
+from .error import S7Error, S7ConnectionError, S7PacketLostError, S7ProtocolError, S7StalePacketError, S7TimeoutError
 from .client_base import ClientMixin, _instrumented
 from .log import PLCLoggerAdapter, OperationLogger
 from .optimizer import ReadItem, ReadPacket, sort_items, merge_items, packetize, extract_results
 from .rate_limiter import RateLimitAlgorithm, RateLimitBehavior, RequestRateLimiter
 from .tags import Tag, decode_tag, encode_tag
 
-from .szl import parse_cp_info_szl, parse_cpu_info_szl, parse_order_code_szl, parse_protection_szl
+from .szl import parse_cp_info_szl, parse_cpu_info_szl, parse_cpu_state_szl, parse_order_code_szl, parse_protection_szl
 from .type import (
     Area,
     Block,
@@ -421,7 +421,7 @@ def _heartbeat_loop(self) -> None:
             try:
                 with self._reconnect_lock:
                     if self.connected and self.connection is not None:
-                        self.get_cpu_state()
+                        self._heartbeat_probe()
                         self._is_alive = True
             except Exception as e:
                 logger.warning(f"Heartbeat probe failed: {e}")
@@ -434,6 +434,20 @@ def _heartbeat_loop(self) -> None:
                     except S7ConnectionError:
                         logger.warning("Heartbeat reconnection failed")
 
+    def _heartbeat_probe(self) -> None:
+        """Ask the PLC for its CPU state to prove the connection is alive.
+
+        A PLC that answers with a refusal (for example a CPU without
+        SZL 0x0424) has still answered, so only a transport failure or a lost
+        or stale reply is raised.
+        """
+        try:
+            self.get_cpu_state()
+        except (S7PacketLostError, S7StalePacketError):
+            raise
+        except (S7ProtocolError, RuntimeError) as e:
+            logger.debug(f"Heartbeat probe was refused, connection is alive: {e}")
+
     def connect(self, address: str, rack: int, slot: int, tcp_port: int = 102) -> "Client":
         """
         Connect to S7 PLC.
@@ -1387,15 +1401,18 @@ def get_cpu_info(self) -> S7CpuInfo:
 
     def get_cpu_state(self) -> str:
         """
-        Get CPU state (running/stopped).
+        Get the CPU operating state by reading SZL 0x0424.
 
         Returns:
-            CPU state string
-        """
-        request = self.protocol.build_cpu_state_request()
-        response = self._send_receive(request)
+            ``"S7CpuStatusRun"``, ``"S7CpuStatusStop"`` or ``"S7CpuStatusUnknown"``.
+            Startup, HOLD and DEFECT are reported as ``"S7CpuStatusStop"``, as
+            native Snap7 does.
 
-        return self.protocol.extract_cpu_state(response)
+        Raises:
+            S7ConnectionError: If the client is not connected.
+            S7ProtocolError: If the PLC refuses SZL 0x0424 or its reply is too short.
+        """
+        return parse_cpu_state_szl(self.read_szl(0x0424, 0))
 
     def get_block_info(self, block_type: Block, db_number: int) -> TS7BlockInfo:
         """

snap7/s7protocol.py

@@ -1487,12 +1487,16 @@ def check_userdata_response(
 
     def build_cpu_state_request(self) -> bytes:
         """
-        Build CPU state request.
+        Build the request ``get_cpu_state()`` used to send.
+
+        .. deprecated:: 3.5
+           This is a read request without an address specification, which is
+           not a valid S7 PDU. The clients read SZL 0x0424 instead; see
+           :func:`snap7.szl.parse_cpu_state_szl`. Kept for backwards compatibility.
 
         Returns:
-            Complete S7 PDU for CPU state query
+            S7 PDU for the old CPU state query
         """
-        # Simple CPU state request - in real S7 this would be a userdata function
         header = struct.pack(
             ">BBHHHH",
             0x32,  # Protocol ID
@@ -1503,24 +1507,26 @@ def build_cpu_state_request(self) -> bytes:
             0x0000,  # Data length
         )
 
-        # Use a custom function code for CPU state
-        parameters = struct.pack(">B", 0x04)  # Use READ_AREA function for simplicity
+        parameters = struct.pack(">B", 0x04)  # READ_AREA function code, nothing else
 
         return header + parameters
 
     def extract_cpu_state(self, response: Dict[str, Any]) -> str:
         """
-        Extract CPU state from response.
+        Return ``"S7CpuStatusRun"`` whatever the response holds.
+
+        .. deprecated:: 3.5
+           This never looked at the response. The clients read SZL 0x0424
+           instead; see :func:`snap7.szl.parse_cpu_state_szl`. Kept for
+           backwards compatibility.
 
         Args:
-            response: Parsed S7 response
+            response: Parsed S7 response (ignored)
 
         Returns:
-            CPU state string in S7CpuStatus format (e.g., 'S7CpuStatusRun')
+            Always ``"S7CpuStatusRun"``
         """
-        # Map internal states to S7 status format for API compatibility with master branch
-        # The cpu_statuses dict in type.py uses: {0: "S7CpuStatusUnknown", 4: "S7CpuStatusStop", 8: "S7CpuStatusRun"}
-        return "S7CpuStatusRun"  # Default state for pure Python server
+        return "S7CpuStatusRun"
 
     def parse_response(self, pdu: bytes) -> Dict[str, Any]:
         """

snap7/server/__init__.py

@@ -1021,9 +1021,9 @@ def _parse_read_address(self, request: Dict[str, Any]) -> Optional[Tuple[S7Area,
                 )
                 return (area, db_number, start, byte_count)
 
-            # Fallback to defaults if parsing failed
-            logger.warning("Using default address values - address parsing may have failed")
-            return (S7Area.DB, 1, 0, 4)
+            # A read request without an address specification is malformed
+            logger.warning("Read request carries no address specification")
+            return None
 
         except Exception as e:
             logger.error(f"Error parsing read address: {e}")
@@ -1837,6 +1837,7 @@ def _handle_szl(self, request: Dict[str, Any], userdata_params: Dict[str, Any],
         - 0x001C: Component identification (for get_cpu_info)
         - 0x0011: Module identification (for get_order_code)
         - 0x0131: Communication parameters (for get_cp_info)
+        - 0x0424: Current operating mode transition (for get_cpu_state)
         - 0x0232: Protection level (for get_protection)
 
         Args:
@@ -1933,10 +1934,19 @@ def _get_szl_data(self, szl_id: int, szl_index: int) -> Optional[bytes]:
             # anl_sch: factory setting protection
             return struct.pack(">HHHHH", 1, 0, 0, 0, 0)  # No protection
 
+        # SZL 0x0424: current operating mode transition (get_cpu_state)
+        # Record layout: ereig(2) + ae(1) + bzu-id(1) + res(4) + anlinfo1..4(4) + time(8) = 20 bytes.
+        # bzu-id carries the requested mode in bits 0-3: 4 is STOP, 8 is RUN.
+        elif szl_id == 0x0424:
+            # Event IDs: 0x4302 "STARTUP to RUN", 0x4304 "STOP caused by PG STOP operation"
+            event_ids = {CPUState.RUN: 0x4302, CPUState.STOP: 0x4304}
+            record = struct.pack(">HBB", event_ids.get(self.cpu_state, 0x0000), 0xFF, int(self.cpu_state)) + bytes(16)
+            return struct.pack(">HH", len(record), 1) + record
+
         # SZL 0x0000: SZL list
         elif szl_id == 0x0000:
             # Return list of available SZL IDs
-            available_ids = [0x0000, 0x0011, 0x001C, 0x0131, 0x0232]
+            available_ids = [0x0000, 0x0011, 0x001C, 0x0131, 0x0232, 0x0424]
             szl_bytes = b""
             for id_val in available_ids:
                 szl_bytes += struct.pack(">H", id_val)

snap7/szl.py

@@ -14,6 +14,7 @@
 
 import struct
 
+from .error import S7ProtocolError
 from .type import S7CpInfo, S7CpuInfo, S7OrderCode, S7Protection, S7SZL
 
 
@@ -56,6 +57,41 @@ def parse_cpu_info_szl(szl: S7SZL) -> S7CpuInfo:
     return info
 
 
+def parse_cpu_state_szl(szl: S7SZL) -> str:
+    """Decode SZL 0x0424 (current operating mode transition) into an ``S7CpuStatus*`` string.
+
+    The payload is the 4-byte partial-list header (``LENTHDR``, ``N_DR``)
+    followed by one 20-byte record::
+
+        ereig(2)  ae(1)  bzu-id(1)  res(4)  anlinfo1..4(4)  time(8)
+
+    ``bzu-id`` holds the requested operating mode in bits 0-3 and the previous
+    one in bits 4-7 (1-4 are STOP variants, 5 and 7 are startup, 8 is RUN,
+    0xA is HOLD, 0xD is DEFECT). The layout follows Wireshark's S7comm
+    dissector (``packet-s7comm_szl_ids.c``) and matches the byte native Snap7
+    reads in ``opGetPlcStatus``.
+
+    Like native Snap7, only RUN is reported as ``S7CpuStatusRun``, a zero byte
+    as ``S7CpuStatusUnknown``, and every other mode (startup, HOLD and DEFECT
+    included) as ``S7CpuStatusStop``. Snap7 compares the whole byte against
+    0x08; this looks at the requested-mode nibble only, so a RUN reply that
+    also carries a previous mode is still reported as RUN.
+
+    Raises:
+        S7ProtocolError: If the reply is too short to hold the mode byte.
+    """
+    data = _szl_data(szl)
+    if len(data) < 8:
+        raise S7ProtocolError(f"SZL 0x0424 reply too short for an operating mode: {len(data)} bytes")
+
+    mode = data[7]
+    if mode == 0x00:
+        return "S7CpuStatusUnknown"
+    if mode & 0x0F == 0x08:
+        return "S7CpuStatusRun"
+    return "S7CpuStatusStop"
+
+
 def parse_cp_info_szl(szl: S7SZL) -> S7CpInfo:
     """Decode SZL 0x0131 (communication processor info) into an :class:`S7CpInfo`.
 

tests/test_async_client.py

@@ -406,9 +406,13 @@ async def test_download_collects_all_fragments(client: AsyncClient) -> None:
 
 
 @pytest.mark.asyncio
-async def test_get_cpu_state(client: AsyncClient) -> None:
-    state = await client.get_cpu_state()
-    assert isinstance(state, str)
+async def test_get_cpu_state(client: AsyncClient, server: Server) -> None:
+    try:
+        for status, expected in ((4, "S7CpuStatusStop"), (0, "S7CpuStatusUnknown"), (8, "S7CpuStatusRun")):
+            server.set_cpu_status(status)
+            assert await client.get_cpu_state() == expected
+    finally:
+        server.set_cpu_status(8)
 
 
 @pytest.mark.asyncio

tests/test_server.py

@@ -17,7 +17,8 @@
 from snap7.error import S7ConnectionError, S7ProtocolError, error_text, server_errors
 from snap7.s7protocol import S7Function, S7PDUType
 from snap7.server import EVC_DATA_READ, EVC_DATA_WRITE, EVC_SERVER_STARTED, EVC_SERVER_STOPPED, Server, ServerISOConnection
-from snap7.type import Block, Parameter, SrvArea, SrvEvent, mkEvent, mkLog
+from snap7.szl import parse_cpu_state_szl
+from snap7.type import S7SZL, Block, Parameter, SrvArea, SrvEvent, mkEvent, mkLog
 
 logging.basicConfig(level=logging.WARNING)
 
@@ -1074,10 +1075,76 @@ def test_set_plc_system_datetime(self) -> None:
     # ------------------------------------------------------------------
     # get_cpu_state (SZL-based CPU state request)
     # ------------------------------------------------------------------
-    def test_get_cpu_state(self) -> None:
-        """get_cpu_state should return a string state."""
-        state = self.client.get_cpu_state()
-        self.assertIsInstance(state, str)
+    def test_get_cpu_state_follows_server(self) -> None:
+        """get_cpu_state reports the state the server is in, read from SZL 0x0424."""
+        try:
+            for status, expected in ((4, "S7CpuStatusStop"), (0, "S7CpuStatusUnknown"), (8, "S7CpuStatusRun")):
+                self.server.set_cpu_status(status)
+                self.assertEqual(self.client.get_cpu_state(), expected)
+        finally:
+            self.server.set_cpu_status(8)
+
+    def test_szl_0424_record_layout(self) -> None:
+        """SZL 0x0424 is a partial-list header plus one 20-byte record with the mode at offset 3."""
+        szl = self.client.read_szl(0x0424, 0)
+        data = bytes(b & 0xFF for b in szl.Data[: szl.Header.LengthDR])
+        self.assertEqual(len(data), 24)
+        self.assertEqual(struct.unpack(">HH", data[:4]), (20, 1))
+        self.assertEqual(data[4:8], bytes([0x43, 0x02, 0xFF, 0x08]))
+
+    def test_read_without_address_is_refused(self) -> None:
+        """A read request with no address specification gets an error, not DB1 defaults."""
+        request = self.client.protocol.build_cpu_state_request()
+        with self.assertRaises(S7ProtocolError):
+            self.client._send_receive(request)
+        # The connection survives the refusal
+        self.assertEqual(len(self.client.db_read(1, 0, 4)), 4)
+
+    def test_heartbeat_probe(self) -> None:
+        """A refused probe still proves the link; a transport failure does not."""
+        self.client._heartbeat_probe()
+        with patch.object(self.client, "get_cpu_state", side_effect=S7ProtocolError("refused")):
+            self.client._heartbeat_probe()
+        with patch.object(self.client, "get_cpu_state", side_effect=S7ConnectionError("gone")):
+            with self.assertRaises(S7ConnectionError):
+                self.client._heartbeat_probe()
+
+
+@pytest.mark.server
+class TestParseCpuStateSzl(unittest.TestCase):
+    """parse_cpu_state_szl maps the bzu-id byte of SZL 0x0424 to an S7CpuStatus string."""
+
+    @staticmethod
+    def _szl(mode: int) -> S7SZL:
+        payload = struct.pack(">HH", 20, 1) + bytes([0x43, 0x02, 0xFF, mode]) + bytes(16)
+        szl = S7SZL()
+        szl.Header.LengthDR = len(payload)
+        szl.Header.NDR = 1
+        for i, b in enumerate(payload):
+            szl.Data[i] = b
+        return szl
+
+    def test_modes(self) -> None:
+        cases = {
+            0x00: "S7CpuStatusUnknown",
+            0x08: "S7CpuStatusRun",
+            0x58: "S7CpuStatusRun",  # previous mode startup, requested mode RUN
+            0x03: "S7CpuStatusStop",  # STOP (self initialization), seen on older CPUs
+            0x04: "S7CpuStatusStop",
+            0x84: "S7CpuStatusStop",  # previous mode RUN, requested mode STOP
+            0x05: "S7CpuStatusStop",  # startup
+            0x0A: "S7CpuStatusStop",  # HOLD
+            0x0D: "S7CpuStatusStop",  # DEFECT
+        }
+        for mode, expected in cases.items():
+            with self.subTest(mode=hex(mode)):
+                self.assertEqual(parse_cpu_state_szl(self._szl(mode)), expected)
+
+    def test_short_reply_is_an_error(self) -> None:
+        szl = self._szl(0x08)
+        szl.Header.LengthDR = 7
+        with self.assertRaises(S7ProtocolError):
+            parse_cpu_state_szl(szl)
 
 
 @pytest.mark.server