feat(s7commplus): add batched symbolic read

CHANGES.md

@@ -6,6 +6,7 @@ CHANGES
 
 Major release: new `s7commplus` package with S7CommPlus protocol support.
 
+* Validate batched symbolic read item coverage and preserve explicit PLC errors.
 * Fix classic S7 TPKT bounds and COTP Class 0 header validation in sync/async clients.
 * Reject incomplete reads and mismatched or missing read/write acknowledgements.
 * Size read/write chunks in whole elements and preserve ctypes write datatypes;

s7commplus/__init__.py

@@ -16,6 +16,7 @@
 from .async_client import S7CommPlusAsyncClient as AsyncClient
 from .alarm import Alarm, AlarmNotification, AlarmText, LanguageId
 from .blob_decompressor import decompress_blob, find_and_decompress
+from .client import DBWriteItem, SymbolicReadItem
 from .client import S7CommPlusClient as Client
 from .connection import S7CommPlusConnection
 from .server import CPUState, DataBlock
@@ -39,6 +40,7 @@
     "AsyncClient",
     "CPUState",
     "Client",
+    "DBWriteItem",
     "DataBlock",
     "ExploreDataBlock",
     "LanguageId",
@@ -47,6 +49,7 @@
     "Server",
     "SubscriptionItem",
     "SubscriptionNotification",
+    "SymbolicReadItem",
     "Tag",
     "block_interface_from_explore",
     "datablocks_from_explore",

s7commplus/async_client.py

@@ -7,17 +7,20 @@
 import logging
 import ssl
 import struct
+from collections.abc import Sequence
 from typing import Any, Optional
 
 from . import typeinfo
 from .blob_decompressor import find_and_decompress
 from .client import (
     DBWriteItem,
+    SymbolicReadItem,
     _build_area_read_payload,
     _build_area_write_payload,
     _build_explore_payload,
     _build_explore_request,
     _build_invoke_payload,
+    _build_multi_symbolic_read_payload,
     _build_read_payload,
     _build_subscription_request,
     _build_symbolic_read_payload,
@@ -710,6 +713,30 @@ async def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int
             raise RuntimeError("Symbolic read failed")
         return results[0]
 
+    async def read_symbolic_multi(self, items: Sequence[SymbolicReadItem]) -> list[Optional[bytes]]:
+        """Read multiple variables using S7CommPlus symbolic (LID-based) access.
+
+        .. warning:: This method is **experimental** and may change.
+
+        Args:
+            items: `(access_area, lids)` tuples, or three-tuples adding a
+                symbol CRC.
+
+        Returns:
+            One entry per requested item, in request order.
+
+        Raises:
+            RuntimeError: If the PLC does not answer every requested item.
+        """
+        if not items:
+            return []
+        payload = _build_multi_symbolic_read_payload(items, self._protocol_version)
+        response = await self._send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
+        results = _parse_read_response(response, expected_count=len(items))
+        if len(results) != len(items):
+            raise RuntimeError(f"Symbolic multi-read failed: PLC returned {len(results)} of {len(items)} items")
+        return results
+
     async def write_symbolic(
         self, access_area: int, lids: list[int], data: bytes, symbol_crc: int = 0, *, datatype: DataType = DataType.BLOB
     ) -> None:

s7commplus/client.py

@@ -44,6 +44,7 @@
 
 _T = TypeVar("_T")
 DBWriteItem: TypeAlias = tuple[int, int, bytes, DataType]
+SymbolicReadItem: TypeAlias = tuple[int, list[int]] | tuple[int, list[int], int]
 
 
 def _normalize_write_item(item: DBWriteItem) -> tuple[int, int, bytes, DataType]:
@@ -53,6 +54,14 @@ def _normalize_write_item(item: DBWriteItem) -> tuple[int, int, bytes, DataType]
     return db_number, start, data, DataType(datatype)
 
 
+def _normalize_symbolic_read_item(item: SymbolicReadItem) -> tuple[int, list[int], int]:
+    if len(item) == 2:
+        access_area, lids = item
+        return access_area, lids, 0
+    access_area, lids, symbol_crc = item
+    return access_area, lids, symbol_crc
+
+
 class S7CommPlusClient:
     """S7CommPlus client for S7-1200/1500 PLCs.
 
@@ -392,6 +401,33 @@ def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int = 0)
             raise RuntimeError("Symbolic read failed")
         return results[0]
 
+    def read_symbolic_multi(self, items: Sequence[SymbolicReadItem]) -> list[Optional[bytes]]:
+        """Read multiple variables using S7CommPlus symbolic (LID-based) access.
+
+        .. warning:: This method is **experimental** and may change.
+
+        Args:
+            items: ``(access_area, lids)`` tuples, or three-tuples adding a
+                symbol CRC.
+
+        Returns:
+            One entry per requested item, in request order.
+
+        Raises:
+            RuntimeError: If the PLC does not answer every requested item.
+        """
+        if self._connection is None:
+            raise RuntimeError("Not connected")
+        if not items:
+            return []
+
+        payload = _build_multi_symbolic_read_payload(items, self._connection.protocol_version)
+        response = self._connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
+        results = _parse_read_response(response, expected_count=len(items))
+        if len(results) != len(items):
+            raise RuntimeError(f"Symbolic multi-read failed: PLC returned {len(results)} of {len(items)} items")
+        return results
+
     def write_symbolic(
         self, access_area: int, lids: list[int], data: bytes, symbol_crc: int = 0, *, datatype: DataType = DataType.BLOB
     ) -> None:
@@ -866,7 +902,7 @@ def _build_read_payload(items: list[tuple[int, int, int]], protocol_version: int
     return bytes(payload)
 
 
-def _parse_read_response(response: bytes) -> list[Optional[bytes]]:
+def _parse_read_response(response: bytes, expected_count: Optional[int] = None) -> list[Optional[bytes]]:
     """Parse a GetMultiVariables response payload.
 
     Args:
@@ -898,6 +934,8 @@ def _parse_read_response(response: bytes) -> list[Optional[bytes]]:
             break
         raw_bytes, consumed = decode_pvalue_to_bytes(response, offset)
         offset += consumed
+        if expected_count is not None and (item_nr > expected_count or item_nr in values):
+            raise RuntimeError(f"Symbolic multi-read failed: unexpected or duplicate item {item_nr}")
         values[item_nr] = raw_bytes
 
     errors: dict[int, int] = {}
@@ -908,8 +946,13 @@ def _parse_read_response(response: bytes) -> list[Optional[bytes]]:
             break
         err_value, consumed = decode_uint64_vlq(response, offset)
         offset += consumed
+        if expected_count is not None and (err_item_nr > expected_count or err_item_nr in values or err_item_nr in errors):
+            raise RuntimeError(f"Symbolic multi-read failed: unexpected or duplicate item {err_item_nr}")
         errors[err_item_nr] = err_value
 
+    if expected_count is not None and len(values) + len(errors) != expected_count:
+        raise RuntimeError(f"Symbolic multi-read failed: PLC answered {len(values) + len(errors)} of {expected_count} items")
+
     max_item = max(max(values.keys(), default=0), max(errors.keys(), default=0))
     results: list[Optional[bytes]] = []
     for i in range(1, max_item + 1):
@@ -1111,29 +1154,49 @@ def _build_symbolic_read_payload(
     """Build a GetMultiVariables payload for symbolic (LID-based) access.
 
     Used for optimized block access on S7-1200/1500 where byte offsets
-    are unreliable.  The PLC navigates its symbol tree using the LIDs.
+    are unreliable. The PLC navigates its symbol tree using the LIDs.
 
     For DBs, ``access_sub_area`` is ``DB_VALUE_ACTUAL``.  For controller
     areas (M/I/Q), it's ``CONTROLLER_AREA_VALUE_ACTUAL``.
     """
-    # Determine sub-area based on access_area
-    if access_area >= 0x8A0E0000:
-        access_sub_area = Ids.DB_VALUE_ACTUAL
-    else:
-        access_sub_area = Ids.CONTROLLER_AREA_VALUE_ACTUAL
+    return _build_multi_symbolic_read_payload([(access_area, lids, symbol_crc)], protocol_version=protocol_version)
 
-    addr_bytes, field_count = encode_item_address(
-        access_area=access_area,
-        access_sub_area=access_sub_area,
-        lids=lids,
-        symbol_crc=symbol_crc,
-    )
+
+def _build_multi_symbolic_read_payload(items: Sequence[SymbolicReadItem], protocol_version: int = ProtocolVersion.V2) -> bytes:
+    """Build a GetMultiVariables payload for reading multiple symbolic LID addresses at once.
+
+    Used for optimized block access on S7-1200/1500 where byte offsets
+    are unreliable. The PLC navigates its symbol tree using the LIDs.
+
+    For DBs, ``access_sub_area`` is ``DB_VALUE_ACTUAL``.  For controller
+    areas (M/I/Q), it's ``CONTROLLER_AREA_VALUE_ACTUAL``.
+
+    Args:
+        items: List of ``(access_area, lids)`` tuples, or three-tuples adding a
+            symbol CRC, one per variable.
+
+    Returns:
+        Encoded payload bytes.
+    """
+    addresses: list[bytes] = []
+    total_field_count = 0
+    for access_area, lids, symbol_crc in (_normalize_symbolic_read_item(item) for item in items):
+        access_sub_area = Ids.DB_VALUE_ACTUAL if access_area >= 0x8A0E0000 else Ids.CONTROLLER_AREA_VALUE_ACTUAL
+        addr_bytes, field_count = encode_item_address(
+            access_area=access_area,
+            access_sub_area=access_sub_area,
+            lids=lids,
+            symbol_crc=symbol_crc,
+        )
+        addresses.append(addr_bytes)
+        total_field_count += field_count
 
     payload = bytearray()
     payload += struct.pack(">I", 0)
-    payload += encode_uint32_vlq(1)  # one item
-    payload += encode_uint32_vlq(field_count)
-    payload += addr_bytes
+    payload += encode_uint32_vlq(len(items))
+    payload += encode_uint32_vlq(total_field_count)
+    for addr in addresses:
+        payload += addr
     payload += encode_object_qualifier(protocol_version=protocol_version)
     payload += struct.pack(">I", 0)
     return bytes(payload)

tests/test_s7_codec.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from s7commplus.client import _build_symbolic_read_payload, _build_symbolic_write_payload
+from s7commplus.client import _build_multi_symbolic_read_payload, _build_symbolic_read_payload, _build_symbolic_write_payload
 from s7commplus.codec import (
     _pvalue_element_size,
     decode_float32,
@@ -389,6 +389,22 @@ def test_db_area_still_uses_db_sub_area(self) -> None:
         assert encode_uint32_vlq(Ids.DB_VALUE_ACTUAL) in payload
 
 
+class TestMultiSymbolicReadPayload:
+    """A batched symbolic read packs every address into one GetMultiVariables request."""
+
+    DB_AREA = 0x8A0E0001
+
+    def test_single_item_matches_scalar_builder(self) -> None:
+        assert _build_multi_symbolic_read_payload([(self.DB_AREA, [1, 4], 0)]) == _build_symbolic_read_payload(
+            self.DB_AREA, [1, 4]
+        )
+
+    def test_symbol_crc_is_optional(self) -> None:
+        assert _build_multi_symbolic_read_payload([(self.DB_AREA, [1, 4])]) == _build_multi_symbolic_read_payload(
+            [(self.DB_AREA, [1, 4], 0)]
+        )
+
+
 class TestPValueBlob:
     def test_basic_blob(self) -> None:
         data = bytes([1, 2, 3, 4])

tests/test_s7_server.py

@@ -156,6 +156,19 @@ def test_read_multiple_values(self, server: S7CommPlusServer) -> None:
         finally:
             client.disconnect()
 
+    def test_read_symbolic_multi(self, server: S7CommPlusServer) -> None:
+        client = S7CommPlusClient()
+        client.connect("127.0.0.1", port=TEST_PORT)
+        try:
+            # LIDs are (1-based offset, size): temperature at 0, pressure at 4
+            results = client.read_symbolic_multi([(0x8A0E0001, [1, 4]), (0x8A0E0001, [5, 4])])
+            assert len(results) == 2
+            assert results[0] is not None and results[1] is not None
+            assert abs(struct.unpack(">f", results[0])[0] - 23.5) < 0.001
+            assert abs(struct.unpack(">f", results[1])[0] - 1.013) < 0.001
+        finally:
+            client.disconnect()
+
     def test_write_and_read_back(self, server: S7CommPlusServer) -> None:
         client = S7CommPlusClient()
         client.connect("127.0.0.1", port=TEST_PORT)

tests/test_s7_symbolic_multi_errors.py

@@ -0,0 +1,63 @@
+"""Batched symbolic read response accounting."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from s7commplus.async_client import S7CommPlusAsyncClient
+from s7commplus.client import S7CommPlusClient
+from s7commplus.codec import encode_pvalue_blob
+from s7commplus.protocol import ProtocolVersion
+
+
+ITEMS = [(0x8A0E0001, [1, 4]), (0x8A0E0001, [5, 4])]
+
+
+@pytest.mark.parametrize("async_client", [False, True])
+@pytest.mark.parametrize(
+    ("response", "expected"),
+    [
+        (
+            b"\x00\x02" + encode_pvalue_blob(b"second") + b"\x01" + encode_pvalue_blob(b"first") + b"\x00\x00",
+            [b"first", b"second"],
+        ),
+        (b"\x00\x01" + encode_pvalue_blob(b"first") + b"\x00\x02\x01\x00", [b"first", None]),
+    ],
+)
+async def test_symbolic_multi_preserves_item_order_and_errors(
+    async_client: bool, response: bytes, expected: list[bytes | None]
+) -> None:
+    if async_client:
+        client = S7CommPlusAsyncClient()
+        client._send_request = AsyncMock(return_value=response)
+        assert await client.read_symbolic_multi(ITEMS) == expected
+        client._send_request.assert_awaited_once()
+    else:
+        sync = S7CommPlusClient()
+        sync._connection = MagicMock(protocol_version=ProtocolVersion.V2)
+        sync._connection.send_request.return_value = response
+        assert sync.read_symbolic_multi(ITEMS) == expected
+        sync._connection.send_request.assert_called_once()
+
+
+@pytest.mark.parametrize("async_client", [False, True])
+@pytest.mark.parametrize(
+    "response",
+    [
+        b"\x00\x02" + encode_pvalue_blob(b"second") + b"\x00\x00",  # unanswered first item
+        b"\x00\x03" + encode_pvalue_blob(b"extra") + b"\x00\x00",  # out of range
+        b"\x00\x01" + encode_pvalue_blob(b"first") + b"\x01" + encode_pvalue_blob(b"duplicate") + b"\x00\x00",
+        b"\x01",  # whole-request rejection
+    ],
+)
+async def test_symbolic_multi_rejects_incomplete_or_invalid_replies(async_client: bool, response: bytes) -> None:
+    with pytest.raises(RuntimeError, match="Symbolic multi-read failed"):
+        if async_client:
+            client = S7CommPlusAsyncClient()
+            client._send_request = AsyncMock(return_value=response)
+            await client.read_symbolic_multi(ITEMS)
+        else:
+            sync = S7CommPlusClient()
+            sync._connection = MagicMock(protocol_version=ProtocolVersion.V2)
+            sync._connection.send_request.return_value = response
+            sync.read_symbolic_multi(ITEMS)

tests/test_s7_unit.py

@@ -618,6 +618,11 @@ def test_db_write_multi_uses_one_substreamed_request_per_item(self) -> None:
             ]
         )
 
+    def test_read_symbolic_multi_not_connected(self) -> None:
+        client = S7CommPlusClient()
+        with pytest.raises(RuntimeError, match="Not connected"):
+            client.read_symbolic_multi([(0x8A0E0001, [1, 4])])
+
     def test_explore_not_connected(self) -> None:
         client = S7CommPlusClient()
         with pytest.raises(RuntimeError, match="Not connected"):