feat(s7commplus): add typed symbolic tag API

README.rst

@@ -113,8 +113,21 @@ PUT/GET enabled.
 * **Multi-variable read optimizer** -- merges scattered reads into minimal PDU
   exchanges with parallel dispatch
 * **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 addressing and live browsing** -- cache typed tag descriptors from
+  the PLC and read or write by name. Writes use the datatype and SymbolCRC from
+  the browse result; safe reads refresh once if a changed CRC reveals a layout
+  update::
+
+      tag = client.resolve_tag("DB1.Motor.Speed")
+      value = client.read_tag(tag.name)
+      client.write_tag(tag.name, b"\x41\x20\x00\x00")
+
+      results = client.read_tags(["DB1.Motor.Speed", "DB1.Motor.Running"])
+      for result in results:
+          if result.success:
+              print(result.tag.name, result.value)
+          else:
+              print(result.tag.name, result.error)
 * **Symbolic data subscriptions** -- monitor values using access sequences
   returned by ``browse()``::
 

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 .catalog import ArrayDimension, SymbolCatalog, SymbolicTag, TagResult
 from .client import DBWriteItem, SymbolicReadItem
 from .client import S7CommPlusClient as Client
 from .connection import S7CommPlusConnection
@@ -37,6 +38,7 @@
     "Alarm",
     "AlarmNotification",
     "AlarmText",
+    "ArrayDimension",
     "AsyncClient",
     "CPUState",
     "Client",
@@ -49,8 +51,11 @@
     "Server",
     "SubscriptionItem",
     "SubscriptionNotification",
+    "SymbolCatalog",
     "SymbolicReadItem",
+    "SymbolicTag",
     "Tag",
+    "TagResult",
     "block_interface_from_explore",
     "datablocks_from_explore",
     "decompress_blob",

s7commplus/async_client.py

@@ -7,7 +7,7 @@
 import logging
 import ssl
 import struct
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from typing import Any, Awaitable, Callable, Optional, TypeVar
 
 from snap7.error import S7ConnectionError, S7ProtocolError
@@ -17,11 +17,13 @@
 from .client import (
     DBWriteItem,
     SymbolicReadItem,
+    SymbolicWriteItem,
     _build_area_read_payload,
     _build_area_write_payload,
     _build_explore_payload,
     _build_explore_request,
     _build_invoke_payload,
+    _build_multi_symbolic_write_payload,
     _build_multi_symbolic_read_payload,
     _build_read_payload,
     _build_subscription_request,
@@ -32,7 +34,9 @@
     _parse_cpu_state,
     _parse_read_response,
     _parse_write_response,
+    _parse_write_response_errors,
 )
+from .catalog import SymbolCatalog, SymbolicTag, TagResult
 from .codec import (
     decode_header,
     encode_header,
@@ -116,6 +120,7 @@ def __init__(self) -> None:
         self._connected = False
         self._lock = asyncio.Lock()
         self._connect_params: Optional[dict[str, Any]] = None
+        self._symbol_catalog: Optional[SymbolCatalog] = None
 
         # V2+ IntegrityId tracking
         self._integrity_id_read: int = 0
@@ -197,6 +202,7 @@ async def connect(
             tls_key: Path to client private key (PEM)
             tls_ca: Path to CA certificate for PLC verification (PEM)
         """
+        self._symbol_catalog = None
         self._connect_params = {
             "host": host,
             "port": port,
@@ -503,6 +509,7 @@ async def disconnect(self) -> None:
         self._incoming_bio = None
         self._outgoing_bio = None
         self._oms_secret = None
+        self._symbol_catalog = None
         self._server_session_version = None
         self._session_setup_ok = False
         self._protection_level = None
@@ -798,6 +805,99 @@ async def write_symbolic(
         response = await self._send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
 
+    async def refresh_tag_catalog(self) -> SymbolCatalog:
+        """Browse the PLC and replace the cached symbolic tag catalog."""
+        self._symbol_catalog = SymbolCatalog.from_browse(await self.browse())
+        return self._symbol_catalog
+
+    def invalidate_tag_catalog(self) -> None:
+        """Discard cached browse metadata after a PLC layout change."""
+        self._symbol_catalog = None
+
+    async def resolve_tag(self, name: str) -> SymbolicTag:
+        """Resolve a browsed tag name to its typed symbolic descriptor."""
+        catalog = self._symbol_catalog or await self.refresh_tag_catalog()
+        return catalog.resolve(name)
+
+    async def read_tag(self, name: str) -> bytes:
+        """Read one symbolic tag by name, refreshing once if its CRC changed."""
+        result = (await self.read_tags([name]))[0]
+        if result.error is not None:
+            raise result.error
+        assert result.value is not None
+        return result.value
+
+    async def read_tags(self, names: Sequence[str]) -> list[TagResult]:
+        """Read names in one request and return a success/error for every item."""
+        if not names:
+            return []
+        tags = [await self.resolve_tag(name) for name in names]
+        values = await self.read_symbolic_multi([(tag.access_area, list(tag.lids), tag.symbol_crc) for tag in tags])
+        results = [
+            TagResult(tag=tag, value=value)
+            if value is not None
+            else TagResult(tag=tag, error=RuntimeError(f"Symbolic read failed for {tag.name!r}"))
+            for tag, value in zip(tags, values)
+        ]
+        retry_indices = [index for index, result in enumerate(results) if not result.success and result.tag.symbol_crc]
+        if not retry_indices:
+            return results
+
+        refreshed = await self.refresh_tag_catalog()
+        changed: list[tuple[int, SymbolicTag]] = []
+        for index in retry_indices:
+            try:
+                tag = refreshed.resolve(results[index].tag.name)
+            except KeyError:
+                continue
+            if tag.symbol_crc != results[index].tag.symbol_crc:
+                changed.append((index, tag))
+        if not changed:
+            return results
+
+        retry_values = await self.read_symbolic_multi([(tag.access_area, list(tag.lids), tag.symbol_crc) for _, tag in changed])
+        for (index, tag), value in zip(changed, retry_values):
+            results[index] = (
+                TagResult(tag=tag, value=value)
+                if value is not None
+                else TagResult(tag=tag, error=RuntimeError(f"Symbolic read failed for {tag.name!r} after CRC refresh"))
+            )
+        return results
+
+    async def write_tag(self, name: str, data: bytes) -> None:
+        """Write one symbolic tag by name using its resolved PValue datatype."""
+        result = (await self.write_tags({name: data}))[0]
+        if result.error is not None:
+            raise result.error
+
+    async def write_tags(self, values: Mapping[str, bytes]) -> list[TagResult]:
+        """Write names once and return per-item results without automatic retry."""
+        if not self._connected:
+            raise RuntimeError("Not connected")
+        if not values:
+            return []
+        tags = [await self.resolve_tag(name) for name in values]
+        unsupported = [tag.name for tag in tags if tag.datatype is None]
+        if unsupported:
+            raise ValueError(f"No S7CommPlus wire datatype mapping for: {', '.join(unsupported)}")
+        items: list[SymbolicWriteItem] = [
+            (tag.access_area, list(tag.lids), data, tag.symbol_crc, tag.datatype)
+            for tag, data in zip(tags, values.values())
+            if tag.datatype is not None
+        ]
+        payload = _build_multi_symbolic_write_payload(items, self._protocol_version)
+        response = await self._send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
+        try:
+            errors = _parse_write_response_errors(response, expected_count=len(tags))
+        except RuntimeError as error:
+            return [TagResult(tag=tag, error=error) for tag in tags]
+        return [
+            TagResult(tag=tag, error=RuntimeError(f"Symbolic write failed for {tag.name!r}: PLC error {errors[index]}"))
+            if index in errors
+            else TagResult(tag=tag)
+            for index, tag in enumerate(tags, 1)
+        ]
+
     async def list_datablocks(self) -> list[dict[str, Any]]:
         """List all datablocks on the PLC via EXPLORE.
 
@@ -870,6 +970,9 @@ async def browse(self) -> list[dict[str, Any]]:
                     "opt_bitoffset": v.opt_bitoffset,
                     "nonopt_address": v.nonopt_address,
                     "nonopt_bitoffset": v.nonopt_bitoffset,
+                    "symbol_crc": v.symbol_crc,
+                    "array_dimensions": v.array_dimensions,
+                    "string_length": v.string_length,
                 }
             )
         return variables

s7commplus/catalog.py

@@ -0,0 +1,137 @@
+"""Typed symbolic tag descriptors built from S7CommPlus browse metadata."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from collections.abc import Iterator
+from typing import Any, Optional
+
+from .protocol import DataType
+from .typeinfo import Softdatatype
+
+
+_WIRE_TYPES: dict[Softdatatype, DataType] = {
+    Softdatatype.BOOL: DataType.BOOL,
+    Softdatatype.BBOOL: DataType.BOOL,
+    Softdatatype.BYTE: DataType.BYTE,
+    Softdatatype.CHAR: DataType.BYTE,
+    Softdatatype.WORD: DataType.WORD,
+    Softdatatype.INT: DataType.INT,
+    Softdatatype.DWORD: DataType.DWORD,
+    Softdatatype.DINT: DataType.DINT,
+    Softdatatype.REAL: DataType.REAL,
+    Softdatatype.DATE: DataType.UINT,
+    Softdatatype.TIMEOFDAY: DataType.UDINT,
+    Softdatatype.TIME: DataType.DINT,
+    Softdatatype.S5TIME: DataType.WORD,
+    Softdatatype.DATEANDTIME: DataType.TIMESTAMP,
+    Softdatatype.STRING: DataType.S7STRING,
+    Softdatatype.LREAL: DataType.LREAL,
+    Softdatatype.ULINT: DataType.ULINT,
+    Softdatatype.LINT: DataType.LINT,
+    Softdatatype.LWORD: DataType.LWORD,
+    Softdatatype.USINT: DataType.USINT,
+    Softdatatype.UINT: DataType.UINT,
+    Softdatatype.UDINT: DataType.UDINT,
+    Softdatatype.SINT: DataType.SINT,
+    Softdatatype.WCHAR: DataType.UINT,
+    Softdatatype.WSTRING: DataType.WSTRING,
+    Softdatatype.LTIME: DataType.TIMESPAN,
+    Softdatatype.LTOD: DataType.ULINT,
+    Softdatatype.LDT: DataType.TIMESTAMP,
+}
+
+
+@dataclass(frozen=True)
+class ArrayDimension:
+    """One PLC array dimension."""
+
+    lower_bound: int
+    element_count: int
+
+
+@dataclass(frozen=True)
+class SymbolicTag:
+    """Resolved symbolic address and type metadata for one browsed tag."""
+
+    name: str
+    access_area: int
+    lids: tuple[int, ...]
+    softdatatype: Softdatatype
+    datatype: Optional[DataType]
+    symbol_crc: int = 0
+    array_dimensions: tuple[ArrayDimension, ...] = ()
+    string_length: int = 0
+    opt_address: int = 0
+    opt_bitoffset: int = 0
+    nonopt_address: int = 0
+    nonopt_bitoffset: int = 0
+
+    @classmethod
+    def from_browse(cls, item: dict[str, Any]) -> "SymbolicTag":
+        """Create a descriptor from one :meth:`Client.browse` result."""
+        name = str(item["name"])
+        parts = str(item["access_sequence"]).split(".")
+        if len(parts) < 2 or any(not part for part in parts):
+            raise ValueError(f"Tag {name!r} has an invalid access sequence")
+        try:
+            access_area, *lids = (int(part, 16) for part in parts)
+            softdatatype = Softdatatype[item["data_type"]]
+        except (KeyError, ValueError) as exc:
+            raise ValueError(f"Tag {name!r} has unsupported browse metadata") from exc
+
+        dimensions = tuple(ArrayDimension(int(lower), int(count)) for lower, count in item.get("array_dimensions", ()))
+        return cls(
+            name=name,
+            access_area=access_area,
+            lids=tuple(lids),
+            softdatatype=softdatatype,
+            datatype=_WIRE_TYPES.get(softdatatype),
+            symbol_crc=int(item.get("symbol_crc", 0)),
+            array_dimensions=dimensions,
+            string_length=int(item.get("string_length", 0)),
+            opt_address=int(item.get("opt_address", 0)),
+            opt_bitoffset=int(item.get("opt_bitoffset", 0)),
+            nonopt_address=int(item.get("nonopt_address", 0)),
+            nonopt_bitoffset=int(item.get("nonopt_bitoffset", 0)),
+        )
+
+
+@dataclass(frozen=True)
+class TagResult:
+    """Per-item result returned by a named batch operation."""
+
+    tag: SymbolicTag
+    value: Optional[bytes] = None
+    error: Optional[Exception] = None
+
+    @property
+    def success(self) -> bool:
+        return self.error is None
+
+
+class SymbolCatalog:
+    """An in-memory, name-indexed snapshot of PLC browse metadata."""
+
+    def __init__(self, tags: list[SymbolicTag]) -> None:
+        self._tags: dict[str, SymbolicTag] = {}
+        for tag in tags:
+            if tag.name in self._tags:
+                raise ValueError(f"Duplicate symbolic tag name: {tag.name!r}")
+            self._tags[tag.name] = tag
+
+    @classmethod
+    def from_browse(cls, variables: list[dict[str, Any]]) -> "SymbolCatalog":
+        return cls([SymbolicTag.from_browse(variable) for variable in variables])
+
+    def resolve(self, name: str) -> SymbolicTag:
+        try:
+            return self._tags[name]
+        except KeyError as exc:
+            raise KeyError(f"Unknown symbolic tag: {name!r}") from exc
+
+    def __len__(self) -> int:
+        return len(self._tags)
+
+    def __iter__(self) -> Iterator[SymbolicTag]:
+        return iter(self._tags.values())