Add diagnostic buffer reading and data change subscriptions (#690)

Implements #683 (diagnostic buffer) and #684 (data subscriptions):

Diagnostic buffer (#683):
- read_diagnostic_buffer() on snap7.Client: reads SZL 0x00A0, parses
  20-byte entries with BCD timestamps into structured dicts
- Exposed through s7.Client.read_diagnostic_buffer() via legacy fallback

Data change subscriptions (#684, experimental):
- create_subscription(items, cycle_ms): builds a CREATE_OBJECT request
  with subscription-class attributes (cycle time, credit limit,
  reference list) — PLC pushes updates for monitored variables
- delete_subscription(subscription_id): sends DELETE_OBJECT to clean up
- Exposed through s7.Client with S7CommPlus requirement check

All subscription methods marked experimental. The subscription protocol
is modeled after S7CommPlusDriver's alarm subscription pattern, adapted
for data variable monitoring.

Closes #683, closes #684.

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: gijzelaerr

s7/_s7commplus_client.py

@@ -12,7 +12,7 @@
 from typing import Any, Optional
 
 from .connection import S7CommPlusConnection
-from .protocol import FunctionCode, Ids
+from .protocol import FunctionCode, Ids, ElementID, DataType, ObjectId
 from .vlq import encode_uint32_vlq, decode_uint32_vlq, decode_uint64_vlq
 from .codec import (
     encode_item_address,
@@ -281,6 +281,47 @@ def browse(self) -> list[dict[str, Any]]:
 
         return variables
 
+    def create_subscription(self, items: list[tuple[int, int, int]], cycle_ms: int = 0) -> int:
+        """Create a data change subscription.
+
+        .. warning:: This method is **experimental** and may change.
+
+        The PLC will push data updates for the specified variables. Use
+        :meth:`receive_notification` to receive the pushed data.
+
+        Args:
+            items: List of (db_number, start_offset, size) tuples to monitor.
+            cycle_ms: Cycle time in milliseconds (0 = on change).
+
+        Returns:
+            Subscription object ID assigned by the PLC.
+        """
+        if self._connection is None:
+            raise RuntimeError("Not connected")
+
+        payload = _build_subscription_request(items, cycle_ms, self._connection.session_id)
+        response = self._connection.send_request(FunctionCode.CREATE_OBJECT, payload)
+
+        # Parse the CreateObject response to get the subscription object ID
+        sub_id, consumed = decode_uint32_vlq(response, 0)
+        logger.info(f"Subscription created, id={sub_id:#x}")
+        return sub_id
+
+    def delete_subscription(self, subscription_id: int) -> None:
+        """Delete a data change subscription.
+
+        .. warning:: This method is **experimental** and may change.
+
+        Args:
+            subscription_id: ID returned by :meth:`create_subscription`.
+        """
+        if self._connection is None:
+            raise RuntimeError("Not connected")
+
+        payload = struct.pack(">I", subscription_id) + struct.pack(">I", 0)
+        self._connection.send_request(FunctionCode.DELETE_OBJECT, payload)
+        logger.info(f"Subscription {subscription_id:#x} deleted")
+
     def __enter__(self) -> "S7CommPlusClient":
         return self
 
@@ -714,3 +755,86 @@ def _parse_explore_fields(response: bytes, db_number: int, db_name: str) -> list
             continue
 
     return fields
+
+
+# ---------------------------------------------------------------------------
+# Subscription helpers (experimental)
+# ---------------------------------------------------------------------------
+
+_SUBSCRIPTION_RELATION_ID = 0x7FFFC001
+
+
+def _build_subscription_request(items: list[tuple[int, int, int]], cycle_ms: int, session_id: int) -> bytes:
+    """Build a CREATE_OBJECT request for a data change subscription.
+
+    The subscription object is modeled after the S7CommPlusDriver alarm
+    subscription pattern, adapted for data variable monitoring.
+
+    Args:
+        items: List of (db_number, start_offset, size) to monitor.
+        cycle_ms: Cycle time in milliseconds (0 = on change).
+        session_id: Current session ID.
+
+    Returns:
+        CREATE_OBJECT payload.
+    """
+    payload = bytearray()
+
+    # Session container
+    payload += struct.pack(">I", session_id)
+    payload += bytes([0x00, DataType.UDINT])
+    payload += encode_uint32_vlq(0)
+    payload += struct.pack(">I", 0)
+
+    # Start subscription object
+    payload += bytes([ElementID.START_OF_OBJECT])
+    payload += struct.pack(">I", ObjectId.GET_NEW_RID_ON_SERVER)
+    payload += encode_uint32_vlq(Ids.CLASS_SUBSCRIPTION)
+    payload += encode_uint32_vlq(0)
+    payload += encode_uint32_vlq(0)
+
+    # Subscription attributes
+    payload += bytes([ElementID.ATTRIBUTE])
+    payload += encode_uint32_vlq(Ids.OBJECT_VARIABLE_TYPE_NAME)
+    payload += bytes([0x00, DataType.WSTRING])
+    name = f"PySub_{_SUBSCRIPTION_RELATION_ID:#x}".encode("utf-8")
+    payload += encode_uint32_vlq(len(name))
+    payload += name
+
+    payload += bytes([ElementID.ATTRIBUTE])
+    payload += encode_uint32_vlq(Ids.SUBSCRIPTION_FUNCTION_CLASS_ID)
+    payload += bytes([0x00, DataType.USINT])
+    payload += bytes([0x02])
+
+    payload += bytes([ElementID.ATTRIBUTE])
+    payload += encode_uint32_vlq(Ids.SUBSCRIPTION_ACTIVE)
+    payload += bytes([0x00, DataType.BOOL])
+    payload += bytes([0x01])
+
+    payload += bytes([ElementID.ATTRIBUTE])
+    payload += encode_uint32_vlq(Ids.SUBSCRIPTION_CYCLE_TIME)
+    payload += bytes([0x00, DataType.UDINT])
+    payload += encode_uint32_vlq(cycle_ms)
+
+    payload += bytes([ElementID.ATTRIBUTE])
+    payload += encode_uint32_vlq(Ids.SUBSCRIPTION_CREDIT_LIMIT)
+    payload += bytes([0x00, DataType.INT])
+    payload += struct.pack(">h", 10)  # 10 credits
+
+    # Build reference list from items
+    ref_list = bytearray()
+    for db_number, start, size in items:
+        access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF)
+        ref_list += struct.pack(">I", access_area)
+
+    payload += bytes([ElementID.ATTRIBUTE])
+    payload += encode_uint32_vlq(Ids.SUBSCRIPTION_REFERENCE_LIST)
+    payload += bytes([0x10, DataType.UDINT])  # 0x10 = array
+    payload += encode_uint32_vlq(len(items))
+    payload += ref_list
+
+    # Close subscription object
+    payload += bytes([ElementID.TERMINATING_OBJECT])
+    payload += struct.pack(">I", 0)
+
+    return bytes(payload)

s7/client.py

@@ -285,6 +285,42 @@ def browse(self) -> list[dict[str, Any]]:
             raise RuntimeError("browse() requires S7CommPlus connection")
         return self._plus.browse()
 
+    def read_diagnostic_buffer(self) -> list[dict[str, Any]]:
+        """Read the PLC diagnostic buffer.
+
+        .. warning:: This method is **experimental** and may change.
+
+        Uses the legacy S7 protocol (SZL read).
+        """
+        if self._legacy is None:
+            raise RuntimeError("Not connected")
+        return self._legacy.read_diagnostic_buffer()
+
+    def create_subscription(self, items: list[tuple[int, int, int]], cycle_ms: int = 0) -> int:
+        """Create a data change subscription (S7CommPlus only).
+
+        .. warning:: This method is **experimental** and may change.
+
+        Args:
+            items: List of (db_number, start_offset, size) tuples.
+            cycle_ms: Cycle time in milliseconds (0 = on change).
+
+        Returns:
+            Subscription ID.
+        """
+        if self._plus is None:
+            raise RuntimeError("create_subscription() requires S7CommPlus connection")
+        return self._plus.create_subscription(items, cycle_ms)
+
+    def delete_subscription(self, subscription_id: int) -> None:
+        """Delete a data change subscription (S7CommPlus only).
+
+        .. warning:: This method is **experimental** and may change.
+        """
+        if self._plus is None:
+            raise RuntimeError("delete_subscription() requires S7CommPlus connection")
+        self._plus.delete_subscription(subscription_id)
+
     def __getattr__(self, name: str) -> Any:
         """Delegate unknown methods to the legacy client."""
         if name.startswith("_"):

snap7/client.py

@@ -1851,6 +1851,65 @@ def read_szl_list(self) -> bytes:
         # Return raw data
         return bytes(szl.Data[: szl.Header.LengthDR])
 
+    def read_diagnostic_buffer(self) -> list[dict[str, Any]]:
+        """Read the PLC diagnostic buffer.
+
+        .. warning:: This method is **experimental** and may change.
+
+        Returns a list of diagnostic entries, newest first. Each entry
+        is a dict with keys ``event_id``, ``timestamp``, and ``description``.
+
+        Returns:
+            List of diagnostic buffer entries.
+        """
+        # SZL ID 0x00A0, index 0 = diagnostic buffer
+        szl = self.read_szl(0x00A0, 0)
+        raw = bytes(szl.Data[: szl.Header.LengthDR])
+
+        entries: list[dict[str, Any]] = []
+        # Each diagnostic entry is 20 bytes
+        entry_size = 20
+        offset = 0
+        while offset + entry_size <= len(raw):
+            event_id = struct.unpack(">H", raw[offset : offset + 2])[0]
+
+            # BCD-encoded timestamp at offset 2..9
+            ts_bytes = raw[offset + 2 : offset + 10]
+            try:
+                ts = self._parse_bcd_timestamp(ts_bytes)
+            except Exception:
+                ts = None
+
+            # Additional info at offset 10..19
+            info = raw[offset + 10 : offset + entry_size]
+
+            entries.append(
+                {
+                    "event_id": event_id,
+                    "timestamp": ts,
+                    "info": info.hex(),
+                }
+            )
+            offset += entry_size
+
+        return entries
+
+    @staticmethod
+    def _parse_bcd_timestamp(data: bytes) -> datetime:
+        """Parse a BCD-encoded S7 timestamp (8 bytes) to datetime."""
+
+        def bcd(b: int) -> int:
+            return (b >> 4) * 10 + (b & 0x0F)
+
+        year = bcd(data[0])
+        year += 2000 if year < 90 else 1900
+        month = bcd(data[1])
+        day = bcd(data[2])
+        hour = bcd(data[3])
+        minute = bcd(data[4])
+        second = bcd(data[5])
+        return datetime(year, month, day, hour, minute, second)
+
     def iso_exchange_buffer(self, data: bytearray) -> bytearray:
         """
         Exchange raw ISO PDU.