feat(s7commplus): add symbolic data subscriptions

CHANGES.md

@@ -12,6 +12,7 @@ Major release: new `s7commplus` package with S7CommPlus protocol support.
 * S7CommPlus PLC start/stop via INVOKE
 * S7CommPlus object browsing via EXPLORE
 * S7CommPlus live symbol browsing (`client.browse()`) and datablock listing (experimental)
+* S7CommPlus symbolic data subscriptions and notification decoding (experimental)
 * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental)
 * S7CommPlus CPU state reading and block transfer (upload/download)
 * **Symbolic (LID-based) access for optimized DBs** (experimental):

README.rst

@@ -95,6 +95,22 @@ PUT/GET enabled.
 * **S7 routing** -- connect to PLCs on remote subnets via a gateway PLC
 * **Symbolic addressing** -- read/write by tag name instead of raw addresses
 * **Live symbol browsing** -- resolve tag names directly from the PLC
+* **Symbolic data subscriptions** -- monitor values using access sequences
+  returned by ``browse()``::
+
+      from s7commplus import Client
+
+      client = Client()
+      client.connect("192.168.1.10", 0, 1, password="secret")
+      subscription_id = client.create_subscription(["8A0E0007.A"], cycle_ms=100)
+      notification = client.receive_subscription_notification()
+      value = notification.values[1]
+      client.delete_subscription(subscription_id)
+      client.disconnect()
+
+  Reference IDs default to the one-based position of each access sequence.
+  Subscriptions use symbolic LIDs and therefore cannot be created from raw DB
+  byte offsets.
 * **TIA Portal XML import** -- import symbol tables from TIA Portal exports
 
 **Help us test!** If you have access to any Siemens S7 PLC, we would greatly

s7commplus/__init__.py

@@ -13,33 +13,39 @@
     data = client.db_read(1, 0, 4)
 """
 
+from .async_client import S7CommPlusAsyncClient as AsyncClient
 from .blob_decompressor import decompress_blob, find_and_decompress
 from .client import S7CommPlusClient as Client
-from .async_client import S7CommPlusAsyncClient as AsyncClient
-from .server import S7CommPlusServer as Server, DataBlock, CPUState
 from .connection import S7CommPlusConnection
+from .server import CPUState, DataBlock
+from .server import S7CommPlusServer as Server
+from .subscription import SubscriptionItem, SubscriptionNotification
 from .tag_browser import (
-    Tag,
-    Member,
     DataBlock as ExploreDataBlock,
-    tags_from_explore,
+)
+from .tag_browser import (
+    Member,
+    Tag,
     block_interface_from_explore,
     datablocks_from_explore,
+    tags_from_explore,
 )
 
 __all__ = [
-    "Client",
     "AsyncClient",
-    "Server",
-    "DataBlock",
     "CPUState",
+    "Client",
+    "DataBlock",
+    "ExploreDataBlock",
+    "Member",
     "S7CommPlusConnection",
-    "decompress_blob",
-    "find_and_decompress",
+    "Server",
+    "SubscriptionItem",
+    "SubscriptionNotification",
     "Tag",
-    "Member",
-    "ExploreDataBlock",
-    "tags_from_explore",
     "block_interface_from_explore",
     "datablocks_from_explore",
+    "decompress_blob",
+    "find_and_decompress",
+    "tags_from_explore",
 ]

s7commplus/client.py

@@ -5,7 +5,8 @@
 
 import logging
 import struct
-from typing import Any, Callable, Optional, TypeVar
+from collections.abc import Callable, Sequence
+from typing import Any, Optional, TypeVar
 
 from snap7.error import S7ConnectionError
 
@@ -16,9 +17,17 @@
     encode_item_address,
     encode_object_qualifier,
     encode_pvalue_blob,
+    parse_create_object_session_id,
 )
 from .connection import S7CommPlusConnection
 from .protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, ProtocolVersion
+from .subscription import (
+    SubscriptionItem,
+    SubscriptionNotification,
+    build_delete_subscription_request,
+    build_subscription_request,
+    parse_subscription_notification,
+)
 from .vlq import decode_uint32_vlq, decode_uint64_vlq, encode_uint32_vlq
 
 logger = logging.getLogger(__name__)
@@ -37,6 +46,8 @@ def __init__(self) -> None:
         # Last-used connect() arguments, kept so operations can transparently
         # reconnect on firmware that RSTs the session after a symbolic read.
         self._connect_params: Optional[dict[str, Any]] = None
+        self._subscription_change_counter = 1
+        self._subscription_relation_id = 0x7FFFC001
 
     @property
     def connected(self) -> bool:
@@ -614,31 +625,64 @@ def _explore_type_info_container(self) -> list["typeinfo.PObject"]:
         response = self._connection.send_request(FunctionCode.EXPLORE, payload, integrity_tail=5, reassemble=True)
         return typeinfo.extract_type_info_objects(response)
 
-    def create_subscription(self, items: list[tuple[int, int, int]], cycle_ms: int = 0) -> int:
+    def create_subscription(
+        self,
+        items: Sequence[SubscriptionItem | str],
+        cycle_ms: int = 100,
+        credit_limit: int = 10,
+    ) -> 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
-        ``receive_notification()`` to receive the pushed data.
+        The PLC pushes an initial value and subsequent changes. Access-sequence
+        strings are returned by :meth:`browse`; explicit
+        :class:`SubscriptionItem` objects can supply a symbol CRC, sub-area, or
+        stable reference ID.
 
         Args:
-            items: List of (db_number, start_offset, size) tuples to monitor.
-            cycle_ms: Cycle time in milliseconds (0 = on change).
+            items: Symbolic access-sequence strings or subscription items.
+            cycle_ms: Sampling cycle in milliseconds.
+            credit_limit: Number of notification credits. The default of 10
+                matches the value accepted by real S7-1500 PLCs.
 
         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
+        if self._connection.subscription_container_id == 0:
+            raise RuntimeError("PLC did not provide a subscription container object")
+
+        normalized = [SubscriptionItem.from_access_sequence(item) if isinstance(item, str) else item for item in items]
+        payload, integrity_tail = build_subscription_request(
+            self._connection.subscription_container_id,
+            normalized,
+            cycle_ms=cycle_ms,
+            credit_limit=credit_limit,
+            change_counter=self._subscription_change_counter,
+            relation_id=self._subscription_relation_id,
+        )
+        response = self._connection.send_request(
+            FunctionCode.CREATE_OBJECT,
+            payload,
+            integrity_tail=integrity_tail,
+        )
+        object_ids, _, return_value = parse_create_object_session_id(response)
+        if return_value != 0 or not object_ids:
+            raise RuntimeError(f"Subscription creation failed: PLC returned 0x{return_value:X}")
+
+        self._subscription_change_counter = self._subscription_change_counter % 0xFF + 1
+        self._subscription_relation_id = (self._subscription_relation_id + 1) & 0xFFFFFFFF
+        subscription_id = object_ids[0]
+        logger.info(f"Subscription created, id={subscription_id:#x}")
+        return subscription_id
+
+    def receive_subscription_notification(self) -> SubscriptionNotification:
+        """Block until the PLC sends one data-subscription notification."""
+        if self._connection is None:
+            raise RuntimeError("Not connected")
+        return parse_subscription_notification(self._connection.receive_notification())
 
     def delete_subscription(self, subscription_id: int) -> None:
         """Delete a data change subscription.
@@ -650,8 +694,12 @@ def delete_subscription(self, subscription_id: int) -> None:
         """
         if self._connection is None:
             raise RuntimeError("Not connected")
+        if self._connection.subscription_container_id == 0:
+            raise RuntimeError("PLC did not provide a subscription container object")
 
-        payload = struct.pack(">I", subscription_id) + struct.pack(">I", 0)
+        # Subscription children are owned by the session's second CreateObject
+        # result. The reference driver deletes that container, not the child ID.
+        payload = build_delete_subscription_request(self._connection.subscription_container_id, self._connection.protocol_version)
         self._connection.send_request(FunctionCode.DELETE_OBJECT, payload)
         logger.info(f"Subscription {subscription_id:#x} deleted")
 

s7commplus/connection.py

@@ -45,6 +45,7 @@
 import ssl
 import struct
 import tempfile
+from collections import deque
 from types import TracebackType
 from typing import Any, Optional, Type
 
@@ -293,6 +294,7 @@ def __init__(
         self._incoming_bio: Optional[ssl.MemoryBIO] = None
         self._outgoing_bio: Optional[ssl.MemoryBIO] = None
         self._session_id: int = 0
+        self._subscription_container_id: int = 0
         self._sequence_number: int = 0
         self._protocol_version: int = 0  # Detected from PLC response
         self._tls_active: bool = False
@@ -328,6 +330,7 @@ def __init__(
 
         # Password for post-auth legitimation (V1-initial PLCs)
         self._connect_password: str = ""
+        self._notification_frames: deque[bytes] = deque()
 
         # Effective protection level, read once the session is up
         self._protection_level: Optional[int] = None
@@ -346,6 +349,11 @@ def session_id(self) -> int:
         """Session ID assigned by the PLC."""
         return self._session_id
 
+    @property
+    def subscription_container_id(self) -> int:
+        """Object ID assigned to the session's subscription container."""
+        return self._subscription_container_id
+
     @property
     def tls_active(self) -> bool:
         """Whether TLS encryption is active on this connection."""
@@ -671,13 +679,15 @@ def disconnect(self) -> None:
         self._outgoing_bio = None
         self._oms_secret = None
         self._session_id = 0
+        self._subscription_container_id = 0
         self._sequence_number = 0
         self._protocol_version = 0
         self._server_session_version = None
         self._with_integrity_id = False
         self._integrity_id_read = 0
         self._integrity_id_write = 0
         self._protection_level = None
+        self._notification_frames.clear()
         self._iso_conn.disconnect()
 
     def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: int = 4, reassemble: bool = False) -> bytes:
@@ -785,6 +795,9 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
 
         # Receive response
         response_frame = self._recv_s7_data()
+        while self._is_notification_frame(response_frame):
+            self._notification_frames.append(response_frame)
+            response_frame = self._recv_s7_data()
         logger.debug(f"=== RECV RESPONSE === raw frame ({len(response_frame)} bytes): {response_frame.hex(' ')}")
 
         # Parse frame header, use data_length to exclude trailer
@@ -844,6 +857,39 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
 
         return resp_payload
 
+    @staticmethod
+    def _is_notification_frame(frame: bytes) -> bool:
+        """Return whether a complete frame contains an unsolicited notification."""
+        try:
+            version, data_length, consumed = decode_header(frame)
+        except (IndexError, ValueError):
+            return False
+        data = frame[consumed : consumed + data_length]
+        if version == ProtocolVersion.V3 and data:
+            hash_length = data[0]
+            if hash_length and len(data) > 1 + hash_length:
+                data = data[1 + hash_length :]
+        return bool(data) and data[0] == Opcode.NOTIFICATION
+
+    def receive_notification(self) -> bytes:
+        """Receive one unsolicited S7CommPlus notification frame.
+
+        Notifications observed while waiting for a request response are queued,
+        so callers do not lose updates when protocol traffic interleaves. This
+        method must not run concurrently with :meth:`send_request` because both
+        consume the same connection stream.
+        """
+        if not self._connected:
+            from snap7.error import S7ConnectionError
+
+            raise S7ConnectionError("Not connected")
+        frame = self._notification_frames.popleft() if self._notification_frames else self._recv_s7_data()
+        if not self._is_notification_frame(frame):
+            from snap7.error import S7ConnectionError
+
+            raise S7ConnectionError("Expected an S7CommPlus notification")
+        return frame
+
     # Sanity caps for fragment reassembly — generous vs. any real PLC EXPLORE response,
     # but bounded so a malformed/adversarial stream can't drive unbounded allocation.
     _MAX_REASSEMBLED_BYTES = 16 * 1024 * 1024
@@ -1084,6 +1130,7 @@ def _wstring_attr(attr_id: int, s: str) -> bytes:
 
         # First ObjectId is the new session id; second (if any) is for notifications.
         self._session_id = object_ids[0]
+        self._subscription_container_id = object_ids[1] if len(object_ids) > 1 else 0
         self._protocol_version = version
 
         logger.debug(

s7commplus/protocol.py

@@ -195,11 +195,17 @@ class Ids(IntEnum):
     # Subscription classes (for data change notifications)
     CLASS_SUBSCRIPTIONS = 255
     CLASS_SUBSCRIPTION = 1001
+    SUBSCRIPTION_MISSED_SENDINGS = 1002
+    SUBSCRIPTION_SUBSYSTEM_ERROR = 1003
+    SUBSCRIPTION_ROUTE_MODE = 1040
     SUBSCRIPTION_CYCLE_TIME = 1049
     SUBSCRIPTION_ACTIVE = 1041
     SUBSCRIPTION_CREDIT_LIMIT = 1053
     SUBSCRIPTION_REFERENCE_LIST = 1048
     SUBSCRIPTION_FUNCTION_CLASS_ID = 1082
+    SUBSCRIPTION_DISABLED = 1051
+    SUBSCRIPTION_COUNT = 1052
+    SUBSCRIPTION_TICKS = 1054
 
     # Alarm subscription
     ALARM_SUBSCRIPTION_REF_CLASS_RID = 2662