diff --git a/s7commplus/tag_browser.py b/s7commplus/tag_browser.py new file mode 100644 index 00000000..3f32f6f3 --- /dev/null +++ b/s7commplus/tag_browser.py @@ -0,0 +1,358 @@ +"""Parse symbolic tags and block interfaces from S7CommPlus EXPLORE responses. + +S7-1200/1500 PLCs answer an EXPLORE request with a zlib blob compressed +against a Siemens preset dictionary. Once decompressed with +:func:`s7commplus.blob_decompressor.decompress_blob`, the payload is clean +XML. Two shapes are handled here: + +* **I/Q/M areas** -> ```` with ```` entries that carry + a direct ```` and ```` address (dict + ``IntfDescTag`` / Adler-32 ``0xce9b821b``). Parsed into :class:`Tag`. + +* **Data blocks / FBs** -> ```` with ```` entries whose + data type is a *reference* (``RID="0x0200_00XX"``) into the Siemens + SoftDataType table, not an inline string (dict ``DebugInfo_IntfDesc`` / + Adler-32 ``0x66052b13``). Parsed into :class:`Member`. + +Validated end-to-end against a live S7-1200 G2 (FW V4.1): I/Q/M tags and an +optimized DB (``Data_block_1``, 11 members, names + Bool/Int/Real types). +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +import zlib +from dataclasses import dataclass, field + +from .blob_decompressor import decompress_blob + +# --------------------------------------------------------------------------- +# I/Q/M symbolic tags () +# --------------------------------------------------------------------------- + +# EXPLORE relation IDs per area (S7-1200 V3 / FW V4.x). +AREA_RID = {"I": 80, "Q": 81, "M": 82} + +# -> TIA area letter. +_RANGE_LETTER = {"Input": "I", "Output": "Q", "Memory": "M"} +# -> TIA address suffix for non-bit widths. +_WIDTH_SUFFIX = {"Byte": "B", "Word": "W", "DWord": "D", "LWord": "L"} + + +@dataclass +class Tag: + """A symbolic I/Q/M tag resolved from an EXPLORE response.""" + + name: str + data_type: str # , e.g. Bool / Byte / Word / DWord + address: str # TIA syntax, e.g. %I0.0 / %MB100 / %MW102 + lid: int + byte_offset: int + bit_offset: int | None + + +def _address(rng: str, width: str, byte_no: int, bit_no: int | None) -> str: + letter = _RANGE_LETTER.get(rng, "?") + if width == "Bit": + return f"%{letter}{byte_no}.{bit_no or 0}" + return f"%{letter}{_WIDTH_SUFFIX.get(width, '')}{byte_no}" + + +def parse_ident_container(xml_text: str) -> list[Tag]: + """Parse a decompressed ```` document into :class:`Tag`s.""" + tags: list[Tag] = [] + for ident in ET.fromstring(xml_text).findall("Ident"): + acc = ident.find("./Access/SimpleAccess") + if acc is None: + continue # only SimpleAccess idents carry a direct I/Q/M address + byte_no = int(acc.get("ByteNumber", "0")) + bit_raw = acc.get("BitNumber") + bit_no = int(bit_raw) if bit_raw is not None else None + tags.append( + Tag( + name=ident.get("Name", ""), + data_type=ident.findtext("SimpleType", default=""), + address=_address(acc.get("Range", ""), acc.get("Width", ""), byte_no, bit_no), + lid=int(ident.get("LID", "0")), + byte_offset=byte_no, + bit_offset=bit_no, + ) + ) + return tags + + +def tags_from_explore(explore_payload: bytes) -> list[Tag]: + """Decompress a raw I/Q/M EXPLORE payload and parse its symbolic tags. + + Returns an empty list if the payload contains no preset-dict zlib stream. + """ + pos = _find_preset_stream(explore_payload, _IDENT_ADLER) + if pos is None: + return [] + return parse_ident_container(decompress_blob(explore_payload, offset=pos)) + + +# --------------------------------------------------------------------------- +# Data-block / FB interfaces () +# --------------------------------------------------------------------------- + +# Adler-32 of the ``IntfDescTag`` preset dict (I/Q/M symbols). +_IDENT_ADLER = 0xCE9B821B +# Adler-32 of interface-description preset dicts. ``DebugInfo_IntfDesc`` carries +# the optimized (symbolic-access) block interface; ``IntfDesc`` the standard one. +_INTFDESC_ADLERS = (0x66052B13, 0x4B8416F0) + +# Siemens SoftDataType ids, from thomas-v2/S7CommPlusDriver Core/Softdatatype.cs +# (LGPL-3.0). A type is a RID of the form 0x0200_00XX where XX is the +# SoftDataType id below. +SOFTDATATYPE = { + 0: "Void", + 1: "Bool", + 2: "Byte", + 3: "Char", + 4: "Word", + 5: "Int", + 6: "DWord", + 7: "DInt", + 8: "Real", + 9: "Date", + 10: "TimeOfDay", + 11: "Time", + 12: "S5Time", + 13: "S5Count", + 14: "DateAndTime", + 15: "InternetTime", + 16: "Array", + 17: "Struct", + 18: "EndStruct", + 19: "String", + 20: "Pointer", + 21: "MultiFB", + 22: "Any", + 23: "BlockFB", + 24: "BlockFC", + 25: "BlockDB", + 26: "BlockSDB", + 28: "Counter", + 29: "Timer", + 30: "IEC_Counter", + 31: "IEC_Timer", + 37: "BlockUDT", + 40: "BBool", + 48: "LReal", + 49: "ULInt", + 50: "LInt", + 51: "LWord", + 52: "USInt", + 53: "UInt", + 54: "UDInt", + 55: "SInt", + 61: "WChar", + 62: "WString", + 63: "Variant", + 64: "LTime", + 65: "LTimeOfDay", + 66: "LDT", + 67: "DTL", +} + + +@dataclass +class Member: + """A single variable inside a data block or FB/OB interface.""" + + name: str + data_type: str # readable type: SoftDataType name or the inline complex type + lid: int + offset: int # StdO: standard (non-optimized) byte offset + rid: int # raw type RID + section: str = "" # Input/Output/InOut/Static/Temp/Constant (FB/OB); "" for a DB + fields: list[Member] = field(default_factory=list) # sub-members for a UDT/Struct type + + +def _member_type(member: ET.Element) -> str: + """Resolve a ````'s data type to a readable name. + + Complex members (arrays, UDTs, strings, structs) already carry a ready-made + ``Type`` attribute (e.g. ``Array[0..5] of IEC_TIMER``). Scalar elementary + members carry only a RID of the form ``0x0200_00XX`` whose low byte is a + Siemens SoftDataType id. (Arrays use ``0x0201_00XX``, hence the explicit + ``Type``; the raw RID stays available on :attr:`Member.rid`.) + """ + explicit = member.get("Type") + if explicit: + return explicit + try: + rid = int(member.get("RID", "0") or "0", 16) + except ValueError: + return "" + if (rid & 0xFFFF0000) == 0x02000000: + return SOFTDATATYPE.get(rid & 0xFFFF, f"SoftType#{rid & 0xFFFF}") + return f"0x{rid:08X}" + + +def parse_block_interface(xml_text: str) -> list[Member]: + """Parse a decompressed ```` document into :class:`Member`s. + + Works for DB, FB, OB and UDT interfaces. The block's own variables come from + its primary ```` — directly under ```` for a flat + DB/UDT, or from the ```` entries for an FB/OB. Members + are returned in declaration order, each tagged with its section + (Input/Output/... for FB/OB; empty for DB/UDT). + + When a member is of a UDT (or other structured type), the interface embeds + that type's definition under the primary part's ````; the member's + ``SubPartIndex`` selects it. Such members are expanded recursively into + :attr:`Member.fields`. Library types (e.g. ``IEC_TIMER``) are referenced but + not inlined, so they expand to no fields. + """ + root = ET.fromstring(xml_text) + primary = root.find("Part") # first top-level Part = the block's own source + if primary is None: + return [] + return _parse_source_part(primary, set()) + + +def _parse_source_part(part: ET.Element, seen: set[int]) -> list[Member]: + """Parse one source ```` (DBSource/BlockSource/DataTypeSource). + + ``seen`` carries the identities of the ```` elements already being + expanded on the current path, so a malformed ``SubPartIndex`` cycle stops + instead of recursing forever. + """ + sub_el = part.find("SubParts") + subparts = list(sub_el) if sub_el is not None else [] # index-addressable + + members: list[Member] = [] + # Flat DB/UDT: variables are direct children of . + root_el = part.find("./Payload/Root") + if root_el is not None: + for m in root_el.findall("Member"): + if m.get("RID") is not None: + members.append(_build_member(m, "", subparts, seen)) + # FB/OB: variables live in the entries of . + for sp in subparts: + kind = sp.get("Kind", "") + if not kind.endswith("Section"): + continue + section = kind[: -len("Section")] + for m in sp.findall("./Payload/Member/Member"): + if m.get("RID") is not None: + members.append(_build_member(m, section, subparts, seen)) + return members + + +def _build_member(m: ET.Element, section: str, subparts: list, seen: set[int]) -> Member: + """Build a :class:`Member`, expanding a structured type via ``SubPartIndex``.""" + member = Member( + name=m.get("Name", ""), + data_type=_member_type(m), + lid=int(m.get("LID", "0")), + offset=int(m.get("StdO", "0")), + rid=int(m.get("RID", "0") or "0", 16), + section=section, + ) + spi = m.get("SubPartIndex") + if spi is not None: + idx = int(spi) + # Only an inlined type definition (a *Source part) carries sub-members; + # a *Section index or a library reference expands to nothing. + if 0 <= idx < len(subparts) and subparts[idx].get("Kind", "").endswith("Source"): + target = subparts[idx] + if id(target) not in seen: + member.fields = _parse_source_part(target, seen | {id(target)}) + return member + + +def block_interface_from_explore(explore_payload: bytes) -> list[Member]: + """Decompress a block EXPLORE payload and parse its interface members. + + The EXPLORE response for a block RID bundles several preset-dict streams + (IdentES, interface, comments, ...). This targets the interface dictionary + specifically rather than taking the first stream. Returns an empty list if + no interface stream is present. + """ + for adler in _INTFDESC_ADLERS: + pos = _find_preset_stream(explore_payload, adler) + if pos is not None: + return parse_block_interface(decompress_blob(explore_payload, offset=pos)) + return [] + + +# --------------------------------------------------------------------------- +# shared helpers +# --------------------------------------------------------------------------- + + +def _find_preset_stream(data: bytes, adler: int) -> int | None: + """Return the offset of the first preset-dict zlib stream matching ``adler``. + + A block EXPLORE response contains multiple ``78 xx`` FDICT streams; callers + need the one for a specific dictionary, not merely the first. + """ + i = 0 + # A header needs 6 bytes (CMF + FLG + 4-byte DICTID), so the last valid + # start position is len(data) - 6 inclusive. + end = len(data) - 5 + while i < end: + if data[i] == 0x78 and (data[i + 1] & 0x20) and int.from_bytes(data[i + 2 : i + 6], "big") == adler: + return i + i += 1 + return None + + +def _find_zlib_stream(data: bytes) -> int | None: + """Return the offset of the first standard (non-preset) zlib stream. + + Standard-compression ``PlcContentInfo`` uses CMF ``0x78`` but its FLG byte + varies with the compression level (``78 01`` / ``78 5e`` / ``78 9c`` / + ``78 da``). Match any ``0x78`` whose 2-byte header checksums correctly + (``CMF*256 + FLG`` divisible by 31) and has no preset-dict flag set, rather + than hardcoding a single FLG value. + """ + for i in range(len(data) - 1): + if data[i] == 0x78 and not (data[i + 1] & 0x20) and ((data[i] << 8) | data[i + 1]) % 31 == 0: + return i + return None + + +@dataclass +class DataBlock: + """A data block advertised in the PlcContentInfo listing.""" + + name: str + number: int + rid: int + + +def datablocks_from_explore(wildcard_payload: bytes) -> list[DataBlock]: + """Parse the DB list from a ``0x8A11FFFF`` wildcard EXPLORE response. + + PlcContentInfo is a *standard* zlib stream embedded in a larger BLOB, so it + is raw-inflated (the Adler-32 trailer would fail on the trailing bytes). The + stream header is located tolerantly (any compression level), not assumed to + be ``78 da``. Returns only ``Type="DB"`` blocks. + """ + pos = _find_zlib_stream(wildcard_payload) + if pos is None: + return [] + try: + xml_bytes = zlib.decompressobj(wbits=-15).decompress(wildcard_payload[pos + 2 :]) + except zlib.error: + return [] + dbs: list[DataBlock] = [] + for entity in ET.fromstring(xml_bytes.decode("utf-8")).findall('.//Entity[@Id="Block"]'): + header = entity.find("Header") + if header is None or header.get("Type") != "DB": + continue + try: + dbs.append( + DataBlock( + name=header.get("Name", ""), + number=int(header.get("Number", "0")), + rid=int(entity.get("Rid", "0")), + ) + ) + except ValueError: + continue + return dbs diff --git a/tests/test_tag_browser.py b/tests/test_tag_browser.py new file mode 100644 index 00000000..d78764fc --- /dev/null +++ b/tests/test_tag_browser.py @@ -0,0 +1,273 @@ +"""Tests for the symbolic tag / block-interface browser. + +The I/Q/M fixture is a real preset-dictionary EXPLORE blob captured from a live +S7-1200 G2 (FW V4.1) M area (``IntfDescTag`` dict, Adler-32 ``0xce9b821b``). +The block-interface fixture is a real EXPLORE blob for an optimized DB +(``Data_block_1``) from the same PLC, exercising the ``DebugInfo_IntfDesc`` +dict (Adler-32 ``0x66052b13``) and the RID -> SoftDataType mapping. +""" + +import base64 +import zlib + +from s7commplus.tag_browser import ( + DataBlock, + Member, + Tag, + block_interface_from_explore, + datablocks_from_explore, + parse_block_interface, + parse_ident_container, + tags_from_explore, +) + +# Live S7-1200 G2 (FW V4.1) EXPLORE response for the M area, from the zlib +# header onward: b"\x78\x7d" + dict-adler(0xce9b821b) + raw deflate. +_M_AREA_BLOB = base64.b64decode( + "eH3Om4Ibs7GvyM1RKEstKs7MzwNmJj1gMZCal5yfkpmXbqtUWpKma6Fkb2eDrgs5AHJTi7JTi2hVgOG" + "MSCNiii8sgQEPChQ/lABjMQloHHZPGBqi+QKokhJfIBdfBkguBzuAHKeXg1MAVqejlWCgtEI1pxthL3t" + "JcnoKHrejFWEuVHW8CfEZCE+aAaZnHI43o1XKN1BC84kFyWkfvRgAACgQWbajk3JAFAIAOJgAAAJ4fe" + "JynqGzsa/IzVEoSy0qBgoAvatnoKSQmpecn5KZl26rVFqSpmuhZI/NBiLNBwAy/SHEAKKiogAAAAA=" +) + +# Live S7-1200 G2 EXPLORE response for optimized DB "Data_block_1" (DB25), from +# the IdentES stream onward. Contains several preset-dict streams; the parser +# must skip IdentES (0x5814b03b) and target the interface dict (0x66052b13). +_DB_INTERFACE_BLOB = base64.b64decode( + "eH1YFLA7rZXBToQwEIZfxezdUCosECtmdRc9oAfhBRqZYJPSGrZoeHtni2ukS0+7lx6+djqdmX+m" + "p4mgNhGbwehXK5Gj73/ECSejJKTrNCZhlJCUuP18WCvDu0/7xprcrY4mYZTF6xRtktVVHXo49fAb" + "D488PPbwxMPTZR5cfkCdqfLCVfkbSOB7eNSDMnlMWDADbKu/ldS8eRZ7o/sRR9T9NbllgcuZ1XPJ" + "VTvwFkr4ApnjP7VAWc37FkzBOyHHvMLpRMgTZcEMs11VDOrdYBBcCjNOphRvXNzA43Mn00nH8aZp" + "xGRY4suxKvAgFO9Hm7IDeoEOYylBteYDmzz7LeGf/n8AgprvX6OhQEAVAKOTcAAUAIZ3mAAAAnh9" + "ZgUrE42Y224bOQyGX8XwPWGROi+aAE1TYHvVxWLRW0NHINhsncZOEezTl5qjPVM3vUlGE/ET+YtD" + "Snm7Mrp1YbzckNuLDrTZ9T1o2Lv7u77icbL3O4TISRhez/PD9K+m75kUz/77A881pEhrb9WUA0Mz" + "udneffry8Y8W/T423/a4Szln8p5AIQpQpAyEEA1QwWKKTsmjHepPOyuV1xPdbHtNWWbVTj3cNPpX" + "TjT5h9ldrohX320I+q4h9zoiEk/idlUeb7ZfiGu66NNp+MoGuvK20QfxSYmpnbUoedLdh/v3+n2T" + "8p9nrl2s/Sa98GchijEhGw8YkwelLYE30oJ1yQWSJsla2ejPh2NLhg3DVZZaqhK5oVkHqlQCV50C" + "4ZRxwjprq9m2PTqzyRyFpGLA6aJBZV/ABRVA+sCvtYjV0tLG5GBTVR4cEq/jLPKTRJDCCMxBoo+4" + "tJEuVO+KBJfQgLIhQ6iSQyJHhLy2dnZpQ6FaVF6ArDlyPK6AjzlweNkEgw6VXsXD0aAK0UJIxPFE" + "ChxPSWC1z84a61Gs1qmOs0xKDcJGVhprBC+UZc29taiVMxRXGkShCvISxiPbRG1Za68guFyCSCkI" + "HZY2WlcKaAyw4xyP4um+mgop8/uSpE5FLm1SirypMUCKVYIKQYEzPPSBXIq2RGlrZ7Mbs+d2rtzt" + "KLFswknYLJzP4J2roGQUEKRLQNJirlkgJ9f2rLe3xByqbfsUNI7V5lgey4ndLHtuHPvwcjoMlXUu" + "X13if26Mvtr6ztNzGo20p5fH4/7YTpn88/C0Tw/p8ToQByCKFVHO/p2e2tFlX5/Lt/1/h+bqEqgn" + "oBmJ7NF3TqMVV43cf5/2f326X5LcSJI0ktQVkp5ID78kGTWS9IphJkb+JcNPcZkVw46MdPh6vCLQ" + "rDi5keSuxOVG3jN3w0P4nyt2eVN3NQbZiulPsX7E8lWs3Rf4ahrexJox5QiXwPanEfj12Dv5cGTo" + "KVyVEe2oI8nOTbmi4qzmM18XTr9BJTFq2roe84Y7y+Z40S6+X3acqb10nWXdRbpjYTO4hqGWWDy6" + "BPFNam5cnTeHzWEwnQduHqCneSDPno2an705s9bnA3GGIjoz6TJtGph++d0gzW13/e22jcVvNXL6" + "R0wvkRjmt9LHv6ZSuOvuxu8WV7LbHxNPf/Kjk3JAFAEAEJgAAAJ4fTxVQ2oDAAAAAAEAo6E/QBUA" + "o7sMQBQAo70tQBQAo70lQBQAo5NaAAEAo5NlQBQAoqKiAAAAAA==" +) + +# Live S7-1200 G2 EXPLORE response for DB "Data_block_1" after adding a member of +# a user UDT ("elettrovalvola"), from the interface stream onward. The blob +# bundles the DB interface and the UDT's own definition; the UDT member is +# expanded into Member.fields. +_DB_UDT_INTERFACE_BLOB = base64.b64decode( + "eH1mBSsTrVjZbltHDP0Vww99YzXD2dPYQGynaIAWKdogD30RZm2NOpEiyUbQry/nbrqLZbtIX2zd" + "hWc4HJLn8D7fGe2yMU4P5HLCQGerloO6s7u5ajseJXt7QhwpCf3XcX7o9tZQz6jp7d+u6QczDDVz" + "QwZ0VHJxfvXu49tXde/rUD1b81VMKaFzCJJzBhKlBu+DBsw866xidNx03acqpfz1gBfnbUQpyLJq" + "HqKM9pZtbnZvN5nCvrrmOLhr6LiNIueW7Iis8t3F+Uekjs7aZOpqrEOXzlT0LvRC6IHM6h4JWV2r" + "K3xre9l1k7f1cLA9pZ/z5z/by8vXH3bU2OhgzuI91UzhKaBMGVjyBqQJBQJ3CpzmQoRUCreFjH66" + "3ddMOaO1E0cmMGuwKiuQyWWwXnoQztNtxUIxtOpqYqMJPBbpwHK0IK3h9EtwEEwznrzgLvC5jbC+" + "OJsF2Mg1eeYT+CIQHFpETmsra+Y26Ivh0jEQJQWQ2WZwIXki5qS95pZLpRc2THDpgwEfkfYT0NN+" + "cgSjXLJGG8fZYp1inZFCKGAmOEqWEsAxSfFTzhiupNUYFjEITGZOS2jHySYoA7Y4Cd6m7FmMnik/" + "t1GqoOdaAzlO+5H0uiu6QEx0P0ehYhZzmxiDUCJ4iKEIkN5LsJounUcbg8lBmDK3YVlrn7QDHiL5" + "pgyFWQsDxkbrUegoSmuz6rPn8tjWq86YM3RkJjHrEjhrC0gRGHhhI6AwPJXEOMGej4i/NoBej+4v" + "J9J01vhZ36TyXT4cdpsHf/dA/NtXZSXDi3Oakbrqp8pwzro3b8ysYPT1j1pfSU0Ltwqt7Wa1jTyG" + "v27zczX2bHRJPndMUgtd8R5k36BsdnlNpLj294dNxxrH1tyU9ftmYw2TuGahMdrg0vb+br/eVwVN" + "fzfbdbyNd6cBeQfI2QJRHP07bKssW5dd/rL+tKmuzgHVAKh7RPLooWkuM1zZ4/69Xf/67maOZHsk" + "gT2SPIGkBqTbJ5G07JHUAkMPGOlJDDfsSy8wTI8RN5/3JwJ0jDjaHsme2Jft8XbE9Bv/D/FRfjbu" + "st8kshOwroelMbPOQjR2+2dhdZ9yyOeAmh0BP+9bJ2/3BHrwJ8PITR9HFI2bYoHKj9Hc0Sh0eAEq" + "sj6mKBd4TxRrz7i8P6JWoH335X5z+GH6entvrjDZ4IEc4qSP2lehGW2yGxLP9hOGfpiS/MDoDZnP" + "ifuy1eHV4BSMqMdCV1MgGl2PWqHxZnO26UyHCymPF2jt6Akef+vRS25srUYvCSbH1qMVm/QfLvT4" + "iWsBVl2c2g7fJBYdZyWh4TNYGy/WGq8qt9C/gWv6fjvM0iOyWJ7fIgceZ45BiK4bRTh9vEoeiysm" + "gsVIVKawVEVCUsZw4YJUuQQ55Yaxc8OUO9GzlPA1HWeqVi9ErdJTUcv7vEFlSEYozvlS1n7DLk4o" + "2xq65tEydovPebViUZMwAmQfmHsl+St03zunEIX945Gvgs8ZDNPpSw3e//L7tkb3xe9PPhc99/ZE" + "zg+J9R/1ey2u99vus0CtxlPSZCrWuY5JSx/BRZNJejOSxNYwCCmRcNdRRrRTse4VBsclyWA6cpCM" + "kWzPhtSqliJm7atcn4vBlB0vlglQ0lWBrzMEbSSoSIJY5EDiN89tsko+o/WgKCdJ3NJkEHikP8KV" + "KJmn8WkhiBWS9C+YSAZHT0K1JmcxGrIMgkfvMcbFOtLohEEU2k+UIEuq+2EOQtS8uJBJ88qFbxhz" + "oNkNVBA01BVE8JJ0uKIzVYbAcui6zQvFbdbJJBpnQHmaLcilAp5LBBLXqIV1tF82F7ePqsMS157Y" + "upL1t8hCgol/3d7vn4J5kRik4XWzzg//hwAkUvK7T/kptKMIXAiFo/xrZNcSZNiXHBSXOuHSoAIp" + "0qfj/YigNCPABb/XCt6MK/hRch9V8oJgzZhTR7/liETtiEPZjEEX3Lhqvtq/Xh0pp78z/fBy+S/w" + "ygU7o5NyQBQBABCYAAACeH08VUNqAwAAAAABAKOhP0AVAKO7DEAUAKO9LUAUAKO9JUAUAKOTWgAB" + "AKOTZUAUAKKiogAAAAA=" +) + + +def test_parse_ident_container_addresses(): + xml = ( + "" + 'Bool' + '' + 'Word' + '' + "" + ) + tags = parse_ident_container(xml) + assert tags == [ + Tag("input_1", "Bool", "%I0.0", 10, 0, 0), + Tag("mtag_word", "Word", "%MW102", 13, 102, None), + ] + + +def test_tags_from_live_g2_m_area(): + tags = tags_from_explore(_M_AREA_BLOB) + got = {t.name: (t.data_type, t.address) for t in tags} + assert got == { + "merker_1": ("Bool", "%M0.2"), + "mtag_byte": ("Byte", "%MB100"), + "mtag_word": ("Word", "%MW102"), + "mtag_dword": ("DWord", "%MD104"), + "mtag_bool": ("Bool", "%M108.0"), + } + + +def test_tags_from_explore_no_stream_returns_empty(): + assert tags_from_explore(b"no zlib stream here") == [] + + +def test_parse_block_interface_types(): + # types are RID references into the SoftDataType table; a RID + # outside the 0x0200_00XX namespace is a complex type left as raw hex. + xml = ( + "" + '' + '' + '' + '' + "" + ) + assert parse_block_interface(xml) == [ + Member("flag", "Bool", 9, 0, 0x02000001), + Member("speed", "Int", 12, 16, 0x02000005), + Member("gain", "Real", 14, 32, 0x02000008), + Member("timer", "0x0300ABCD", 20, 48, 0x0300ABCD), + ] + + +def test_parse_block_interface_fb_sections(): + # FB/OB interfaces are organized in sections; the section-header s + # (Input/Static, no RID) must be skipped, real members tagged with their + # section. Complex members carry a ready-made ``Type`` attribute (arrays), + # scalar ones resolve from the RID. + xml = ( + '' + '' + '' + "" + "" + '' + '' + "" + '' + '' + '' + '' + "" + "" + ) + assert parse_block_interface(xml) == [ + Member("puls_start_stop", "Bool", 9, 0, 0x02000001, "Input"), + Member("num_step", "Int", 19, 0, 0x02000005, "Static"), + Member("timers", "Array[0..5] of IEC_TIMER", 20, 16, 0x0201001F, "Static"), + Member("echo", "Array[0..2] of Bool", 21, 784, 0x02010001, "Static"), + ] + + +def test_parse_block_interface_udt_member_expands_fields(): + # A UDT member carries a readable ``Type`` ("MyUdt", TIA quotes), a RID in + # the UDT object namespace (0x91......) and a ``SubPartIndex`` pointing at the + # UDT's definition embedded under . That definition is expanded + # into Member.fields, not surfaced as top-level members. + xml = ( + '' + '' + '' + "" + '' + '' + '' + "" + "" + ) + assert parse_block_interface(xml) == [ + Member("flag", "Bool", 9, 0, 0x02000001, ""), + Member( + "valve", + '"MyUdt"', + 10, + 16, + 0x91000001, + "", + fields=[ + Member("open_cmd", "Bool", 1, 0, 0x02000001, ""), + Member("level", "Int", 2, 16, 0x02000005, ""), + ], + ), + ] + + +def test_block_interface_from_live_g2_db_with_udt(): + members = block_interface_from_explore(_DB_UDT_INTERFACE_BLOB) + # 12 top-level members; the embedded UDT definition is not surfaced flat. + assert len(members) == 12 + valve = members[-1] + assert (valve.name, valve.data_type, valve.rid) == ( + "elettrovalvola_1", + '"elettrovalvola"', + 0x91000001, + ) + assert [(f.name, f.data_type) for f in valve.fields] == [ + ("fc_apertura", "Bool"), + ("fc_chiusura", "Bool"), + ("stato_ev", "Int"), + ("allarme_ev", "Int"), + ("cons_ev", "Bool"), + ("perc_apertura", "Real"), + ] + + +def test_block_interface_from_live_g2_db(): + members = block_interface_from_explore(_DB_INTERFACE_BLOB) + got = [(m.name, m.data_type) for m in members] + assert got == [ + ("selettore_man_auto", "Bool"), + ("puls_start_stop_ciclo", "Bool"), + ("setpoint_freq_motore", "Int"), + ("kp_PID", "Real"), + ("ki_PID", "Real"), + ("kd_PID", "Real"), + ("cons_motore", "Bool"), + ("retroazione_motore", "Int"), + ("temperatura_motore", "Int"), + ("tensione_misurata", "Real"), + ("corrente_misurata", "Real"), + ] + + +def test_datablocks_from_explore_standard_zlib(): + # PlcContentInfo is a standard zlib stream (78 da) embedded in a larger BLOB; + # trailing bytes follow it, so decoding must raw-inflate and tolerate them. + xml = ( + b'' + b'
' + b'
' + b"" + ) + stream = zlib.compress(xml, 9) # -> 78 da header + assert stream[:2] == b"\x78\xda" + payload = b"\x00\x01prefix" + stream + b"trailing-blob-bytes\xff\x00" + assert datablocks_from_explore(payload) == [ + DataBlock("Data_block_1", 25, 2316173337), + ] + + +def test_datablocks_from_explore_no_stream_returns_empty(): + assert datablocks_from_explore(b"no zlib here") == []