From d79154f5dcb4d6980aea49abe115e053ad08232c Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 5 Nov 2026 20:24:01 +0200 Subject: [PATCH 1/4] Make tag encoding and decoding public Moves the codec from snap7.client to snap7.tags as decode_tag and encode_tag. The private names stay as aliases, because released ha-s7 versions import them. --- doc/API/tags.rst | 16 ++++ snap7/async_client.py | 10 +- snap7/client.py | 190 ++------------------------------------ snap7/tags.py | 208 ++++++++++++++++++++++++++++++++++++++++++ tests/test_tags.py | 41 ++++++++- 5 files changed, 276 insertions(+), 189 deletions(-) diff --git a/doc/API/tags.rst b/doc/API/tags.rst index 5f741adb..094d0878 100644 --- a/doc/API/tags.rst +++ b/doc/API/tags.rst @@ -79,6 +79,22 @@ forms like ``M7.1`` or ``IW22`` (dispatched to the nodeS7 parser):: parse_tag("M7.1") # raises: ambiguous under strict=True parse_tag("M7.1", strict=False) # → NodeS7Tag (BOOL) +Encoding and decoding +--------------------- + +:func:`~snap7.tags.decode_tag` and :func:`~snap7.tags.encode_tag` convert +between a tag's raw bytes and a Python value, for applications that batch their +own raw reads and writes instead of using ``read_tag`` and ``write_tag``:: + + from s7.tags import decode_tag, encode_tag, parse_tag + + tag = parse_tag("DB1.DBD0:REAL") + value = decode_tag(tag, client.db_read(tag.db_number, tag.byte_offset, tag.size)) + + buf = bytearray(tag.size) + encode_tag(tag, buf, 21.5) + client.db_write(tag.db_number, tag.byte_offset, buf) + Supported types --------------- diff --git a/snap7/async_client.py b/snap7/async_client.py index 9adbee18..e3ed6097 100644 --- a/snap7/async_client.py +++ b/snap7/async_client.py @@ -21,8 +21,8 @@ from .datatypes import S7DataTypes, S7WordLen from .error import S7Error, S7ConnectionError, S7ProtocolError, S7TimeoutError from .client_base import ClientMixin, _instrumented -from .tags import Tag -from .client import Client, _decode_tag, _encode_tag +from .tags import Tag, decode_tag, encode_tag +from .client import Client from .szl import parse_cp_info_szl, parse_cpu_info_szl, parse_order_code_szl, parse_protection_szl from .client import _parse_force_szl from .rate_limiter import RateLimitAlgorithm, RateLimitBehavior, RequestRateLimiter @@ -794,7 +794,7 @@ async def read_tag(self, tag: "Union[Tag, str]", encoding: str = "latin-1") -> A if resolved.is_symbolic: raise NotImplementedError("Symbolic (LID-based) tag access is not supported by the classic S7 client.") data = await self.read_area(Area(resolved.area), resolved.db_number, resolved.byte_offset, resolved.size) - return _decode_tag(resolved, bytearray(data), encoding=encoding) + return decode_tag(resolved, bytearray(data), encoding=encoding) async def write_tag(self, tag: "Union[Tag, str]", value: Any, encoding: str = "latin-1") -> int: """Write a typed value by :class:`~snap7.tags.Tag` or address string. Returns 0 on success.""" @@ -806,7 +806,7 @@ async def write_tag(self, tag: "Union[Tag, str]", value: Any, encoding: str = "l # Preserve the other bits of the byte current = await self.read_area(Area(resolved.area), resolved.db_number, resolved.byte_offset, 1) buf[0] = current[0] - _encode_tag(resolved, buf, value, encoding=encoding) + encode_tag(resolved, buf, value, encoding=encoding) return await self.write_area(Area(resolved.area), resolved.db_number, resolved.byte_offset, buf) async def read_tags(self, tags: "Sequence[Union[Tag, str]]", encoding: str = "latin-1") -> list[Any]: @@ -814,7 +814,7 @@ async def read_tags(self, tags: "Sequence[Union[Tag, str]]", encoding: str = "la resolved = [Tag.from_string(t) if isinstance(t, str) else t for t in tags] items = [{"area": Area(t.area), "db_number": t.db_number, "start": t.byte_offset, "size": t.size} for t in resolved] _code, data_list = await self.read_multi_vars(items) - return [_decode_tag(t, bytearray(d), encoding=encoding) for t, d in zip(resolved, data_list)] + return [decode_tag(t, bytearray(d), encoding=encoding) for t, d in zip(resolved, data_list)] async def db_get(self, db_number: int, size: int = 0) -> bytearray: """Get entire DB. diff --git a/snap7/client.py b/snap7/client.py index dbfe8472..7fa42635 100644 --- a/snap7/client.py +++ b/snap7/client.py @@ -29,8 +29,7 @@ from .log import PLCLoggerAdapter, OperationLogger from .optimizer import ReadItem, ReadPacket, sort_items, merge_items, packetize, extract_results from .rate_limiter import RateLimitAlgorithm, RateLimitBehavior, RequestRateLimiter -from .tags import Tag, _STRING_RE -from . import util +from .tags import Tag, decode_tag, encode_tag from .szl import parse_cp_info_szl, parse_cpu_info_szl, parse_order_code_szl, parse_protection_szl from .type import ( @@ -55,184 +54,9 @@ logger = logging.getLogger(__name__) - -def _decode_tag(tag: Tag, data: bytearray, encoding: str = "latin-1") -> Any: - """Decode a Tag's raw bytes into a typed Python value.""" - upper = tag.datatype.upper() - # Variable-length string types - match = _STRING_RE.match(upper) - if match: - kind, length = match.group(1), int(match.group(2)) - if kind == "FSTRING": - return util.get_fstring(data, 0, length, encoding=encoding) - if kind == "STRING": - return util.get_string(data, 0, encoding=encoding) - if kind == "WSTRING": - return util.get_wstring(data, 0) - - # Arrays - if tag.count > 1: - per = tag.size // tag.count - return [_decode_scalar(upper, data[i * per : (i + 1) * per], tag.bit) for i in range(tag.count)] - - return _decode_scalar(upper, data, tag.bit) - - -def _decode_scalar(datatype: str, data: bytearray, bit: int) -> Any: - """Decode a single scalar value of the given type.""" - if datatype == "BOOL": - return util.get_bool(data, 0, bit) - if datatype == "BYTE": - return util.get_byte(data, 0) - if datatype == "SINT": - return util.get_sint(data, 0) - if datatype == "USINT": - return util.get_usint(data, 0) - if datatype == "CHAR": - return util.get_char(data, 0) - if datatype == "INT": - return util.get_int(data, 0) - if datatype == "UINT": - return util.get_uint(data, 0) - if datatype == "WORD": - return util.get_word(data, 0) - if datatype == "WCHAR": - return util.get_wchar(data, 0) - if datatype == "DATE": - return util.get_date(data, 0) - if datatype == "DINT": - return util.get_dint(data, 0) - if datatype == "UDINT": - return util.get_udint(data, 0) - if datatype == "DWORD": - return util.get_dword(data, 0) - if datatype == "REAL": - return util.get_real(data, 0) - if datatype == "TIME": - return util.get_time(data, 0) - if datatype == "TOD": - return util.get_tod(data, 0) - if datatype == "LINT": - return util.get_lint(data, 0) - if datatype == "ULINT": - return util.get_ulint(data, 0) - if datatype == "LWORD": - return util.get_lword(data, 0) - if datatype == "LREAL": - return util.get_lreal(data, 0) - if datatype == "LTIME": - return util.get_ltime(data, 0) - if datatype == "LTOD": - return util.get_ltod(data, 0) - if datatype == "LDT": - return util.get_ldt(data, 0) - if datatype == "DT": - return util.get_dt(data, 0) - if datatype == "DTL": - return util.get_dtl(data, 0) - raise ValueError(f"Unsupported tag datatype: {datatype}") - - -def _encode_tag(tag: Tag, buf: bytearray, value: Any, encoding: str = "latin-1") -> None: - """Encode a typed Python value into a Tag's byte buffer.""" - upper = tag.datatype.upper() - match = _STRING_RE.match(upper) - if match: - kind, length = match.group(1), int(match.group(2)) - if kind == "FSTRING": - util.set_fstring(buf, 0, value, length, encoding=encoding) - return - if kind == "STRING": - util.set_string(buf, 0, value, length, encoding=encoding) - return - if kind == "WSTRING": - util.set_wstring(buf, 0, value, length) - return - - if tag.count > 1: - per = tag.size // tag.count - for i, v in enumerate(value): - _encode_scalar(upper, buf, i * per, v, tag.bit) - return - - _encode_scalar(upper, buf, 0, value, tag.bit) - - -def _encode_scalar(datatype: str, buf: bytearray, offset: int, value: Any, bit: int) -> None: - """Encode a single scalar value at the given offset.""" - if datatype == "BOOL": - util.set_bool(buf, offset, bit, value) - return - if datatype in ("BYTE", "USINT"): - util.set_byte(buf, offset, value) if datatype == "BYTE" else util.set_usint(buf, offset, value) - return - if datatype == "SINT": - util.set_sint(buf, offset, value) - return - if datatype == "CHAR": - util.set_char(buf, offset, value) - return - if datatype == "INT": - util.set_int(buf, offset, value) - return - if datatype == "UINT": - util.set_uint(buf, offset, value) - return - if datatype == "WORD": - util.set_word(buf, offset, value) - return - if datatype == "WCHAR": - util.set_wchar(buf, offset, value) - return - if datatype == "DATE": - util.set_date(buf, offset, value) - return - if datatype == "DINT": - util.set_dint(buf, offset, value) - return - if datatype == "UDINT": - util.set_udint(buf, offset, value) - return - if datatype == "DWORD": - util.set_dword(buf, offset, value) - return - if datatype == "REAL": - util.set_real(buf, offset, value) - return - if datatype == "TIME": - util.set_time(buf, offset, value) - return - if datatype == "TOD": - util.set_tod(buf, offset, value) - return - if datatype == "LINT": - util.set_lint(buf, offset, value) - return - if datatype == "ULINT": - util.set_ulint(buf, offset, value) - return - if datatype == "LWORD": - util.set_lword(buf, offset, value) - return - if datatype == "LREAL": - util.set_lreal(buf, offset, value) - return - if datatype == "LTIME": - util.set_ltime(buf, offset, value) - return - if datatype == "LTOD": - util.set_ltod(buf, offset, value) - return - if datatype == "LDT": - util.set_ldt(buf, offset, value) - return - if datatype == "DT": - util.set_dt(buf, offset, value) - return - if datatype == "DTL": - util.set_dtl(buf, offset, value) - return - raise ValueError(f"Unsupported tag datatype: {datatype}") +# Old private names; ha-s7 releases up to 0.1.4 import these from here. +_decode_tag = decode_tag +_encode_tag = encode_tag def _parse_force_szl(raw: bytes) -> list[ForceEntry]: @@ -864,7 +688,7 @@ def read_tag(self, tag: "Union[Tag, str]", encoding: str = "latin-1") -> Any: if resolved.is_symbolic: raise NotImplementedError("Symbolic (LID-based) tag access is not supported by the classic S7 client.") data = self.read_area(Area(resolved.area), resolved.db_number, resolved.byte_offset, resolved.size) - return _decode_tag(resolved, bytearray(data), encoding=encoding) + return decode_tag(resolved, bytearray(data), encoding=encoding) def write_tag(self, tag: "Union[Tag, str]", value: Any, encoding: str = "latin-1") -> int: """Write a typed value by :class:`~snap7.tags.Tag` or address string. @@ -886,7 +710,7 @@ def write_tag(self, tag: "Union[Tag, str]", value: Any, encoding: str = "latin-1 if resolved.datatype.upper() == "BOOL": current = self.read_area(Area(resolved.area), resolved.db_number, resolved.byte_offset, 1) buf[0] = current[0] - _encode_tag(resolved, buf, value, encoding=encoding) + encode_tag(resolved, buf, value, encoding=encoding) return self.write_area(Area(resolved.area), resolved.db_number, resolved.byte_offset, buf) def read_tags(self, tags: "Sequence[Union[Tag, str]]", encoding: str = "latin-1") -> list[Any]: @@ -905,7 +729,7 @@ def read_tags(self, tags: "Sequence[Union[Tag, str]]", encoding: str = "latin-1" resolved = [Tag.from_string(t) if isinstance(t, str) else t for t in tags] items = [{"area": Area(t.area), "db_number": t.db_number, "start": t.byte_offset, "size": t.size} for t in resolved] _code, data_list = self.read_multi_vars(items) - return [_decode_tag(t, bytearray(d), encoding=encoding) for t, d in zip(resolved, data_list)] + return [decode_tag(t, bytearray(d), encoding=encoding) for t, d in zip(resolved, data_list)] def db_read(self, db_number: int, start: int, size: int) -> bytearray: """ diff --git a/snap7/tags.py b/snap7/tags.py index ae851a08..79da0fab 100644 --- a/snap7/tags.py +++ b/snap7/tags.py @@ -47,6 +47,7 @@ from pathlib import Path from typing import Any, Union +from . import util from .type import Area # Type name → byte size for fixed-size types @@ -465,6 +466,213 @@ def parse_tag(address: str, *, strict: bool = True, name: str = "") -> Tag: return NodeS7Tag.parse(s, name) +# --------------------------------------------------------------------------- +# Raw bytes <-> Python values +# --------------------------------------------------------------------------- + + +def decode_tag(tag: Tag, data: bytearray, encoding: str = "latin-1") -> Any: + """Decode a Tag's raw bytes into a typed Python value. + + Args: + tag: The tag describing the value. + data: The tag's raw bytes, starting at offset 0, as read from the PLC. + encoding: Character encoding for ``STRING`` and ``FSTRING``. + + Returns: + The decoded value, or a list of values when ``tag.count`` is above 1. + + Raises: + ValueError: If the tag's type is unknown or has no decoder. + """ + upper = tag.datatype.upper() + # Variable-length string types + match = _STRING_RE.match(upper) + if match: + kind, length = match.group(1), int(match.group(2)) + if kind == "FSTRING": + return util.get_fstring(data, 0, length, encoding=encoding) + if kind == "STRING": + return util.get_string(data, 0, encoding=encoding) + if kind == "WSTRING": + return util.get_wstring(data, 0) + + # Arrays + if tag.count > 1: + per = tag.size // tag.count + return [_decode_scalar(upper, data[i * per : (i + 1) * per], tag.bit) for i in range(tag.count)] + + return _decode_scalar(upper, data, tag.bit) + + +def _decode_scalar(datatype: str, data: bytearray, bit: int) -> Any: + """Decode a single scalar value of the given type.""" + if datatype == "BOOL": + return util.get_bool(data, 0, bit) + if datatype == "BYTE": + return util.get_byte(data, 0) + if datatype == "SINT": + return util.get_sint(data, 0) + if datatype == "USINT": + return util.get_usint(data, 0) + if datatype == "CHAR": + return util.get_char(data, 0) + if datatype == "INT": + return util.get_int(data, 0) + if datatype == "UINT": + return util.get_uint(data, 0) + if datatype == "WORD": + return util.get_word(data, 0) + if datatype == "WCHAR": + return util.get_wchar(data, 0) + if datatype == "DATE": + return util.get_date(data, 0) + if datatype == "DINT": + return util.get_dint(data, 0) + if datatype == "UDINT": + return util.get_udint(data, 0) + if datatype == "DWORD": + return util.get_dword(data, 0) + if datatype == "REAL": + return util.get_real(data, 0) + if datatype == "TIME": + return util.get_time(data, 0) + if datatype == "TOD": + return util.get_tod(data, 0) + if datatype == "LINT": + return util.get_lint(data, 0) + if datatype == "ULINT": + return util.get_ulint(data, 0) + if datatype == "LWORD": + return util.get_lword(data, 0) + if datatype == "LREAL": + return util.get_lreal(data, 0) + if datatype == "LTIME": + return util.get_ltime(data, 0) + if datatype == "LTOD": + return util.get_ltod(data, 0) + if datatype == "LDT": + return util.get_ldt(data, 0) + if datatype == "DT": + return util.get_dt(data, 0) + if datatype == "DTL": + return util.get_dtl(data, 0) + raise ValueError(f"Unsupported tag datatype: {datatype}") + + +def encode_tag(tag: Tag, buf: bytearray, value: Any, encoding: str = "latin-1") -> None: + """Encode a typed Python value into a Tag's byte buffer. + + Args: + tag: The tag describing the value. + buf: Buffer of at least ``tag.size`` bytes. For ``BOOL`` it must hold the + byte's current value, since the other bits are kept. + value: The value, or a sequence of values when ``tag.count`` is above 1. + encoding: Character encoding for ``STRING`` and ``FSTRING``. + + Raises: + ValueError: If the tag's type is unknown or has no encoder. + """ + upper = tag.datatype.upper() + match = _STRING_RE.match(upper) + if match: + kind, length = match.group(1), int(match.group(2)) + if kind == "FSTRING": + util.set_fstring(buf, 0, value, length, encoding=encoding) + return + if kind == "STRING": + util.set_string(buf, 0, value, length, encoding=encoding) + return + if kind == "WSTRING": + util.set_wstring(buf, 0, value, length) + return + + if tag.count > 1: + per = tag.size // tag.count + for i, v in enumerate(value): + _encode_scalar(upper, buf, i * per, v, tag.bit) + return + + _encode_scalar(upper, buf, 0, value, tag.bit) + + +def _encode_scalar(datatype: str, buf: bytearray, offset: int, value: Any, bit: int) -> None: + """Encode a single scalar value at the given offset.""" + if datatype == "BOOL": + util.set_bool(buf, offset, bit, value) + return + if datatype in ("BYTE", "USINT"): + util.set_byte(buf, offset, value) if datatype == "BYTE" else util.set_usint(buf, offset, value) + return + if datatype == "SINT": + util.set_sint(buf, offset, value) + return + if datatype == "CHAR": + util.set_char(buf, offset, value) + return + if datatype == "INT": + util.set_int(buf, offset, value) + return + if datatype == "UINT": + util.set_uint(buf, offset, value) + return + if datatype == "WORD": + util.set_word(buf, offset, value) + return + if datatype == "WCHAR": + util.set_wchar(buf, offset, value) + return + if datatype == "DATE": + util.set_date(buf, offset, value) + return + if datatype == "DINT": + util.set_dint(buf, offset, value) + return + if datatype == "UDINT": + util.set_udint(buf, offset, value) + return + if datatype == "DWORD": + util.set_dword(buf, offset, value) + return + if datatype == "REAL": + util.set_real(buf, offset, value) + return + if datatype == "TIME": + util.set_time(buf, offset, value) + return + if datatype == "TOD": + util.set_tod(buf, offset, value) + return + if datatype == "LINT": + util.set_lint(buf, offset, value) + return + if datatype == "ULINT": + util.set_ulint(buf, offset, value) + return + if datatype == "LWORD": + util.set_lword(buf, offset, value) + return + if datatype == "LREAL": + util.set_lreal(buf, offset, value) + return + if datatype == "LTIME": + util.set_ltime(buf, offset, value) + return + if datatype == "LTOD": + util.set_ltod(buf, offset, value) + return + if datatype == "LDT": + util.set_ldt(buf, offset, value) + return + if datatype == "DT": + util.set_dt(buf, offset, value) + return + if datatype == "DTL": + util.set_dtl(buf, offset, value) + return + raise ValueError(f"Unsupported tag datatype: {datatype}") + + # --------------------------------------------------------------------------- # nodeS7 syntax tables and helpers # --------------------------------------------------------------------------- diff --git a/tests/test_tags.py b/tests/test_tags.py index b76706d8..868103e1 100644 --- a/tests/test_tags.py +++ b/tests/test_tags.py @@ -5,7 +5,7 @@ import pytest -from snap7.tags import NodeS7Tag, PLC4XTag, Tag, from_browse, load_csv, load_json, load_tia_xml, parse_tag +from snap7.tags import NodeS7Tag, PLC4XTag, Tag, decode_tag, encode_tag, from_browse, load_csv, load_json, load_tia_xml, parse_tag from snap7.type import Area @@ -147,6 +147,45 @@ def test_unknown_type_raises(self) -> None: Tag(Area.DB, 1, 0, "MYSTERY").size # noqa: B018 +class TestTagCodec: + @pytest.mark.parametrize( + "address, value", + [ + ("DB1.DBX0.3:BOOL", True), + ("DB1:0:INT", -2), + ("DB1:0:REAL", 1.5), + ("DB1:0:INT[3]", [1, -2, 3]), + ("DB1:0:STRING[8]", "abc"), + ("DB1:0:WSTRING[8]", "abc"), + ], + ) + def test_round_trip(self, address: str, value: object) -> None: + tag = parse_tag(address) + buf = bytearray(tag.size) + encode_tag(tag, buf, value) + assert decode_tag(tag, buf) == value + + def test_known_bytes(self) -> None: + tag = parse_tag("DB1:0:INT") + assert decode_tag(tag, bytearray(bytes.fromhex("fffe"))) == -2 + buf = bytearray(2) + encode_tag(tag, buf, -2) + assert buf.hex() == "fffe" + + def test_unknown_type_raises(self) -> None: + tag = Tag(area=Area.DB, db_number=1, byte_offset=0, datatype="BOGUS") + with pytest.raises(ValueError): + decode_tag(tag, bytearray(2)) + with pytest.raises(ValueError): + encode_tag(tag, bytearray(2), 1) + + def test_private_names_still_work(self) -> None: + from snap7.client import _decode_tag, _encode_tag + + assert _decode_tag is decode_tag + assert _encode_tag is encode_tag + + class TestLoadCsv: """Load tags from CSV files and strings.""" From b0250cbd7470e1b02a252ae8181366f6a48f5414 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 5 Nov 2026 20:24:26 +0200 Subject: [PATCH 2/4] Add the changelog entry --- CHANGES.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/CHANGES.md b/CHANGES.md index d5223c66..6f0c382d 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,6 +1,15 @@ CHANGES ======== +Unreleased +---------- + +### New features + +* Add `snap7.tags.decode_tag` and `snap7.tags.encode_tag`, which convert between a tag's raw bytes and a Python + value, for applications that batch their own raw reads and writes. The private names in `snap7.client` still + work (#973). + 3.3.1 ----- From 00453ed6508778bcd97eb25147d8f0779a1eec12 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 5 Nov 2026 20:26:06 +0200 Subject: [PATCH 3/4] Refuse tags with no known size in strict parse_tag Also gives S5TIME a size and a decoder. Writing it stays unsupported, since there is no set_s5time. --- doc/API/tags.rst | 2 +- snap7/tags.py | 31 ++++++++++++++++++++++--------- tests/test_tags.py | 26 ++++++++++++++++++++++++++ 3 files changed, 49 insertions(+), 10 deletions(-) diff --git a/doc/API/tags.rst b/doc/API/tags.rst index 094d0878..ce44429d 100644 --- a/doc/API/tags.rst +++ b/doc/API/tags.rst @@ -102,7 +102,7 @@ Supported types ``UINT``, ``WORD``, ``DINT``, ``UDINT``, ``DWORD``, ``LINT``, ``ULINT``, ``LWORD``, ``REAL``, ``LREAL``, ``TIME``, ``LTIME``, ``TOD``, ``LTOD``, ``DATE``, ``DT``, ``LDT``, ``DTL``, ``STRING[n]``, ``WSTRING[n]``, -``FSTRING[n]``. +``FSTRING[n]``. ``S5TIME`` can be read but not yet written. Arrays are supported for any fixed-size type via ``[count]`` suffix. diff --git a/snap7/tags.py b/snap7/tags.py index 79da0fab..a45986d6 100644 --- a/snap7/tags.py +++ b/snap7/tags.py @@ -62,6 +62,7 @@ "WORD": 2, "WCHAR": 2, "DATE": 2, + "S5TIME": 2, "DINT": 4, "UDINT": 4, "DWORD": 4, @@ -440,9 +441,11 @@ def parse_tag(address: str, *, strict: bool = True, name: str = "") -> Tag: Args: address: Tag address string. strict: When ``True`` (default), require one of the dialect markers - above. Bare short forms like ``M7.1`` or ``IW22`` raise + above and a type with a known size. Bare short forms like ``M7.1`` + or ``IW22``, unknown types and ``STRING`` without a length raise :class:`ValueError`. When ``False``, bare forms are dispatched - to the nodeS7 parser (which accepts them). + to the nodeS7 parser (which accepts them), and the type is not + checked until :attr:`Tag.size` is read. name: Optional tag name to store on the resulting Tag. Returns: @@ -450,20 +453,28 @@ def parse_tag(address: str, *, strict: bool = True, name: str = "") -> Tag: detected. Raises: - ValueError: If the input is ambiguous under strict mode, or if - the selected parser fails to parse. + ValueError: If the input is ambiguous or its type has no known size + under strict mode, or if the selected parser fails to parse. """ s = address.strip() + tag: Tag if "," in s: - return NodeS7Tag.parse(s, name) - if ":" in s: - return PLC4XTag.parse(s, name) - if strict: + tag = NodeS7Tag.parse(s, name) + elif ":" in s: + tag = PLC4XTag.parse(s, name) + elif strict: raise ValueError( f"Ambiguous tag syntax {address!r}: must contain ',' (nodeS7) " f"or ':TYPE' (PLC4X). Pass strict=False to accept bare short forms." ) - return NodeS7Tag.parse(s, name) + else: + tag = NodeS7Tag.parse(s, name) + if strict: + try: + tag.size + except ValueError as e: + raise ValueError(f"Tag address {address!r} has a type with no known size: {tag.datatype}") from e + return tag # --------------------------------------------------------------------------- @@ -527,6 +538,8 @@ def _decode_scalar(datatype: str, data: bytearray, bit: int) -> Any: return util.get_wchar(data, 0) if datatype == "DATE": return util.get_date(data, 0) + if datatype == "S5TIME": + return util.get_s5time(data, 0) if datatype == "DINT": return util.get_dint(data, 0) if datatype == "UDINT": diff --git a/tests/test_tags.py b/tests/test_tags.py index 868103e1..0655b232 100644 --- a/tests/test_tags.py +++ b/tests/test_tags.py @@ -6,6 +6,7 @@ import pytest from snap7.tags import NodeS7Tag, PLC4XTag, Tag, decode_tag, encode_tag, from_browse, load_csv, load_json, load_tia_xml, parse_tag +from snap7 import util from snap7.type import Area @@ -186,6 +187,31 @@ def test_private_names_still_work(self) -> None: assert _encode_tag is encode_tag +class TestParseTagKnownSize: + @pytest.mark.parametrize("address", ["DB1.DBW0:BOGUS", "DB1.DBB0:STRING", "DB1:0:WSTRING", "MW0:BOGUS"]) + def test_strict_refuses_a_type_without_a_size(self, address: str) -> None: + with pytest.raises(ValueError, match="no known size"): + parse_tag(address) + + def test_not_strict_leaves_the_type_unchecked(self) -> None: + tag = parse_tag("DB1.DBW0:BOGUS", strict=False) + assert tag.datatype == "BOGUS" + with pytest.raises(ValueError, match="Unknown S7 type"): + tag.size + + def test_explicit_parsers_are_unchanged(self) -> None: + assert PLC4XTag.parse("DB1.DBW0:BOGUS").datatype == "BOGUS" + + def test_s5time_has_a_size_and_decodes(self) -> None: + tag = parse_tag("DB1.DBW0:S5TIME") + assert tag.size == 2 + assert decode_tag(tag, bytearray(bytes.fromhex("0123"))) == util.get_s5time(bytearray(bytes.fromhex("0123")), 0) + + def test_s5time_write_is_refused(self) -> None: + with pytest.raises(ValueError, match="S5TIME"): + encode_tag(parse_tag("DB1.DBW0:S5TIME"), bytearray(2), "10ms") + + class TestLoadCsv: """Load tags from CSV files and strings.""" From 24bf123033280bffff3a11a14ce710a6b8204474 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 5 Nov 2026 20:26:43 +0200 Subject: [PATCH 4/4] Add the changelog entry --- CHANGES.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGES.md b/CHANGES.md index 6f0c382d..c1d79024 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -10,6 +10,12 @@ Unreleased value, for applications that batch their own raw reads and writes. The private names in `snap7.client` still work (#973). +### Bug fixes + +* `parse_tag` now raises `ValueError` under `strict=True`, the default, for a type with no known size, such as + `DB1.DBB0:STRING` without a length. Previously this failed only when `.size` was read. `S5TIME` now has a size + and can be read, but not yet written (#974). + 3.3.1 -----