diff --git a/otopcua-uns-loader/otopcua_uns.py b/otopcua-uns-loader/otopcua_uns.py index 7b6e9a3..c0d1404 100644 --- a/otopcua-uns-loader/otopcua_uns.py +++ b/otopcua-uns-loader/otopcua_uns.py @@ -298,6 +298,64 @@ def sample_values(endpoint, n): return [("", str(e), "?")] +def browse_tree(endpoint, max_depth=8): + """Recursively descend the OtOpcUa address space and count leaf variables, returning + (folder_count, leaf_count, leaf_paths). A node with no children is a leaf signal — this + correctly handles the DEEP Equipment UNS tree (Area/Line/Equipment/[FolderPath]/Signal), + unlike browse_summary which assumes the flat 2-level Galaxy hierarchy.""" + import asyncio + from asyncua import Client + + async def walk(node, path, depth, acc): + if depth >= max_depth: + return + for ch in await node.get_children(): + try: + name = (await ch.read_browse_name()).Name + except Exception: + continue + child_path = path + "/" + name + grandkids = await ch.get_children() + if grandkids: + acc["folders"] += 1 + await walk(ch, child_path, depth + 1, acc) + else: + acc["leaves"] += 1 + acc["paths"].append(child_path) + + async def run(): + acc = {"folders": 0, "leaves": 0, "paths": []} + async with Client(endpoint) as c: + for k in await c.nodes.objects.get_children(): + if (await k.read_browse_name()).Name != "OtOpcUa": + continue + await walk(k, "OtOpcUa", 0, acc) + return acc["folders"], acc["leaves"], acc["paths"] + try: + return asyncio.run(run()) + except Exception as e: + return (f"<{type(e).__name__}: {e}>", 0, []) + + +def cmd_verify_equipment(args): + """Browse the full UNS tree by friendly Area/Line/Equipment/Signal names and report the leaf + signal count. With --expect N, exit non-zero unless exactly N leaf signals are present (the + equipment-namespace structure-materialisation check).""" + folders, leaves, paths = browse_tree(args.opcua_endpoint) + print(f"equipment tree : {folders} folder(s), {leaves} leaf signal(s) on {args.opcua_endpoint}") + for p in sorted(paths)[:args.show]: + print(f" {p}") + if len(paths) > args.show: + print(f" … and {len(paths) - args.show} more") + if args.expect is not None: + passed = leaves == args.expect + print("VERIFY-EQUIPMENT:", + f"PASS ({leaves} == {args.expect})" if passed + else f"FAIL (expected {args.expect}, found {leaves})") + return 0 if passed else 1 + return 0 + + # ── arg parsing ───────────────────────────────────────────────────────────── def main(argv): p = argparse.ArgumentParser(description="Reloadable populate + verify for the OtOpcUa galaxy UNS.") @@ -318,13 +376,17 @@ def main(argv): vp = sub.add_parser("verify") vp.add_argument("--wait", action="store_true", help="poll until the deploy lands") vp.add_argument("--wait-seconds", type=int, default=120) + ep = sub.add_parser("verify-equipment", + help="recursively browse the Equipment UNS tree + count leaf signals") + ep.add_argument("--expect", type=int, default=None, help="assert exactly N leaf signals") + ep.add_argument("--show", type=int, default=20, help="how many leaf paths to print") a = p.parse_args(argv) a.mssql = dict(host=a.sql_host, port=a.sql_port, user=a.sql_user, password=a.sql_password, database=a.sql_db) return { "generate": cmd_generate, "populate": cmd_populate, "clean": cmd_clean, - "status": cmd_status, "verify": cmd_verify, + "status": cmd_status, "verify": cmd_verify, "verify-equipment": cmd_verify_equipment, }[a.cmd](a)