diff --git a/README.rst b/README.rst index 233ceefe..6882e14d 100644 --- a/README.rst +++ b/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()``:: diff --git a/s7commplus/__init__.py b/s7commplus/__init__.py index 1181f88c..29365f4c 100644 --- a/s7commplus/__init__.py +++ b/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", diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 1a54774f..21249443 100644 --- a/s7commplus/async_client.py +++ b/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 diff --git a/s7commplus/catalog.py b/s7commplus/catalog.py new file mode 100644 index 00000000..b879022d --- /dev/null +++ b/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()) diff --git a/s7commplus/client.py b/s7commplus/client.py index b1879452..ef1cf86d 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -5,7 +5,7 @@ import logging import struct -from collections.abc import Callable, Sequence +from collections.abc import Callable, Mapping, Sequence from typing import Any, Optional, TypeAlias, TypeVar from snap7.error import S7ConnectionError, S7ProtocolError @@ -29,6 +29,7 @@ encode_pvalue_typed, parse_create_object_session_id, ) +from .catalog import SymbolCatalog, SymbolicTag, TagResult from .connection import S7CommPlusConnection from .protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, ProtocolVersion from .subscription import ( @@ -45,6 +46,7 @@ _T = TypeVar("_T") DBWriteItem: TypeAlias = tuple[int, int, bytes, DataType] SymbolicReadItem: TypeAlias = tuple[int, list[int]] | tuple[int, list[int], int] +SymbolicWriteItem: TypeAlias = tuple[int, list[int], bytes, int, DataType] def _normalize_write_item(item: DBWriteItem) -> tuple[int, int, bytes, DataType]: @@ -75,6 +77,7 @@ def __init__(self) -> None: self._connect_params: Optional[dict[str, Any]] = None self._subscription_change_counter = 1 self._subscription_relation_id = 0x7FFFC001 + self._symbol_catalog: Optional[SymbolCatalog] = None @property def connected(self) -> bool: @@ -140,6 +143,7 @@ def connect( tls_ca: Path to CA certificate for PLC verification (PEM) password: PLC password for legitimation (V2+ with TLS) """ + self._symbol_catalog = None self._connect_params = { "host": host, "port": port, @@ -201,6 +205,7 @@ def disconnect(self) -> None: self._connection.disconnect() self._connection = None self._connect_params = None + self._symbol_catalog = None def db_read(self, db_number: int, start: int, size: int) -> bytes: """Read raw bytes from a data block. @@ -460,6 +465,108 @@ def write_symbolic( response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload) _parse_write_response(response) + def refresh_tag_catalog(self) -> SymbolCatalog: + """Browse the PLC and replace the cached symbolic tag catalog.""" + self._symbol_catalog = SymbolCatalog.from_browse(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 + + def resolve_tag(self, name: str) -> SymbolicTag: + """Resolve a browsed tag name to its typed symbolic descriptor.""" + catalog = self._symbol_catalog or self.refresh_tag_catalog() + return catalog.resolve(name) + + def read_tag(self, name: str) -> bytes: + """Read one symbolic tag by name, refreshing once if its CRC changed.""" + result = self.read_tags([name])[0] + if result.error is not None: + raise result.error + assert result.value is not None + return result.value + + def read_tags(self, names: Sequence[str]) -> list[TagResult]: + """Read names in one request and return a success/error for every item. + + A failed read with a non-zero SymbolCRC causes one catalog refresh. Only + tags whose CRC actually changed are re-resolved and safely retried. + """ + if not names: + return [] + tags = [self.resolve_tag(name) for name in names] + values = 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 = 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 = 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 + + def write_tag(self, name: str, data: bytes) -> None: + """Write one symbolic tag by name using its resolved PValue datatype.""" + result = self.write_tags({name: data})[0] + if result.error is not None: + raise result.error + + def write_tags(self, values: Mapping[str, bytes]) -> list[TagResult]: + """Write names in one request and return a success/error per item. + + Writes are deliberately never retried: a transport failure can leave + the caller unable to know whether the PLC applied the request. + """ + if self._connection is None: + raise RuntimeError("Not connected") + if not values: + return [] + tags = [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._connection.protocol_version) + response = self._connection.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) + ] + def explore(self, explore_id: int = 0) -> bytes: """Browse the PLC object tree. @@ -678,6 +785,9 @@ 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 @@ -1007,15 +1117,20 @@ def _parse_write_response(response: bytes) -> None: Raises: RuntimeError: If the write failed """ - offset = 0 + errors = _parse_write_response_errors(response) + if errors: + err_str = ", ".join(f"item {nr}: error {val}" for nr, val in errors.items()) + raise RuntimeError(f"Write failed: {err_str}") - return_value, consumed = decode_uint64_vlq(response, offset) - offset += consumed +def _parse_write_response_errors(response: bytes, expected_count: Optional[int] = None) -> dict[int, int]: + """Return the per-item PLC errors in a SetMultiVariables response.""" + return_value, consumed = decode_uint64_vlq(response, 0) + offset = consumed if return_value != 0: raise RuntimeError(f"Write failed with return value {return_value}") - errors: list[tuple[int, int]] = [] + errors: dict[int, int] = {} while offset < len(response): err_item_nr, consumed = decode_uint32_vlq(response, offset) offset += consumed @@ -1023,11 +1138,10 @@ def _parse_write_response(response: bytes) -> None: break err_value, consumed = decode_uint64_vlq(response, offset) offset += consumed - errors.append((err_item_nr, err_value)) - - if errors: - err_str = ", ".join(f"item {nr}: error {val}" for nr, val in errors) - raise RuntimeError(f"Write failed: {err_str}") + if expected_count is not None and (err_item_nr > expected_count or err_item_nr in errors): + raise RuntimeError(f"Symbolic multi-write failed: unexpected or duplicate item {err_item_nr}") + errors[err_item_nr] = err_value + return errors def _build_substreamed_read_payload(session_id: int, access_area: int, access_sub_area: int, lids: list[int]) -> bytes: @@ -1211,25 +1325,35 @@ def _build_symbolic_write_payload( datatype: DataType = DataType.BLOB, ) -> bytes: """Build a SetMultiVariables payload for symbolic (LID-based) access.""" - if access_area >= 0x8A0E0000: - access_sub_area = Ids.DB_VALUE_ACTUAL - else: - access_sub_area = 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, + return _build_multi_symbolic_write_payload( + [(access_area, lids, data, symbol_crc, datatype)], protocol_version=protocol_version ) + +def _build_multi_symbolic_write_payload(items: Sequence[SymbolicWriteItem], protocol_version: int = ProtocolVersion.V2) -> bytes: + """Build one typed SetMultiVariables payload for symbolic addresses.""" + addresses: list[bytes] = [] + total_field_count = 0 + for access_area, lids, _data, symbol_crc, _datatype in items: + access_sub_area = Ids.DB_VALUE_ACTUAL if access_area >= 0x8A0E0000 else Ids.CONTROLLER_AREA_VALUE_ACTUAL + address, field_count = encode_item_address( + access_area=access_area, + access_sub_area=access_sub_area, + lids=lids, + symbol_crc=symbol_crc, + ) + addresses.append(address) + total_field_count += field_count + payload = bytearray() payload += struct.pack(">I", 0) - payload += encode_uint32_vlq(1) - payload += encode_uint32_vlq(field_count) - payload += addr_bytes - payload += encode_uint32_vlq(1) # item number 1 - payload += encode_pvalue_typed(datatype, data) + payload += encode_uint32_vlq(len(items)) + payload += encode_uint32_vlq(total_field_count) + for address in addresses: + payload += address + for index, (_access_area, _lids, data, _symbol_crc, datatype) in enumerate(items, 1): + payload += encode_uint32_vlq(index) + payload += encode_pvalue_typed(datatype, data) payload += bytes([0x00]) payload += encode_object_qualifier(protocol_version=protocol_version) payload += struct.pack(">I", 0) diff --git a/s7commplus/typeinfo.py b/s7commplus/typeinfo.py index 8d27e2ff..0d227f09 100644 --- a/s7commplus/typeinfo.py +++ b/s7commplus/typeinfo.py @@ -565,6 +565,9 @@ class VarInfo: opt_bitoffset: int = 0 nonopt_address: int = 0 nonopt_bitoffset: int = 0 + symbol_crc: int = 0 + array_dimensions: tuple[tuple[int, int], ...] = () + string_length: int = 0 def _tcom_size(obj: PObject | None) -> int: @@ -713,11 +716,30 @@ def build_flat_list(root_nodes: list[Node]) -> list[VarInfo]: for root in root_nodes: if not root.children: continue - _walk(root, "", "", 0, 0, result) + _walk(root, "", "", 0, 0, (), result) return result -def _walk(node: Node, names: str, access_ids: str, opt_off: int, nonopt_off: int, result: list[VarInfo]) -> None: +def _array_dimensions(vte: VartypeListElement | None) -> tuple[tuple[int, int], ...]: + if vte is None: + return () + oi = vte.offset_info + if oi.is_mdim: + return tuple((lower, count) for lower, count in zip(oi.mdim_lower_bounds, oi.mdim_element_count) if count > 0) + if oi.is_1dim: + return ((oi.array_lower_bound, oi.array_element_count),) + return () + + +def _walk( + node: Node, + names: str, + access_ids: str, + opt_off: int, + nonopt_off: int, + array_dimensions: tuple[tuple[int, int], ...], + result: list[VarInfo], +) -> None: # Accumulate this node's name and access-id contribution. if node.node_type == NodeType.ROOT: names = names + node.name @@ -732,6 +754,9 @@ def _walk(node: Node, names: str, access_ids: str, opt_off: int, nonopt_off: int names = names + "." + node.name access_ids = access_ids + "." + f"{node.access_id:X}" + if node.node_type in (NodeType.ARRAY, NodeType.STRUCT_ARRAY): + array_dimensions += _array_dimensions(node.vte) + if node.children: # Descend into a branch — advance the running byte offsets. if node.node_type == NodeType.ARRAY: @@ -750,14 +775,22 @@ def _walk(node: Node, names: str, access_ids: str, opt_off: int, nonopt_off: int if child.node_type == NodeType.ARRAY: child_opt += child.array_adr_offset_opt child_nonopt += child.array_adr_offset_nonopt - _walk(child, names, access_ids, child_opt, child_nonopt, result) + _walk(child, names, access_ids, child_opt, child_nonopt, array_dimensions, result) return # Leaf node — emit if the datatype is a readable leaf. if not is_softdatatype_supported(node.softdatatype): return - info = VarInfo(name=names, access_sequence=access_ids, softdatatype=node.softdatatype) + vte = node.vte + info = VarInfo( + name=names, + access_sequence=access_ids, + softdatatype=node.softdatatype, + symbol_crc=vte.symbol_crc if vte is not None else 0, + array_dimensions=array_dimensions, + string_length=vte.offset_info.unspecified1 if vte is not None else 0, + ) if node.node_type == NodeType.ARRAY: # Basic-array element: offset already includes the element stride. info.opt_address = opt_off @@ -767,7 +800,6 @@ def _walk(node: Node, names: str, access_ids: str, opt_off: int, nonopt_off: int info.opt_address = opt_off + node.vte.offset_info.opt_addr info.nonopt_address = nonopt_off + node.vte.offset_info.nonopt_addr - vte = node.vte if node.softdatatype == Softdatatype.BOOL and vte is not None: info.opt_bitoffset = vte.attribute_bitoffset info.nonopt_bitoffset = vte.nonopt_bitoffset if vte.classic else vte.attribute_bitoffset diff --git a/tests/test_s7_tag_catalog.py b/tests/test_s7_tag_catalog.py new file mode 100644 index 00000000..1d613323 --- /dev/null +++ b/tests/test_s7_tag_catalog.py @@ -0,0 +1,163 @@ +"""Typed, name-based S7CommPlus tag catalog tests.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from snap7.error import S7ConnectionError +from s7commplus.async_client import S7CommPlusAsyncClient +from s7commplus.catalog import ArrayDimension, SymbolCatalog, SymbolicTag +from s7commplus.client import S7CommPlusClient, _build_multi_symbolic_write_payload +from s7commplus.protocol import DataType, ProtocolVersion +from s7commplus.typeinfo import Softdatatype +from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq + + +def _browse_item( + name: str = "DB1.Motor.Speed", + *, + crc: int = 0x12345678, + access_sequence: str = "8A0E0001.A.2", + data_type: str = "REAL", +) -> dict[str, object]: + return { + "name": name, + "access_sequence": access_sequence, + "data_type": data_type, + "symbol_crc": crc, + "array_dimensions": ((1, 4),), + "string_length": 0, + "opt_address": 12, + "opt_bitoffset": 0, + "nonopt_address": 20, + "nonopt_bitoffset": 0, + } + + +class TestSymbolCatalog: + def test_descriptor_preserves_address_type_crc_and_array_metadata(self) -> None: + catalog = SymbolCatalog.from_browse([_browse_item()]) + + tag = catalog.resolve("DB1.Motor.Speed") + + assert tag.access_area == 0x8A0E0001 + assert tag.lids == (0xA, 0x2) + assert tag.softdatatype is Softdatatype.REAL + assert tag.datatype is DataType.REAL + assert tag.symbol_crc == 0x12345678 + assert tag.array_dimensions == (ArrayDimension(1, 4),) + assert tag.opt_address == 12 + assert tag.nonopt_address == 20 + + def test_unknown_name_has_clear_error(self) -> None: + with pytest.raises(KeyError, match="Unknown symbolic tag"): + SymbolCatalog([]).resolve("missing") + + def test_duplicate_name_is_rejected(self) -> None: + with pytest.raises(ValueError, match="Duplicate symbolic tag"): + SymbolCatalog.from_browse([_browse_item(), _browse_item()]) + + +class TestNamedTagIO: + def test_resolve_catalog_is_cached(self) -> None: + client = S7CommPlusClient() + client.browse = MagicMock(return_value=[_browse_item()]) # type: ignore[method-assign] + + assert client.resolve_tag("DB1.Motor.Speed") is client.resolve_tag("DB1.Motor.Speed") + client.browse.assert_called_once() + + def test_read_tags_returns_per_item_results(self) -> None: + client = S7CommPlusClient() + client._symbol_catalog = SymbolCatalog.from_browse( + [_browse_item("DB1.Good", crc=0), _browse_item("DB1.Bad", crc=0, access_sequence="8A0E0001.B")] + ) + client.read_symbolic_multi = MagicMock(return_value=[b"\x00\x01", None]) # type: ignore[method-assign] + + results = client.read_tags(["DB1.Good", "DB1.Bad"]) + + assert results[0].success and results[0].value == b"\x00\x01" + assert not results[1].success and results[1].error is not None + + def test_failed_read_refreshes_and_retries_only_when_crc_changed(self) -> None: + client = S7CommPlusClient() + client.browse = MagicMock( # type: ignore[method-assign] + side_effect=[[_browse_item(crc=1, access_sequence="8A0E0001.A")], [_browse_item(crc=2, access_sequence="8A0E0001.B")]] + ) + client.read_symbolic_multi = MagicMock(side_effect=[[None], [b"\x40\x49\x0f\xdb"]]) # type: ignore[method-assign] + + assert client.read_tag("DB1.Motor.Speed") == b"\x40\x49\x0f\xdb" + assert client.read_symbolic_multi.call_args_list[0].args[0] == [(0x8A0E0001, [0xA], 1)] + assert client.read_symbolic_multi.call_args_list[1].args[0] == [(0x8A0E0001, [0xB], 2)] + + def test_failed_read_is_not_retried_when_crc_is_unchanged(self) -> None: + client = S7CommPlusClient() + client.browse = MagicMock(return_value=[_browse_item(crc=1)]) # type: ignore[method-assign] + client.read_symbolic_multi = MagicMock(return_value=[None]) # type: ignore[method-assign] + + with pytest.raises(RuntimeError, match="Symbolic read failed"): + client.read_tag("DB1.Motor.Speed") + client.read_symbolic_multi.assert_called_once() + + def test_write_uses_resolved_datatype_and_reports_item_errors(self) -> None: + client = S7CommPlusClient() + client._connection = MagicMock(protocol_version=ProtocolVersion.V2) + client._symbol_catalog = SymbolCatalog.from_browse( + [_browse_item("DB1.Real"), _browse_item("DB1.Count", data_type="DINT", access_sequence="8A0E0001.B")] + ) + client._connection.send_request.return_value = ( + encode_uint64_vlq(0) + encode_uint32_vlq(2) + encode_uint64_vlq(0xDEAD) + encode_uint32_vlq(0) + ) + + with patch("s7commplus.client._build_multi_symbolic_write_payload", wraps=_build_multi_symbolic_write_payload) as build: + results = client.write_tags({"DB1.Real": b"\x3f\x80\x00\x00", "DB1.Count": b"\x00\x00\x00\x01"}) + + items = build.call_args.args[0] + assert [item[4] for item in items] == [DataType.REAL, DataType.DINT] + assert results[0].success + assert not results[1].success + + def test_write_is_never_retried_after_ambiguous_transport_failure(self) -> None: + client = S7CommPlusClient() + client._connection = MagicMock(protocol_version=ProtocolVersion.V2) + client._symbol_catalog = SymbolCatalog.from_browse([_browse_item()]) + client._connection.send_request.side_effect = S7ConnectionError("connection lost") + + with pytest.raises(S7ConnectionError, match="connection lost"): + client.write_tag("DB1.Motor.Speed", b"\x3f\x80\x00\x00") + client._connection.send_request.assert_called_once() + + def test_global_plc_write_error_is_reported_for_every_item(self) -> None: + client = S7CommPlusClient() + client._connection = MagicMock(protocol_version=ProtocolVersion.V2) + client._symbol_catalog = SymbolCatalog.from_browse( + [_browse_item("DB1.Real"), _browse_item("DB1.Count", data_type="DINT", access_sequence="8A0E0001.B")] + ) + client._connection.send_request.return_value = encode_uint64_vlq(0x05A9) + + results = client.write_tags({"DB1.Real": b"\x3f\x80\x00\x00", "DB1.Count": b"\x00\x00\x00\x01"}) + + assert len(results) == 2 + assert all(not result.success and result.error is not None for result in results) + + +@pytest.mark.asyncio +async def test_async_read_tag_refreshes_changed_crc_once() -> None: + client = S7CommPlusAsyncClient() + client.browse = AsyncMock( # type: ignore[method-assign] + side_effect=[[_browse_item(crc=1)], [_browse_item(crc=2, access_sequence="8A0E0001.B")]] + ) + client.read_symbolic_multi = AsyncMock(side_effect=[[None], [b"ok"]]) # type: ignore[method-assign] + + assert await client.read_tag("DB1.Motor.Speed") == b"ok" + assert client.read_symbolic_multi.await_count == 2 + + +def test_public_descriptor_can_be_constructed_directly() -> None: + tag = SymbolicTag( + name="DB1.Value", + access_area=0x8A0E0001, + lids=(1,), + softdatatype=Softdatatype.INT, + datatype=DataType.INT, + ) + assert tag.name == "DB1.Value" diff --git a/tests/test_typeinfo.py b/tests/test_typeinfo.py index 45048f2b..f0446a19 100644 --- a/tests/test_typeinfo.py +++ b/tests/test_typeinfo.py @@ -333,9 +333,21 @@ def test_object_list_two_siblings(self) -> None: assert off == len(a + b) -def _vte(lid: int, sdt, oi: "ti.OffsetInfo", attr_flags: int = 0, bitoff: int = 0) -> "ti.VartypeListElement": +def _vte( + lid: int, + sdt, + oi: "ti.OffsetInfo", + attr_flags: int = 0, + bitoff: int = 0, + symbol_crc: int = 0, +) -> "ti.VartypeListElement": return ti.VartypeListElement( - lid=lid, symbol_crc=0, softdatatype=int(sdt), attribute_flags=attr_flags, bitoffsetinfo_flags=bitoff, offset_info=oi + lid=lid, + symbol_crc=symbol_crc, + softdatatype=int(sdt), + attribute_flags=attr_flags, + bitoffsetinfo_flags=bitoff, + offset_info=oi, ) @@ -396,6 +408,7 @@ def test_basic_array_elements(self) -> None: 9, ti.Softdatatype.INT, ti.OffsetInfo(code=10, opt_addr=20, nonopt_addr=40, array_element_count=3, array_lower_bound=0, is_1dim=True), + symbol_crc=0x12345678, ) ], varname_list=["Vals"], @@ -405,6 +418,8 @@ def test_basic_array_elements(self) -> None: assert [v.name for v in infos] == ["DB1.Vals[0]", "DB1.Vals[1]", "DB1.Vals[2]"] assert [v.access_sequence for v in infos] == ["8A0E0001.9.0", "8A0E0001.9.1", "8A0E0001.9.2"] assert [v.opt_address for v in infos] == [20, 22, 24] # base + i*2 (INT stride) + assert all(v.symbol_crc == 0x12345678 for v in infos) + assert all(v.array_dimensions == ((0, 3),) for v in infos) def test_struct_array_inserts_extra_one(self) -> None: root = _root("DB1", 0x8A0E0001, 0x100) @@ -440,6 +455,7 @@ def test_struct_array_inserts_extra_one(self) -> None: # StructArray inserts a ".1" between the array index id and the member LID. assert [v.access_sequence for v in infos] == ["8A0E0001.7.0.1.2", "8A0E0001.7.1.1.2"] assert [v.opt_address for v in infos] == [0, 8] # element stride 8 + assert all(v.array_dimensions == ((0, 2),) for v in infos) class TestMDimArrays: @@ -486,6 +502,7 @@ def test_basic_mdim_array_ordering_bounds_and_offsets(self) -> None: ] assert [v.opt_address for v in infos] == [20, 22, 24, 26, 28, 30] # base + (n-1)*2 assert [v.nonopt_address for v in infos] == [40, 42, 44, 46, 48, 50] + assert all(v.array_dimensions == ((1, 3), (10, 2)) for v in infos) def test_bbool_mdim_access_id_aligns_to_byte(self) -> None: # ARRAY[0..2, 0..1] of BOOL stored as BBOOL: each row of 3 bits rounds up to a byte,