feat(uns-loader): verify-equipment — recursive Equipment UNS tree browse + leaf count

browse_summary assumes the flat 2-level Galaxy hierarchy; the Equipment tree is deep
(Area/Line/Equipment/[FolderPath]/Signal). Add browse_tree (recursive leaf descent) + a
verify-equipment subcommand that reports/asserts the leaf signal count (--expect N), for
verifying OtOpcUa equipment-namespace structure materialisation. Smoke-tested against a live
:4840 (40 folders / 396 leaf signals).
This commit is contained in:
Joseph Doherty
2026-06-06 15:25:17 -04:00
parent eb26bf3248
commit fd34e25cb1
+63 -1
View File
@@ -298,6 +298,64 @@ def sample_values(endpoint, n):
return [("<browse error>", 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)