#!/usr/bin/env python3 """ Generate a fake company's Unified Namespace (UNS) for OtOpcUa, grounded in the REAL AVEVA Galaxy "DEV" hierarchy pulled from the MxAccess gateway (galaxy-hierarchy.json). Mapping onto OtOpcUa's config-DB model (ServerCluster -> Namespace(Kind=Equipment) -> UnsArea -> UnsLine -> Equipment -> Tag), with the canonical UNS path: Enterprise / Site / Area / Line / Equipment / Signal Real grounding: * The 40 $TestMachine instances (TestMachine_001..040) become Equipment ("machines"). * The Galaxy areas they are deployed in (TestArea / TestArea2 / TestArea3) become UNS Areas. * Each machine's real DELMIA + MES receiver children (resolved from Galaxy containment) become tag sub-folders ("delmia" / "mes") whose Tags carry the REAL MXAccess fullTagReference. * Process Tags map the real $TestMachine UDAs (TestChangingInt, TestDouble, ...) to believable signal names while preserving the real reference. Synthesized ("fake company") parts: enterprise/site/area/line names, machine archetypes, and asset metadata (manufacturer/model/serial/SAP id). Everything synthetic is clearly under generated identifiers; every node keeps a `source` block linking back to the real Galaxy object. OtOpcUa UNS-segment rule enforced: Area/Line/Equipment names match ^[a-z0-9-]{1,32}$. """ import json, re, collections, hashlib SRC = "galaxy-hierarchy.json" SEG = re.compile(r"^[a-z0-9-]{1,32}$") def seg(s): assert SEG.match(s), f"invalid UNS segment: {s!r}" return s # ---------------------------------------------------------------- load source gal = json.load(open(SRC)) objs = gal["objects"] byid = {o["gobjectId"]: o for o in objs} def name(i): return byid[i]["tagName"] if i in byid else None def children(pid): return [o for o in objs if o.get("parentGobjectId") == pid] def chain_has(o, frag): return any(frag in c for c in o.get("templateChain", [])) # the 40 top-level machines, grouped by their Galaxy area AREAS_GAL = {} # galaxy area tag -> list of machine objects (sorted by number) for o in objs: if "$TestMachine" in o.get("templateChain", []): # exactly $TestMachine (not .DelmiaReceiver etc.) area = name(o.get("parentGobjectId")) AREAS_GAL.setdefault(area, []).append(o) for a in AREAS_GAL: AREAS_GAL[a].sort(key=lambda o: o["tagName"]) # resolve each machine's real DELMIA / MES receiver child tag names def receiver_refs(machine): delmia = mes = None for c in children(machine["gobjectId"]): if chain_has(c, "DelmiaReceiver"): delmia = c elif chain_has(c, "MESReceiver"): mes = c return delmia, mes # ---------------------------------------------------------------- fake company ENTERPRISE = "northwind" # fictional consumer-goods manufacturer SITE = "birmingham" # fictional plant (maps to Galaxy node/galaxy DEV) GALAXY_NODE = "DESKTOP-6JL3KKO" GALAXY = "DEV" # Galaxy area -> (uns area segment, friendly label, line plan) # line plan: list of (line-segment, count) consuming machines in order. AREA_PLAN = { "TestArea": ("filling", "Filling & Capping", [("line-1", 7), ("line-2", 6), ("line-3", 6)]), "TestArea2": ("blending", "Blending & CIP", [("cip-1", 1)]), "TestArea3": ("packaging", "Packaging & Palletizing", [("pack-1", 5), ("pack-2", 5), ("pack-3", 5), ("pack-4", 5)]), } # believable machine archetypes cycled within a line, per area ARCHETYPES = { "filling": [("rinser","krones","Hydra"),("filler","sidel","SF300"), ("capper","khs","Innofill"),("labeler","krones","Contiroll"), ("inspector","antares-vision","Vmax"),("coder","videojet","1580")], "blending": [("blender","spx-flow","APV-R5")], "packaging": [("cartoner","marchesini","MC820"),("case-packer","bosch","Elematic"), ("palletizer","fanuc","M410"),("stretch-wrapper","lantech","Q300"), ("checkweigher","mettler-toledo","C3570")], } def asset(area_seg, archetype, model_series, gtag, gidx): mfr_model = {a[0]:(a[1],a[2]) for a in ARCHETYPES[area_seg]} mfr, model = mfr_model[archetype] h = int(hashlib.sha1(gtag.encode()).hexdigest(), 16) return { "manufacturer": mfr, "model": f"{model} Srs{2 + h % 4}", "serialNumber": f"SN26{gidx:05d}", "sapId": f"100{200000 + h % 700000}", "zTag": f"Z{area_seg[:3].upper()}{gidx:04d}", "hardwareRevision": f"H{1 + h % 4}.{h % 10}", "softwareRevision": f"S{2 + h % 3}.{(h>>4) % 10}", "equipmentClassRef": f"urn:northwind:equipclass:{archetype}", } # ----------------------------------------------------- signal (Tag) templates def dtype(galaxy_type): return { "Boolean":"Boolean","Integer":"Int32","Double":"Double","Float":"Float", "String":"String","Time":"DateTime","ElapsedTime":"Double", "InternationalizedString":"String", }.get(galaxy_type, "String") # process signals: (uns tag name, real $TestMachine UDA, galaxy type, access, flags) PROCESS_SIGNALS = [ ("speed-rpm", "TestChangingInt", "Integer", "ReadOnly", {}), ("production-count", "TestHistoryValue","Integer", "ReadOnly", {"historize": True}), ("temperature-c", "TestDouble", "Double", "ReadOnly", {}), ("pressure-bar", "TestFloat", "Float", "ReadOnly", {}), ("cycle-time-s", "TestDuration", "ElapsedTime","ReadOnly",{}), ("last-cycle-ts", "TestDateTime", "Time", "ReadOnly", {}), ("safety-interlock", "ProtectedValue", "Boolean", "ReadWrite", {"secured": True}), ("maint-lockout", "ProtectedValue1", "Boolean", "ReadWrite", {"secured": True}), ("motor-fault", "TestAlarm001", "Boolean", "ReadOnly", {"alarm": True}), ("over-temp", "TestAlarm002", "Boolean", "ReadOnly", {"alarm": True}), ("jam-detected", "TestAlarm003", "Boolean", "ReadOnly", {"alarm": True}), ("in-alarm", "InAlarm", "Boolean", "ReadOnly", {}), ] # DELMIA receiver signals (real $DelmiaReceiver attributes) DELMIA_SIGNALS = [ ("work-order", "WorkOrderNumber", "String", "ReadWrite", {}), ("part-number", "PartNumber", "String", "ReadWrite", {}), ("job-step", "JobStepNumber", "String", "ReadWrite", {}), ("recipe-path", "DownloadPath", "String", "ReadWrite", {}), ("recipe-dl-req", "RecipeDownloadFlag","Boolean","ReadWrite", {}), ("recipe-done", "RecipeProcessedFlag","Boolean","ReadOnly", {}), ("recipe-result", "RecipeProcessResult","Boolean","ReadOnly", {}), ("ready", "ReadyFlag", "Boolean","ReadOnly", {}), ] # MES receiver signals (real $MESReceiver attributes) MES_SIGNALS = [ ("move-in-req", "MoveInFlag", "Boolean","ReadWrite", {}), ("move-in-batch", "MoveInBatchID", "Integer","ReadWrite", {}), ("move-in-operator","MoveInOperatorName", "String", "ReadOnly", {}), ("move-in-ok", "MoveInSuccessFlag", "Boolean","ReadOnly", {}), ("move-out-req", "MoveOutFlag", "Boolean","ReadWrite", {}), ("move-out-batch", "MoveOutBatchID", "Integer","ReadWrite", {}), ("move-out-ok", "MoveOutSuccessfulFlag","Boolean","ReadOnly", {}), ("container-id", "MoveOutMesContainerNum","String","ReadOnly", {}), ] def attr_set(obj): return {a["attributeName"] for a in obj["attributes"]} if obj else set() def make_tags(equipment_id, machine, delmia, mes): tags = [] def add(folder, signals, ref_obj): if ref_obj is None: return ref_tag = ref_obj["tagName"] have = attr_set(ref_obj) for uns_name, attr, gtype, access, flags in signals: if attr not in have: continue # only emit signals whose real attribute exists on THIS instance full = f"{ref_tag}.{attr}" tid = f"tag-{equipment_id}-{(folder+'-' if folder else '')}{uns_name}" t = { "tagId": tid, "name": uns_name, "folderPath": folder, "dataType": dtype(gtype), "accessLevel": access, "historize": bool(flags.get("historize", False)), "tagConfig": { "isAlarm": bool(flags.get("alarm", False)), "secured": bool(flags.get("secured", False)), }, "source": { "namespaceKind": "SystemPlatform", "fullTagReference": full, "galaxyType": gtype, }, } tags.append(t) add("", PROCESS_SIGNALS, machine) add("delmia", DELMIA_SIGNALS, delmia) add("mes", MES_SIGNALS, mes) return tags # ---------------------------------------------------------------- build UNS cluster_id = f"{ENTERPRISE}-{SITE}" uns_areas, uns_lines, equipment = [], [], [] src_hierarchy = {"galaxyNode": GALAXY_NODE, "galaxy": GALAXY, "areas": []} gidx = 0 for gal_area, (area_seg, area_label, line_plan) in AREA_PLAN.items(): machines = AREAS_GAL.get(gal_area, []) area_id = f"area-{area_seg}" uns_areas.append({ "unsAreaId": area_id, "clusterId": cluster_id, "name": seg(area_seg), "notes": area_label, "source": {"galaxyArea": gal_area, "galaxyGobjectId": next((o["gobjectId"] for o in objs if o["tagName"] == gal_area), None)}, }) src_area = {"galaxyArea": gal_area, "machines": []} # consume machines line-by-line per the plan cursor = 0 for line_seg, count in line_plan: line_id = f"line-{area_seg}-{line_seg}" uns_lines.append({ "unsLineId": line_id, "unsAreaId": area_id, "name": seg(line_seg), "notes": f"{area_label} {line_seg}", }) archetypes = ARCHETYPES[area_seg] for j in range(count): if cursor >= len(machines): break m = machines[cursor]; cursor += 1; gidx += 1 archetype, _, model_series = archetypes[j % len(archetypes)] eq_seg = seg(f"{archetype}-{gidx:02d}") eq_id = f"eq-{area_seg}-{line_seg}-{eq_seg}" delmia, mes = receiver_refs(m) a = asset(area_seg, archetype, model_series, m["tagName"], gidx) tags = make_tags(eq_id, m, delmia, mes) equipment.append({ "equipmentId": eq_id, "unsLineId": line_id, "name": eq_seg, "machineCode": m["tagName"], "manufacturer": a["manufacturer"], "model": a["model"], "serialNumber": a["serialNumber"], "sapId": a["sapId"], "zTag": a["zTag"], "hardwareRevision": a["hardwareRevision"], "softwareRevision": a["softwareRevision"], "assetLocation": f"{ENTERPRISE}/{SITE}/{area_seg}/{line_seg}", "equipmentClassRef": a["equipmentClassRef"], "unsPath": f"{ENTERPRISE}/{SITE}/{area_seg}/{line_seg}/{eq_seg}", "source": { "namespaceKind": "SystemPlatform", "galaxyTag": m["tagName"], "galaxyGobjectId": m["gobjectId"], "templateChain": m["templateChain"], "galaxyArea": gal_area, "delmiaReceiverTag": delmia["tagName"] if delmia else None, "mesReceiverTag": mes["tagName"] if mes else None, }, "tags": tags, }) src_area["machines"].append({ "tag": m["tagName"], "gobjectId": m["gobjectId"], "unsEquipment": f"{ENTERPRISE}/{SITE}/{area_seg}/{line_seg}/{eq_seg}", "delmiaReceiver": delmia["tagName"] if delmia else None, "mesReceiver": mes["tagName"] if mes else None, }) src_hierarchy["areas"].append(src_area) tag_count = sum(len(e["tags"]) for e in equipment) doc = { "$schema": "internal://otopcua/uns-export/v1", "kind": "fake-company-uns", "description": ("Fictional company UNS generated around the real AVEVA Galaxy " "'DEV' TestMachine instances and the areas they are deployed in. " "Modeled on OtOpcUa's Cluster->Namespace->UnsArea->UnsLine->" "Equipment->Tag schema. Synthetic naming/asset metadata; every " "node links back to its real Galaxy source."), "generatedFrom": { "file": SRC, "galaxyNode": GALAXY_NODE, "galaxy": GALAXY, "gatewayEndpoint": "http://10.100.0.48:5120 (MxAccess gateway, gRPC)", "sourceObjectCount": len(objs), }, "company": {"name": "Northwind Consumer Products", "enterprise": ENTERPRISE, "site": SITE, "siteDescription": "Birmingham bottling & packaging plant"}, "cluster": {"clusterId": cluster_id, "name": f"{ENTERPRISE}-{SITE}-uns", "enterprise": ENTERPRISE, "site": SITE, "redundancyMode": "Hot", "nodeCount": 2}, "namespace": {"namespaceId": f"{cluster_id}-equipment", "clusterId": cluster_id, "kind": "Equipment", "namespaceUri": f"urn:{ENTERPRISE}:{SITE}:uns"}, "uns": {"unsAreas": uns_areas, "unsLines": uns_lines, "equipment": equipment}, "stats": {"areas": len(uns_areas), "lines": len(uns_lines), "equipment": len(equipment), "tags": tag_count}, "sourceGalaxyHierarchy": src_hierarchy, } json.dump(doc, open("company-uns.json", "w"), indent=2) # ---------------------------------------------------------------- tree view lines = [] lines.append(f"Northwind Consumer Products — Unified Namespace") lines.append(f"(generated from Galaxy {GALAXY_NODE}\\{GALAXY}; " f"{len(equipment)} machines, {tag_count} signals)") lines.append("") lines.append(f"{ENTERPRISE}") lines.append(f"└─ {SITE}") area_by_id = {a["unsAreaId"]: a for a in uns_areas} lines_by_area = collections.defaultdict(list) for l in uns_lines: lines_by_area[l["unsAreaId"]].append(l) eq_by_line = collections.defaultdict(list) for e in equipment: eq_by_line[e["unsLineId"]].append(e) for ai, a in enumerate(uns_areas): a_last = ai == len(uns_areas) - 1 a_pre = " " + ("└─ " if a_last else "├─ ") lines.append(f" {'└─' if a_last else '├─'} {a['name']}/ ({a['notes']}; from Galaxy {a['source']['galaxyArea']})") als = lines_by_area[a["unsAreaId"]] a_cont = " " if a_last else " │ " for li, l in enumerate(als): l_last = li == len(als) - 1 lines.append(f"{a_cont}{'└─' if l_last else '├─'} {l['name']}/") eqs = eq_by_line[l["unsLineId"]] l_cont = a_cont + (" " if l_last else "│ ") for ei, e in enumerate(eqs): e_last = ei == len(eqs) - 1 ntags = len(e["tags"]) lines.append(f"{l_cont}{'└─' if e_last else '├─'} {e['name']} " f"[{e['manufacturer']} {e['model']}] ← {e['source']['galaxyTag']} ({ntags} signals)") open("company-uns.tree.txt", "w").write("\n".join(lines) + "\n") print(json.dumps(doc["stats"], indent=2)) print("\nsample equipment:") e = equipment[0] print(f" {e['unsPath']} machineCode={e['machineCode']} delmia={e['source']['delmiaReceiverTag']} mes={e['source']['mesReceiverTag']}") print(" sample tags:") for t in e["tags"][:3] + [e["tags"][12]] + [e["tags"][-1]]: print(f" {t['folderPath'] or '.'}/{t['name']:18s} {t['dataType']:8s} {t['accessLevel']:9s} -> {t['source']['fullTagReference']}") print("\nwrote: company-uns.json, company-uns.tree.txt")