eba71c4ca8
Merge of the two project copies into one self-contained plugin (the superset: variable_op + variables.py, full pcg_op runtime/declarative/preset ops, the CreateWidgetFloatAnimation widget tool, and full Voxel graph authoring). Made fully project- and machine-agnostic — no hardcoded paths: - New src/projectPaths.js auto-detects the host .uproject (walk-up), project name, Editor build target, log file, and engine install (EngineAssociation via launcher manifest/registry, else installed-engine scan). All overridable via UE_* env vars. - Rewired buildOrchestrator/insights/launcher/insightsExporter/server.js and the Python workers (cpp_scaffold, live_coding, apply_graph, console) off the old C:/Github/ihy, IHY*, E:/UE_Versions and UE_5.7 literals onto the resolver. - Voxel made optional: Build.cs auto-detects the Voxel plugin (env UE_MCP_WITH_VOXEL override) and the C++ compiles to stubs under WITH_MCP_VOXEL, so the module builds in projects without Voxel; .uplugin marks Voxel optional. - De-branded the agent-gateway and docs; scrubbed a leaked API key; excluded node_modules/Binaries/Intermediate/__pycache__/secrets from the repo. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
303 lines
9.7 KiB
Python
303 lines
9.7 KiB
Python
"""Network ops for Blueprint assets.
|
|
|
|
Ops:
|
|
audit_bp {bp_path}
|
|
set_actor_flags {bp_path, replicates?, replicate_movement?, net_load_on_client?,
|
|
always_relevant?, only_relevant_to_owner?, use_owner_relevancy?,
|
|
net_cull_distance_squared?, net_update_frequency?,
|
|
min_net_update_frequency?, net_priority?, net_dormancy?}
|
|
set_component_flags {bp_path, component, replicated?, net_addressable?}
|
|
set_custom_event_rpc {bp_path, function?, event?, guid?, mode, reliable?}
|
|
apply_profile {bp_path, profile}
|
|
|
|
RPC modes for set_custom_event_rpc:
|
|
none | server | client | multicast
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import traceback
|
|
|
|
try:
|
|
import unreal # type: ignore
|
|
except ImportError:
|
|
unreal = None
|
|
|
|
|
|
ACTOR_PROP_MAP = {
|
|
"replicates": "replicates",
|
|
"replicate_movement": "replicate_movement",
|
|
"net_load_on_client": "net_load_on_client",
|
|
"always_relevant": "always_relevant",
|
|
"only_relevant_to_owner": "only_relevant_to_owner",
|
|
"use_owner_relevancy": "net_use_owner_relevancy",
|
|
"net_cull_distance_squared": "net_cull_distance_squared",
|
|
"net_update_frequency": "net_update_frequency",
|
|
"min_net_update_frequency": "min_net_update_frequency",
|
|
"net_priority": "net_priority",
|
|
"net_dormancy": "net_dormancy",
|
|
}
|
|
|
|
|
|
def _load_bp(path: str):
|
|
bp = unreal.EditorAssetLibrary.load_asset(path)
|
|
if not isinstance(bp, unreal.Blueprint):
|
|
raise RuntimeError(f"not a Blueprint: {path!r}")
|
|
return bp
|
|
|
|
|
|
def _generated_class(bp_path: str):
|
|
cls_path = f"{bp_path}.{bp_path.rsplit('/', 1)[-1]}_C"
|
|
cls = unreal.load_class(None, cls_path)
|
|
if cls is None:
|
|
raise RuntimeError(f"generated class not found: {cls_path}")
|
|
return cls
|
|
|
|
|
|
def _cdo(bp_path: str):
|
|
return unreal.get_default_object(_generated_class(bp_path))
|
|
|
|
|
|
def _lib():
|
|
L = getattr(unreal, "MCPGraphLibrary", None)
|
|
if L is None:
|
|
raise RuntimeError("MCPGraphLibrary missing - rebuild C++ plugin")
|
|
return L
|
|
|
|
|
|
def _find_graph(bp, name: str):
|
|
L = _lib()
|
|
if name in (None, "", "EventGraph"):
|
|
return L.find_event_graph(bp)
|
|
return L.find_function_graph(bp, name)
|
|
|
|
|
|
def _get_prop(obj, prop):
|
|
try:
|
|
return obj.get_editor_property(prop)
|
|
except Exception:
|
|
return getattr(obj, prop)
|
|
|
|
|
|
def _set_prop(obj, prop, value):
|
|
try:
|
|
obj.set_editor_property(prop, value)
|
|
except Exception:
|
|
setattr(obj, prop, value)
|
|
|
|
|
|
def _save_compile(bp):
|
|
L = getattr(unreal, "MCPGraphLibrary", None)
|
|
messages = []
|
|
try:
|
|
if L:
|
|
messages = list(L.compile_with_messages(bp))
|
|
else:
|
|
unreal.KismetEditorUtilities.compile_blueprint(bp)
|
|
except Exception as e:
|
|
messages = [f"WARN||compile failed from network.py: {e}"]
|
|
unreal.EditorAssetLibrary.save_loaded_asset(bp)
|
|
return messages
|
|
|
|
|
|
def _read_actor_flags(cdo):
|
|
out = {}
|
|
for public_name, prop in ACTOR_PROP_MAP.items():
|
|
try:
|
|
v = _get_prop(cdo, prop)
|
|
out[public_name] = str(v) if public_name == "net_dormancy" else v
|
|
except Exception as e:
|
|
out[public_name] = {"error": str(e)}
|
|
return out
|
|
|
|
|
|
def audit_bp(args: dict) -> dict:
|
|
bp_path = args["bp_path"]
|
|
cdo = _cdo(bp_path)
|
|
result = {
|
|
"ok": True,
|
|
"bp_path": bp_path,
|
|
"actor": _read_actor_flags(cdo),
|
|
"components": [],
|
|
"custom_events": [],
|
|
"notes": [],
|
|
}
|
|
|
|
for name in args.get("components") or []:
|
|
item = {"name": name}
|
|
try:
|
|
comp = _get_prop(cdo, name)
|
|
item["class"] = comp.get_class().get_name()
|
|
for prop in ("replicates", "is_replicated", "net_addressable"):
|
|
try:
|
|
item[prop] = _get_prop(comp, prop)
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
item["error"] = str(e)
|
|
result["components"].append(item)
|
|
|
|
try:
|
|
import apply_graph
|
|
raw = apply_graph.read_graph(bp_path, args.get("function") or "EventGraph")
|
|
for node in raw.get("graph", {}).get("nodes", []):
|
|
if node.get("class") == "K2Node_CustomEvent":
|
|
result["custom_events"].append({
|
|
"guid": node.get("guid"),
|
|
"title": node.get("title"),
|
|
"target": node.get("target"),
|
|
})
|
|
result["notes"].append("Custom event RPC flags require set_custom_event_rpc; read_graph does not expose them.")
|
|
except Exception as e:
|
|
result["notes"].append(f"custom event scan failed: {e}")
|
|
|
|
return result
|
|
|
|
|
|
def set_actor_flags(args: dict) -> dict:
|
|
bp_path = args["bp_path"]
|
|
bp = _load_bp(bp_path)
|
|
cdo = _cdo(bp_path)
|
|
changed = {}
|
|
errors = {}
|
|
for public_name, prop in ACTOR_PROP_MAP.items():
|
|
if public_name not in args:
|
|
continue
|
|
try:
|
|
_set_prop(cdo, prop, args[public_name])
|
|
changed[public_name] = args[public_name]
|
|
except Exception as e:
|
|
errors[public_name] = str(e)
|
|
messages = _save_compile(bp)
|
|
return {"ok": not errors, "bp_path": bp_path, "changed": changed, "errors": errors, "compile_messages": messages}
|
|
|
|
|
|
def _component_from_cdo(cdo, name: str):
|
|
try:
|
|
return _get_prop(cdo, name)
|
|
except Exception:
|
|
pass
|
|
getter = getattr(cdo, "get_component_by_class", None)
|
|
if getter:
|
|
# Fallback is intentionally conservative: caller should pass the BP component variable name.
|
|
return None
|
|
return None
|
|
|
|
|
|
def set_component_flags(args: dict) -> dict:
|
|
bp_path = args["bp_path"]
|
|
bp = _load_bp(bp_path)
|
|
cdo = _cdo(bp_path)
|
|
comp_name = args["component"]
|
|
comp = _component_from_cdo(cdo, comp_name)
|
|
if comp is None:
|
|
return {"ok": False, "error": f"component not found on CDO by name: {comp_name}"}
|
|
|
|
changed = {}
|
|
errors = {}
|
|
for public_name, candidates in {
|
|
"replicated": ("replicates", "is_replicated"),
|
|
"net_addressable": ("net_addressable",),
|
|
}.items():
|
|
if public_name not in args:
|
|
continue
|
|
done = False
|
|
for prop in candidates:
|
|
try:
|
|
_set_prop(comp, prop, args[public_name])
|
|
changed[public_name] = args[public_name]
|
|
done = True
|
|
break
|
|
except Exception:
|
|
continue
|
|
if not done:
|
|
errors[public_name] = f"no writable property among {candidates}"
|
|
messages = _save_compile(bp)
|
|
return {"ok": not errors, "bp_path": bp_path, "component": comp_name, "changed": changed, "errors": errors, "compile_messages": messages}
|
|
|
|
|
|
def set_custom_event_rpc(args: dict) -> dict:
|
|
bp_path = args["bp_path"]
|
|
bp = _load_bp(bp_path)
|
|
graph = _find_graph(bp, args.get("function") or "EventGraph")
|
|
if graph is None:
|
|
return {"ok": False, "error": "graph not found"}
|
|
|
|
target = args.get("guid") or args.get("event")
|
|
if not target:
|
|
return {"ok": False, "error": "set_custom_event_rpc requires guid or event"}
|
|
mode = args.get("mode") or "none"
|
|
reliable = bool(args.get("reliable", False))
|
|
|
|
L = _lib()
|
|
setter = getattr(L, "set_custom_event_network_flags", None)
|
|
if setter is None:
|
|
return {
|
|
"ok": False,
|
|
"error": "MCPGraphLibrary.set_custom_event_network_flags is unavailable. Rebuild/restart UE after the plugin C++ update.",
|
|
"requested": {"target": target, "mode": mode, "reliable": reliable},
|
|
}
|
|
ok = bool(setter(bp, graph, target, mode, reliable))
|
|
messages = _save_compile(bp) if ok else []
|
|
return {"ok": ok, "bp_path": bp_path, "target": target, "mode": mode, "reliable": reliable, "compile_messages": messages}
|
|
|
|
|
|
def apply_profile(args: dict) -> dict:
|
|
profile = args["profile"]
|
|
bp_path = args["bp_path"]
|
|
if profile == "replicated_world_actor":
|
|
return set_actor_flags({
|
|
"bp_path": bp_path,
|
|
"replicates": True,
|
|
"net_load_on_client": True,
|
|
"replicate_movement": bool(args.get("replicate_movement", False)),
|
|
})
|
|
if profile == "always_relevant_world_actor":
|
|
return set_actor_flags({
|
|
"bp_path": bp_path,
|
|
"replicates": True,
|
|
"net_load_on_client": True,
|
|
"always_relevant": True,
|
|
"replicate_movement": bool(args.get("replicate_movement", False)),
|
|
})
|
|
return {"ok": False, "error": f"unknown profile {profile!r}", "available": ["replicated_world_actor", "always_relevant_world_actor"]}
|
|
|
|
|
|
OPS = {
|
|
"audit_bp": audit_bp,
|
|
"set_actor_flags": set_actor_flags,
|
|
"set_component_flags": set_component_flags,
|
|
"set_custom_event_rpc": set_custom_event_rpc,
|
|
"apply_profile": apply_profile,
|
|
}
|
|
|
|
|
|
def entrypoint(payload_json: str) -> str:
|
|
try:
|
|
p = json.loads(payload_json)
|
|
except Exception as e:
|
|
return _tee({"ok": False, "error": f"bad json: {e}"}, {})
|
|
fn = OPS.get(p.get("op"))
|
|
if not fn:
|
|
return _tee({"ok": False, "error": f"unknown op {p.get('op')!r}", "available": list(OPS)}, p)
|
|
try:
|
|
return _tee(fn(p), p)
|
|
except Exception as e:
|
|
return _tee({"ok": False, "error": str(e), "traceback": traceback.format_exc()}, p)
|
|
|
|
|
|
def _tee(result: dict, p: dict) -> str:
|
|
txt = json.dumps(result)
|
|
try:
|
|
if unreal is not None:
|
|
sd = unreal.Paths.project_saved_dir() + "MCP/"
|
|
os.makedirs(sd, exist_ok=True)
|
|
rid = p.get("__rid") or "last"
|
|
with open(sd + f"{rid}.json", "w", encoding="utf-8") as f:
|
|
f.write(txt)
|
|
unreal.log("[MCP-NETWORK] " + txt[:1500])
|
|
except Exception:
|
|
pass
|
|
return txt
|