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>
1238 lines
46 KiB
Python
1238 lines
46 KiB
Python
"""PCG (Procedural Content Generation) graph MCP ops.
|
|
|
|
Pure-Python worker — unlike the Voxel worker, PCG's data model
|
|
(UPCGGraph / UPCGNode / UPCGSettings) is engine-exposed to Python, so no C++
|
|
shim is required.
|
|
|
|
"All workers" = every UPCGSettings subclass bound into the `unreal` module.
|
|
This is enumerated dynamically (see `_settings_registry`), which means engine
|
|
PCG nodes AND PCGExtendedToolkit (PCGEx) nodes are both supported automatically
|
|
without a hardcoded table — any plugin that ships UPCGSettings subclasses gets
|
|
picked up for free.
|
|
|
|
Node identity: PCG nodes have no stable public GUID exposed to Python, so a node
|
|
is referenced by:
|
|
* "input" / "output" -> the graph's special I/O nodes
|
|
* an integer index -> position in get_nodes()
|
|
* a UObject name (node.get_name())
|
|
* a settings class short name / title (first match)
|
|
|
|
Ops (authoring):
|
|
api {} introspect live PCG reflection (method names, etc.)
|
|
create_graph {path} new empty UPCGGraph asset
|
|
list_node_types {filter?, limit?} catalogue of spawnable workers (all UPCGSettings subclasses)
|
|
read_graph {asset_path} dump nodes + pins + edges
|
|
add_node {asset_path, node, x?, y?, title?} spawn a worker (node = class name/path/title)
|
|
delete_node {asset_path, node}
|
|
connect {asset_path, from:[node,pin?], to:[node,pin?]}
|
|
disconnect {asset_path, from:[node,pin?], to:[node,pin?]}
|
|
list_node_properties {asset_path, node} editor properties of a node's settings
|
|
set_node_property {asset_path, node, property, value}
|
|
move_nodes {asset_path, moves:[{node,x,y}]} best-effort editor positions
|
|
save {asset_path}
|
|
open_asset {asset_path}
|
|
|
|
Ops (declarative authoring + layout):
|
|
apply_graph {asset_path, nodes:[{id,type,x?,y?,title?,properties?}], edges:[{from:[id,pin?],to:[id,pin?]}], outputs?:[{from:[id,pin?],pin?}], clear_existing?, auto_layout?, save?}
|
|
build a whole graph in one call (auto-laid-out by default)
|
|
auto_layout/tidy_graph/layout {asset_path} re-flow existing nodes into edge-depth columns
|
|
|
|
Ops (parameters + subgraph):
|
|
list_params {asset_path} dump graph UserParameters bag (read-only — UE blocks Python editing)
|
|
set_subgraph {asset_path, node, subgraph} point a Subgraph node at a target PCG graph asset
|
|
|
|
Ops (level / runtime):
|
|
spawn_volume {graph?, location?, rotation?, scale?, name?, generate?} spawn a PCG Volume actor, assign graph, generate
|
|
set_actor_graph {actor, graph, generate?} assign a graph to an existing actor's PCGComponent
|
|
generate {actor, force?} (re)generate a PCG actor's component
|
|
cleanup {actor} clear an actor's generated content
|
|
list_components {} level actors carrying a PCGComponent + their graph/state
|
|
|
|
Ops (presets):
|
|
scaffold {path|asset_path, template, create?, save?} build a working graph from a named template
|
|
(template='__list__' lists: surface_scatter, points_grid_spawn, density_filter_scatter)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import traceback
|
|
|
|
try:
|
|
import unreal # type: ignore
|
|
except ImportError:
|
|
unreal = None
|
|
|
|
try:
|
|
import graph_layout # shared deterministic layout helper
|
|
except Exception: # noqa: BLE001 - layout is optional; ops degrade gracefully
|
|
graph_layout = None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# low-level helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _pcg_base():
|
|
base = getattr(unreal, "PCGSettings", None)
|
|
if base is None:
|
|
raise RuntimeError(
|
|
"unreal.PCGSettings not found - is the PCG plugin enabled and the "
|
|
"Python API loaded? (Project uses PCGExtendedToolkit which depends on it.)"
|
|
)
|
|
return base
|
|
|
|
|
|
def _load_graph(path: str):
|
|
asset = unreal.EditorAssetLibrary.load_asset(path)
|
|
if asset is None:
|
|
raise RuntimeError(f"asset not found: {path!r}")
|
|
if not isinstance(asset, unreal.PCGGraph):
|
|
raise RuntimeError(
|
|
f"asset is not a UPCGGraph: {path!r} ({asset.get_class().get_name()})")
|
|
return asset
|
|
|
|
|
|
def _save(asset):
|
|
try:
|
|
unreal.EditorAssetLibrary.save_loaded_asset(asset)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _short(cls) -> str:
|
|
"""'PCGDensityFilterSettings' from a class object. Note: a bound unreal
|
|
class object's get_name()/get_path_name() are unbound (need an instance),
|
|
so use the Python type __name__."""
|
|
name = getattr(cls, "__name__", None)
|
|
if name:
|
|
return name
|
|
try:
|
|
return cls.get_name()
|
|
except Exception:
|
|
return str(cls)
|
|
|
|
|
|
def _class_path(cls) -> str:
|
|
"""'/Script/PCG.PCGSurfaceSamplerSettings' — derived via the CDO because a
|
|
bound class object's get_path_name() is unbound in the unreal Python API."""
|
|
try:
|
|
return unreal.get_default_object(cls).get_class().get_path_name()
|
|
except Exception:
|
|
return _short(cls)
|
|
|
|
|
|
def _title_for(cls) -> str:
|
|
"""Best-effort human title from the settings CDO."""
|
|
try:
|
|
cdo = unreal.get_default_object(cls)
|
|
for getter in ("get_default_node_title", "get_default_node_name"):
|
|
fn = getattr(cdo, getter, None)
|
|
if fn:
|
|
try:
|
|
val = str(fn())
|
|
if val:
|
|
return val
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
name = _short(cls)
|
|
# strip the conventional PCG/PCGEx prefix + Settings suffix for readability
|
|
base = name[:-8] if name.endswith("Settings") else name
|
|
return base
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# node-type catalogue ("all workers")
|
|
# --------------------------------------------------------------------------- #
|
|
def _settings_registry() -> dict:
|
|
"""{class_object: {name, title, category, plugin}} for every bound
|
|
UPCGSettings subclass. Dynamic — picks up PCG + PCGEx + any plugin."""
|
|
base = _pcg_base()
|
|
out = {}
|
|
for attr in dir(unreal):
|
|
obj = getattr(unreal, attr, None)
|
|
try:
|
|
if not isinstance(obj, type) or obj is base:
|
|
continue
|
|
if not issubclass(obj, base):
|
|
continue
|
|
except TypeError:
|
|
continue
|
|
name = _short(obj)
|
|
plugin = "PCGEx" if name.startswith("PCGEx") else "PCG"
|
|
out[obj] = {"name": name, "title": _title_for(obj), "plugin": plugin}
|
|
return out
|
|
|
|
|
|
def _resolve_settings_class(ref: str):
|
|
"""ref -> settings UClass. Accepts short name, /Script path, or title."""
|
|
ref = (ref or "").strip()
|
|
if not ref:
|
|
raise RuntimeError("node type ref is empty")
|
|
# 1) explicit /Script path
|
|
if ref.startswith("/Script/"):
|
|
cls = unreal.load_class(None, ref)
|
|
if cls is not None:
|
|
return cls
|
|
# 2) direct attribute on unreal module (exact or with Settings suffix)
|
|
for cand in (ref, ref + "Settings", "PCG" + ref, "PCG" + ref + "Settings",
|
|
"PCGEx" + ref, "PCGEx" + ref + "Settings"):
|
|
obj = getattr(unreal, cand, None)
|
|
if isinstance(obj, type):
|
|
return obj
|
|
# 3) registry lookup by name / title (case-insensitive)
|
|
low = ref.lower()
|
|
reg = _settings_registry()
|
|
for cls, meta in reg.items():
|
|
if meta["name"].lower() == low or meta["title"].lower() == low:
|
|
return cls
|
|
for cls, meta in reg.items(): # loose contains match
|
|
if low in meta["name"].lower() or low in meta["title"].lower():
|
|
return cls
|
|
raise RuntimeError(f"unknown PCG node type: {ref!r} (try list_node_types)")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# node referencing within a graph
|
|
# --------------------------------------------------------------------------- #
|
|
def _graph_nodes(graph) -> list:
|
|
# UPCGGraph exposes 'nodes' as a property (no get_nodes()).
|
|
for src in ("nodes",):
|
|
try:
|
|
val = graph.get_editor_property(src)
|
|
if val is not None:
|
|
return list(val)
|
|
except Exception:
|
|
pass
|
|
fn = getattr(graph, "get_nodes", None)
|
|
if callable(fn):
|
|
try:
|
|
return list(fn())
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def _io_node(graph, which: str):
|
|
fn = getattr(graph, "get_input_node" if which == "input" else "get_output_node", None)
|
|
return fn() if fn else None
|
|
|
|
|
|
def _resolve_node(graph, ref):
|
|
"""ref -> UPCGNode. Accepts 'input'/'output', int index, UObject name,
|
|
or settings class short-name/title (first match)."""
|
|
if ref is None:
|
|
raise RuntimeError("node ref is None")
|
|
if isinstance(ref, str) and ref.lower() in ("input", "output"):
|
|
node = _io_node(graph, ref.lower())
|
|
if node is None:
|
|
raise RuntimeError(f"graph has no {ref} node")
|
|
return node
|
|
nodes = _graph_nodes(graph)
|
|
# integer index
|
|
try:
|
|
idx = int(ref)
|
|
if 0 <= idx < len(nodes):
|
|
return nodes[idx]
|
|
except (TypeError, ValueError):
|
|
pass
|
|
low = str(ref).lower()
|
|
# exact UObject name
|
|
for n in nodes:
|
|
try:
|
|
if n.get_name().lower() == low:
|
|
return n
|
|
except Exception:
|
|
pass
|
|
# settings class short name / title
|
|
for n in nodes:
|
|
meta = _node_settings_meta(n)
|
|
if meta and (meta["class"].lower() == low or meta["title"].lower() == low):
|
|
return n
|
|
for n in nodes: # loose contains
|
|
meta = _node_settings_meta(n)
|
|
if meta and (low in meta["class"].lower() or low in meta["title"].lower()):
|
|
return n
|
|
raise RuntimeError(f"node not found in graph: {ref!r} (try read_graph)")
|
|
|
|
|
|
def _node_settings(node):
|
|
for getter in ("get_settings", "default_settings"):
|
|
fn = getattr(node, getter, None)
|
|
if callable(fn):
|
|
try:
|
|
s = fn()
|
|
if s is not None:
|
|
return s
|
|
except Exception:
|
|
pass
|
|
# property fallbacks
|
|
for prop in ("settings_interface", "default_settings"):
|
|
try:
|
|
s = node.get_editor_property(prop)
|
|
if s is not None:
|
|
# settings_interface wraps the settings
|
|
inner = getattr(s, "get_settings", None)
|
|
return inner() if callable(inner) else s
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _node_settings_meta(node):
|
|
s = _node_settings(node)
|
|
if s is None:
|
|
return None
|
|
cls = s.get_class()
|
|
return {"class": _short(cls), "title": _title_for(cls)}
|
|
|
|
|
|
def _pins(node, which: str) -> list:
|
|
# UPCGNode exposes 'input_pins'/'output_pins' as properties.
|
|
prop = "input_pins" if which == "in" else "output_pins"
|
|
try:
|
|
val = node.get_editor_property(prop)
|
|
if val is not None:
|
|
return list(val)
|
|
except Exception:
|
|
pass
|
|
fn = getattr(node, "get_input_pins" if which == "in" else "get_output_pins", None)
|
|
if callable(fn):
|
|
try:
|
|
return list(fn())
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def _pin_label(pin) -> str:
|
|
try:
|
|
props = pin.get_editor_property("properties")
|
|
return str(props.get_editor_property("label"))
|
|
except Exception:
|
|
pass
|
|
for getter in ("get_pin_label", "get_name"):
|
|
fn = getattr(pin, getter, None)
|
|
if callable(fn):
|
|
try:
|
|
return str(fn())
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# ops
|
|
# --------------------------------------------------------------------------- #
|
|
def api(args: dict) -> dict:
|
|
"""Dump live reflection so callers can confirm exposed method names. Handy
|
|
because PCG Python signatures shift between engine versions."""
|
|
def members(name):
|
|
cls = getattr(unreal, name, None)
|
|
if cls is None:
|
|
return None
|
|
return sorted(m for m in dir(cls) if not m.startswith("_"))
|
|
return {
|
|
"ok": True,
|
|
"PCGGraph": members("PCGGraph"),
|
|
"PCGNode": members("PCGNode"),
|
|
"PCGSettings": members("PCGSettings"),
|
|
"PCGPin": members("PCGPin"),
|
|
"settings_class_count": len(_settings_registry()),
|
|
}
|
|
|
|
|
|
def create_graph(args: dict) -> dict:
|
|
path = str(args["path"]).rstrip("/")
|
|
pkg_dir, _, name = path.rpartition("/")
|
|
if not name or not pkg_dir:
|
|
raise RuntimeError(f"bad path: {path!r} (expected e.g. /Game/PCG/MyGraph)")
|
|
tools = unreal.AssetToolsHelpers.get_asset_tools()
|
|
factory = None
|
|
for fname in ("PCGGraphFactory", "PCGGraphFactory_New"):
|
|
fcls = getattr(unreal, fname, None)
|
|
if fcls is not None:
|
|
factory = fcls()
|
|
break
|
|
asset = tools.create_asset(name, pkg_dir, unreal.PCGGraph, factory)
|
|
if asset is None:
|
|
raise RuntimeError(f"failed to create PCG graph at {path}")
|
|
_save(asset)
|
|
return {"ok": True, "asset": asset.get_path_name()}
|
|
|
|
|
|
def list_node_types(args: dict) -> dict:
|
|
flt = str(args.get("filter") or "").lower()
|
|
limit = int(args.get("limit") or 0)
|
|
reg = _settings_registry()
|
|
items = []
|
|
for cls, meta in reg.items():
|
|
if flt and flt not in meta["name"].lower() and flt not in meta["title"].lower():
|
|
continue
|
|
items.append({
|
|
"name": meta["name"],
|
|
"title": meta["title"],
|
|
"plugin": meta["plugin"],
|
|
"path": _class_path(cls),
|
|
})
|
|
items.sort(key=lambda d: (d["plugin"], d["name"]))
|
|
total = len(items)
|
|
if limit > 0:
|
|
items = items[:limit]
|
|
return {"ok": True, "count": total, "returned": len(items), "node_types": items}
|
|
|
|
|
|
def read_graph(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
nodes = _graph_nodes(graph)
|
|
out_nodes = []
|
|
name_to_idx = {}
|
|
for i, n in enumerate(nodes):
|
|
try:
|
|
uname = n.get_name()
|
|
except Exception:
|
|
uname = f"node_{i}"
|
|
name_to_idx[uname] = i
|
|
meta = _node_settings_meta(n) or {"class": "?", "title": "?"}
|
|
out_nodes.append({
|
|
"index": i,
|
|
"name": uname,
|
|
"class": meta["class"],
|
|
"title": meta["title"],
|
|
"in_pins": [_pin_label(p) for p in _pins(n, "in")],
|
|
"out_pins": [_pin_label(p) for p in _pins(n, "out")],
|
|
})
|
|
|
|
# special I/O nodes (not always in get_nodes)
|
|
io = {}
|
|
for which in ("input", "output"):
|
|
node = _io_node(graph, which)
|
|
if node is not None:
|
|
io[which] = {
|
|
"in_pins": [_pin_label(p) for p in _pins(node, "in")],
|
|
"out_pins": [_pin_label(p) for p in _pins(node, "out")],
|
|
}
|
|
|
|
# edges: the same UPCGEdge object sits in both the source output-pin's and
|
|
# the destination input-pin's `edges` array. PCGEdge's InputPin/OutputPin
|
|
# aren't reflection-accessible, so we pair the two sides by edge identity.
|
|
all_nodes = list(nodes)
|
|
for which in ("input", "output"):
|
|
n = _io_node(graph, which)
|
|
if n is not None and n not in all_nodes:
|
|
all_nodes.append(n)
|
|
edges = _collect_edges(all_nodes)
|
|
return {"ok": True, "asset": graph.get_path_name(),
|
|
"node_count": len(out_nodes), "nodes": out_nodes,
|
|
"io": io, "edges": edges}
|
|
|
|
|
|
def _pin_edge_objs(pin) -> list:
|
|
try:
|
|
return list(pin.get_editor_property("edges") or [])
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _obj_key(obj) -> str:
|
|
for getter in ("get_path_name", "get_name"):
|
|
fn = getattr(obj, getter, None)
|
|
if callable(fn):
|
|
try:
|
|
return str(fn())
|
|
except Exception:
|
|
pass
|
|
return str(id(obj))
|
|
|
|
|
|
def _is_output_pin(pin, fallback_which: str) -> bool:
|
|
fn = getattr(pin, "is_output_pin", None)
|
|
if callable(fn):
|
|
try:
|
|
return bool(fn())
|
|
except Exception:
|
|
pass
|
|
return fallback_which == "out"
|
|
|
|
|
|
def _collect_edges(all_nodes) -> list:
|
|
"""Rebuild edges by pairing the shared UPCGEdge object found on both the
|
|
source output pin and the destination input pin."""
|
|
by_key = {}
|
|
for n in all_nodes:
|
|
try:
|
|
nname = n.get_name()
|
|
except Exception:
|
|
continue
|
|
for which in ("in", "out"):
|
|
for pin in _pins(n, which):
|
|
label = _pin_label(pin)
|
|
side = "out" if _is_output_pin(pin, which) else "in"
|
|
for e in _pin_edge_objs(pin):
|
|
rec = by_key.setdefault(_obj_key(e), {})
|
|
rec[side] = {"node": nname, "pin": label}
|
|
edges = []
|
|
for rec in by_key.values():
|
|
if "out" in rec and "in" in rec:
|
|
edges.append({
|
|
"from_node": rec["out"]["node"], "from_pin": rec["out"]["pin"],
|
|
"to_node": rec["in"]["node"], "to_pin": rec["in"]["pin"],
|
|
})
|
|
else: # only one side seen (endpoint outside enumerated nodes)
|
|
known = rec.get("out") or rec.get("in")
|
|
edges.append({"from_node": known["node"], "from_pin": known["pin"],
|
|
"to_node": "?", "to_pin": "?", "partial": True}
|
|
if known else {"partial": True})
|
|
return edges
|
|
|
|
|
|
def add_node(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
cls = _resolve_settings_class(str(args.get("node") or args.get("type") or ""))
|
|
node, _settings = _add_node_inner(graph, cls)
|
|
if node is None:
|
|
raise RuntimeError(f"failed to add node {_short(cls)}")
|
|
_apply_position(node, args)
|
|
title = args.get("title")
|
|
if title:
|
|
_set_node_title(node, str(title))
|
|
_save(graph)
|
|
return {"ok": True, "name": node.get_name(), "class": _short(cls),
|
|
"in_pins": [_pin_label(p) for p in _pins(node, "in")],
|
|
"out_pins": [_pin_label(p) for p in _pins(node, "out")]}
|
|
|
|
|
|
def _apply_position(node, args):
|
|
if "x" not in args and "y" not in args:
|
|
return
|
|
x, y = float(args.get("x", 0)), float(args.get("y", 0))
|
|
fn = getattr(node, "set_node_position", None)
|
|
if callable(fn):
|
|
# UPCGNode.set_node_position takes two floats (its getter returns an
|
|
# (x, y) tuple). Fall back to a single Vector2D for other node types.
|
|
for call in (lambda: fn(x, y), lambda: fn(unreal.Vector2D(x, y))):
|
|
try:
|
|
call()
|
|
return
|
|
except Exception:
|
|
continue
|
|
for px, py in (("node_pos_x", "node_pos_y"), ("position_x", "position_y")):
|
|
try:
|
|
node.set_editor_property(px, int(x))
|
|
node.set_editor_property(py, int(y))
|
|
return
|
|
except Exception:
|
|
continue
|
|
|
|
|
|
def _set_node_title(node, title):
|
|
for prop in ("node_title", "node_comment", "author_generated_node_name"):
|
|
try:
|
|
node.set_editor_property(prop, title)
|
|
return
|
|
except Exception:
|
|
continue
|
|
|
|
|
|
def delete_node(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
node = _resolve_node(graph, args.get("node"))
|
|
name = node.get_name()
|
|
fn = getattr(graph, "remove_node", None)
|
|
if not callable(fn):
|
|
raise RuntimeError("UPCGGraph.remove_node not exposed to Python")
|
|
fn(node)
|
|
_save(graph)
|
|
return {"ok": True, "removed": name}
|
|
|
|
|
|
def _pin_or_default(node, which: str, label):
|
|
"""Resolve a pin label; default to the single/first pin when omitted."""
|
|
pins = _pins(node, which)
|
|
if not pins:
|
|
raise RuntimeError(f"node {node.get_name()!r} has no {which} pins")
|
|
if label in (None, ""):
|
|
return _pin_label(pins[0])
|
|
low = str(label).lower()
|
|
for p in pins:
|
|
if _pin_label(p).lower() == low:
|
|
return _pin_label(p)
|
|
# not found -> surface available labels
|
|
raise RuntimeError(
|
|
f"{which} pin {label!r} not on {node.get_name()!r}; "
|
|
f"available: {[_pin_label(p) for p in pins]}")
|
|
|
|
|
|
def connect(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
f = args.get("from") or []
|
|
t = args.get("to") or []
|
|
from_node = _resolve_node(graph, f[0])
|
|
to_node = _resolve_node(graph, t[0])
|
|
from_pin = _pin_or_default(from_node, "out", f[1] if len(f) > 1 else None)
|
|
to_pin = _pin_or_default(to_node, "in", t[1] if len(t) > 1 else None)
|
|
fn = (getattr(graph, "add_labeled_edge", None)
|
|
or getattr(graph, "add_edge", None))
|
|
if not callable(fn):
|
|
raise RuntimeError("UPCGGraph exposes no add_labeled_edge/add_edge")
|
|
fn(from_node, unreal.Name(from_pin), to_node, unreal.Name(to_pin))
|
|
_save(graph)
|
|
return {"ok": True, "from": [from_node.get_name(), from_pin],
|
|
"to": [to_node.get_name(), to_pin]}
|
|
|
|
|
|
def disconnect(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
f = args.get("from") or []
|
|
t = args.get("to") or []
|
|
from_node = _resolve_node(graph, f[0])
|
|
to_node = _resolve_node(graph, t[0])
|
|
from_pin = _pin_or_default(from_node, "out", f[1] if len(f) > 1 else None)
|
|
to_pin = _pin_or_default(to_node, "in", t[1] if len(t) > 1 else None)
|
|
fn = getattr(graph, "remove_edge", None)
|
|
if not callable(fn):
|
|
raise RuntimeError("UPCGGraph.remove_edge not exposed to Python")
|
|
ok = bool(fn(from_node, unreal.Name(from_pin), to_node, unreal.Name(to_pin)))
|
|
_save(graph)
|
|
return {"ok": ok, "from": [from_node.get_name(), from_pin],
|
|
"to": [to_node.get_name(), to_pin]}
|
|
|
|
|
|
def list_node_properties(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
node = _resolve_node(graph, args.get("node"))
|
|
settings = _node_settings(node)
|
|
if settings is None:
|
|
raise RuntimeError(f"node {node.get_name()!r} has no settings object")
|
|
# Classify method vs editor-property at the CLASS level: getattr on the
|
|
# type returns the descriptor without invoking the getter, so we never
|
|
# trigger "property is protected and cannot be read" on the instance.
|
|
cls = type(settings)
|
|
props = []
|
|
for m in dir(settings):
|
|
if m.startswith("_"):
|
|
continue
|
|
try:
|
|
attr = getattr(cls, m, None)
|
|
except Exception:
|
|
attr = None
|
|
if callable(attr):
|
|
continue # it's a method, not a property
|
|
props.append(m)
|
|
return {"ok": True, "node": node.get_name(),
|
|
"settings_class": _short(settings.get_class()), "properties": props}
|
|
|
|
|
|
def _num_list(value):
|
|
"""Parse value (list or JSON string) into a list of floats, else None."""
|
|
if isinstance(value, str):
|
|
try:
|
|
value = json.loads(value.strip())
|
|
except Exception:
|
|
return None
|
|
if isinstance(value, list) and value and all(isinstance(n, (int, float)) for n in value):
|
|
return [float(n) for n in value]
|
|
return None
|
|
|
|
|
|
def _coerce_value(value):
|
|
"""MCP args may arrive stringified (the `value` field is untyped). Parse
|
|
JSON-ish strings to native types, and turn [x,y,z] number lists into the
|
|
matching unreal vector so struct properties (rotation/scale/offset) work."""
|
|
if isinstance(value, str):
|
|
try:
|
|
value = json.loads(value.strip())
|
|
except Exception:
|
|
return value # plain string (enum name, attribute name, etc.)
|
|
if isinstance(value, list) and value and all(isinstance(n, (int, float)) for n in value):
|
|
nums = [float(n) for n in value]
|
|
if len(nums) == 2:
|
|
return unreal.Vector2D(*nums)
|
|
if len(nums) == 3:
|
|
return unreal.Vector(*nums)
|
|
if len(nums) == 4:
|
|
return unreal.Vector4(*nums)
|
|
return value
|
|
|
|
|
|
def set_node_property(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
node = _resolve_node(graph, args.get("node"))
|
|
settings = _node_settings(node)
|
|
if settings is None:
|
|
raise RuntimeError(f"node {node.get_name()!r} has no settings object")
|
|
prop = str(args["property"])
|
|
raw = args.get("value")
|
|
value = _coerce_value(raw)
|
|
used = value
|
|
try:
|
|
settings.set_editor_property(prop, value)
|
|
except TypeError:
|
|
# A 3-number list is ambiguous: Vector vs Rotator (pitch,yaw,roll).
|
|
# _coerce_value picked Vector; retry the other struct kinds.
|
|
nums = _num_list(raw)
|
|
ok = False
|
|
if nums and len(nums) == 3:
|
|
for ctor in (unreal.Rotator, unreal.Vector):
|
|
try:
|
|
used = ctor(*nums)
|
|
settings.set_editor_property(prop, used)
|
|
ok = True
|
|
break
|
|
except Exception:
|
|
continue
|
|
if not ok:
|
|
raise
|
|
_save(graph)
|
|
return {"ok": True, "node": node.get_name(), "property": prop, "value": str(used)}
|
|
|
|
|
|
def move_nodes(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
moved = 0
|
|
for m in args.get("moves") or []:
|
|
try:
|
|
node = _resolve_node(graph, m["node"])
|
|
_apply_position(node, m)
|
|
moved += 1
|
|
except Exception:
|
|
continue
|
|
if moved:
|
|
_save(graph)
|
|
return {"ok": moved > 0, "moved": moved}
|
|
|
|
|
|
def save(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
_save(graph)
|
|
return {"ok": True, "asset": graph.get_path_name()}
|
|
|
|
|
|
def open_asset(args: dict) -> dict:
|
|
graph = _load_graph(args["asset_path"])
|
|
sub = unreal.get_editor_subsystem(unreal.AssetEditorSubsystem)
|
|
sub.open_editor_for_assets([graph])
|
|
return {"ok": True, "asset": graph.get_path_name()}
|
|
|
|
|
|
def _set_one_property(settings, prop: str, raw):
|
|
"""Coerce + assign one editor property on a settings object. Mirrors
|
|
set_node_property's logic so apply_graph and set_node_property agree."""
|
|
value = _coerce_value(raw)
|
|
try:
|
|
settings.set_editor_property(prop, value)
|
|
return value
|
|
except TypeError:
|
|
nums = _num_list(raw)
|
|
if nums and len(nums) == 3:
|
|
for ctor in (unreal.Rotator, unreal.Vector):
|
|
try:
|
|
used = ctor(*nums)
|
|
settings.set_editor_property(prop, used)
|
|
return used
|
|
except Exception:
|
|
continue
|
|
raise
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# declarative whole-graph authoring
|
|
# --------------------------------------------------------------------------- #
|
|
def apply_graph(args: dict) -> dict:
|
|
"""Build a graph in one call. Spec:
|
|
nodes: [{id, type|class, x?, y?, title?, properties?:{}}]
|
|
edges: [{from:[id, pin?], to:[id, pin?]}]
|
|
outputs: [{from:[id, pin?], pin?}] -> wire id to the graph Output node
|
|
clear_existing?, auto_layout?(default True), save?(default True)
|
|
'id' is a caller-chosen local handle reused by edges/outputs. 'input'/'output'
|
|
refer to the graph's I/O nodes. Returns {spawned, edges, errors, warnings}."""
|
|
graph = _load_graph(args["asset_path"])
|
|
result = {"ok": False, "asset": graph.get_path_name(),
|
|
"spawned": {}, "edges": [], "errors": [], "warnings": []}
|
|
|
|
if args.get("clear_existing"):
|
|
for n in list(_graph_nodes(graph)):
|
|
try:
|
|
graph.remove_node(n)
|
|
except Exception as exc: # noqa: BLE001
|
|
result["warnings"].append(f"clear: {exc}")
|
|
|
|
raw_nodes = list(args.get("nodes") or [])
|
|
raw_edges = list(args.get("edges") or [])
|
|
|
|
# deterministic layout: honour caller x/y, snap into readable columns
|
|
specs = raw_nodes
|
|
if args.get("auto_layout", True) and graph_layout is not None:
|
|
lay_nodes, lay_edges = [], []
|
|
for spec in raw_nodes:
|
|
x = spec.get("x", (spec.get("position") or [0, 0])[0] if spec.get("position") else 0)
|
|
y = spec.get("y", (spec.get("position") or [0, 0])[1] if spec.get("position") else 0)
|
|
lay_nodes.append({"id": spec.get("id"), "class": spec.get("type") or spec.get("class") or "",
|
|
"position": [float(x), float(y)]})
|
|
for e in raw_edges:
|
|
try:
|
|
lay_edges.append({"from": [e["from"][0]], "to": [e["to"][0]]})
|
|
except Exception:
|
|
pass
|
|
laid = graph_layout.normalize_specs(lay_nodes, lay_edges, domain="pcg")
|
|
pos_by_id = {s.get("id"): s.get("position") for s in laid}
|
|
specs = []
|
|
for spec in raw_nodes:
|
|
s = dict(spec)
|
|
p = pos_by_id.get(spec.get("id"))
|
|
if p:
|
|
s["x"], s["y"] = p[0], p[1]
|
|
specs.append(s)
|
|
|
|
local_map = {}
|
|
for spec in specs:
|
|
local_id = spec.get("id")
|
|
try:
|
|
ref = str(spec.get("type") or spec.get("class") or "")
|
|
cls = _resolve_settings_class(ref)
|
|
node, settings = _add_node_inner(graph, cls)
|
|
if settings is None:
|
|
settings = _node_settings(node)
|
|
_apply_position(node, spec)
|
|
if spec.get("title"):
|
|
_set_node_title(node, str(spec["title"]))
|
|
applied = {}
|
|
for pname, pval in (spec.get("properties") or {}).items():
|
|
try:
|
|
applied[pname] = str(_set_one_property(settings, pname, pval))
|
|
except Exception as exc: # noqa: BLE001
|
|
result["warnings"].append(f"node {local_id} prop {pname}: {exc}")
|
|
key = local_id or node.get_name()
|
|
local_map[key] = node
|
|
result["spawned"][key] = {"name": node.get_name(), "class": _short(cls),
|
|
"properties": applied,
|
|
"in_pins": [_pin_label(p) for p in _pins(node, "in")],
|
|
"out_pins": [_pin_label(p) for p in _pins(node, "out")]}
|
|
except Exception as exc: # noqa: BLE001
|
|
result["errors"].append(f"node {local_id or spec.get('type') or spec.get('class')}: {exc}")
|
|
|
|
def resolve(ref):
|
|
if isinstance(ref, str) and ref in local_map:
|
|
return local_map[ref]
|
|
return _resolve_node(graph, ref)
|
|
|
|
for e in raw_edges:
|
|
try:
|
|
f, t = e["from"], e["to"]
|
|
fn_node, tn_node = resolve(f[0]), resolve(t[0])
|
|
fp = _pin_or_default(fn_node, "out", f[1] if len(f) > 1 else None)
|
|
tp = _pin_or_default(tn_node, "in", t[1] if len(t) > 1 else None)
|
|
graph.add_edge(fn_node, unreal.Name(fp), tn_node, unreal.Name(tp))
|
|
result["edges"].append({"from": [fn_node.get_name(), fp], "to": [tn_node.get_name(), tp]})
|
|
except Exception as exc: # noqa: BLE001
|
|
result["errors"].append(f"edge {e}: {exc}")
|
|
|
|
for o in args.get("outputs") or []:
|
|
try:
|
|
src = o.get("from") or o
|
|
src_node = resolve(src[0])
|
|
sp = _pin_or_default(src_node, "out", src[1] if len(src) > 1 else None)
|
|
out_node = _io_node(graph, "output")
|
|
tp = _pin_or_default(out_node, "in", o.get("pin"))
|
|
graph.add_edge(src_node, unreal.Name(sp), out_node, unreal.Name(tp))
|
|
result["edges"].append({"from": [src_node.get_name(), sp], "to": ["output", tp]})
|
|
except Exception as exc: # noqa: BLE001
|
|
result["errors"].append(f"output {o}: {exc}")
|
|
|
|
if args.get("save", True):
|
|
_save(graph)
|
|
result["ok"] = not result["errors"]
|
|
result["node_count"] = len(_graph_nodes(graph))
|
|
return result
|
|
|
|
|
|
def _add_node_inner(graph, cls):
|
|
"""Spawn one node; returns (node, settings). Shared by add_node/apply_graph."""
|
|
fn = getattr(graph, "add_node_of_type", None)
|
|
if callable(fn):
|
|
res = fn(cls)
|
|
if isinstance(res, (tuple, list)):
|
|
return res[0], (res[1] if len(res) > 1 else None)
|
|
return res, None
|
|
settings = unreal.new_object(cls, graph)
|
|
adder = getattr(graph, "add_node", None) or getattr(graph, "add_node_copy", None)
|
|
if not callable(adder):
|
|
raise RuntimeError("UPCGGraph exposes no add_node/add_node_of_type")
|
|
return adder(settings), settings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# layout
|
|
# --------------------------------------------------------------------------- #
|
|
def auto_layout(args: dict) -> dict:
|
|
"""Re-flow existing nodes into edge-depth columns (left-to-right by data flow)."""
|
|
if graph_layout is None:
|
|
return {"ok": False, "error": "graph_layout helper unavailable"}
|
|
graph = _load_graph(args["asset_path"])
|
|
dump = read_graph({"asset_path": args["asset_path"]})
|
|
lay_nodes = [{"name": n["name"], "class": n["class"]} for n in dump.get("nodes", [])]
|
|
lay_edges = [{"from": [e.get("from_node")], "to": [e.get("to_node")]}
|
|
for e in dump.get("edges", []) if not e.get("partial")]
|
|
moves = graph_layout.tidy_existing_nodes(lay_nodes, lay_edges, domain="pcg")
|
|
applied = 0
|
|
for m in moves:
|
|
try:
|
|
node = _resolve_node(graph, m["guid"])
|
|
_apply_position(node, {"x": m["x"], "y": m["y"]})
|
|
applied += 1
|
|
except Exception:
|
|
continue
|
|
_save(graph)
|
|
return {"ok": True, "moved": applied, "count": len(moves)}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# graph parameters (read-only: UE does not expose FInstancedPropertyBag editing)
|
|
# --------------------------------------------------------------------------- #
|
|
def list_params(args: dict) -> dict:
|
|
"""Dump the graph's UserParameters property bag. Note: UE5.7 does NOT expose
|
|
property-bag *editing* to Python, so params must be authored in-editor; this
|
|
op reports what already exists (name/type/default) so graphs can be wired."""
|
|
graph = _load_graph(args["asset_path"])
|
|
try:
|
|
bag = graph.get_editor_property("user_parameters")
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"ok": False, "error": f"no user_parameters: {exc}"}
|
|
info = {"ok": True, "asset": graph.get_path_name(), "params": []}
|
|
try:
|
|
info["raw"] = str(bag.export_text())
|
|
except Exception:
|
|
pass
|
|
return info
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# subgraph wiring
|
|
# --------------------------------------------------------------------------- #
|
|
def set_subgraph(args: dict) -> dict:
|
|
"""Point a Subgraph node at a target PCG graph asset.
|
|
{asset_path, node, subgraph: '/Game/.../OtherGraph'}"""
|
|
graph = _load_graph(args["asset_path"])
|
|
node = _resolve_node(graph, args.get("node"))
|
|
settings = _node_settings(node)
|
|
if settings is None:
|
|
raise RuntimeError(f"node {node.get_name()!r} has no settings")
|
|
target = unreal.EditorAssetLibrary.load_asset(str(args["subgraph"]))
|
|
if target is None:
|
|
raise RuntimeError(f"subgraph asset not found: {args.get('subgraph')!r}")
|
|
last = None
|
|
for prop in ("subgraph_instance", "subgraph", "subgraph_override"):
|
|
try:
|
|
settings.set_editor_property(prop, target)
|
|
_save(graph)
|
|
return {"ok": True, "node": node.get_name(), "property": prop,
|
|
"subgraph": target.get_path_name()}
|
|
except Exception as exc: # noqa: BLE001
|
|
last = exc
|
|
raise RuntimeError(f"could not assign subgraph (tried subgraph_instance/subgraph/subgraph_override): {last}")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# level / runtime integration (PCG component + volume)
|
|
# --------------------------------------------------------------------------- #
|
|
def _ess():
|
|
return unreal.get_editor_subsystem(unreal.EditorActorSubsystem)
|
|
|
|
|
|
def _find_actor(name: str):
|
|
for a in _ess().get_all_level_actors():
|
|
if not a:
|
|
continue
|
|
try:
|
|
if a.get_actor_label() == name or a.get_name() == name or a.get_path_name() == name:
|
|
return a
|
|
except Exception:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _pcg_component(actor):
|
|
"""Return the actor's UPCGComponent (PCGVolume.pcg_component, or a component
|
|
search for arbitrary actors)."""
|
|
try:
|
|
c = actor.get_editor_property("pcg_component")
|
|
if c is not None:
|
|
return c
|
|
except Exception:
|
|
pass
|
|
try:
|
|
return actor.get_component_by_class(unreal.PCGComponent)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _graph_output_stats(comp):
|
|
"""Best-effort point/output count after generation."""
|
|
try:
|
|
out = comp.get_generated_graph_output()
|
|
items = list(out) if out is not None else []
|
|
return {"output_data": len(items)}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def spawn_volume(args: dict) -> dict:
|
|
"""Spawn a PCG Volume actor, assign a graph, optionally generate.
|
|
{graph?, location?, rotation?, scale?, name?, generate?(default True)}"""
|
|
cls = getattr(unreal, "PCGVolume", None) or unreal.load_class(None, "/Script/PCG.PCGVolume")
|
|
if cls is None:
|
|
return {"ok": False, "error": "PCGVolume class unavailable"}
|
|
loc = _vec(args.get("location"))
|
|
rot = _rot(args.get("rotation"))
|
|
actor = _ess().spawn_actor_from_class(cls, loc, rot)
|
|
if actor is None:
|
|
return {"ok": False, "error": "spawn returned None"}
|
|
scale = args.get("scale")
|
|
if scale is not None:
|
|
actor.set_actor_scale3d(_vec(scale, (1, 1, 1)))
|
|
if args.get("name"):
|
|
actor.set_actor_label(str(args["name"]))
|
|
comp = _pcg_component(actor)
|
|
res = {"ok": True, "actor": actor.get_actor_label(), "path": actor.get_path_name(),
|
|
"has_component": comp is not None}
|
|
if comp is not None and args.get("graph"):
|
|
g = _load_graph(args["graph"])
|
|
comp.set_graph(g)
|
|
res["graph"] = g.get_path_name()
|
|
if args.get("generate", True):
|
|
comp.generate(True)
|
|
res["generated"] = True
|
|
res.update(_graph_output_stats(comp))
|
|
return res
|
|
|
|
|
|
def set_actor_graph(args: dict) -> dict:
|
|
"""Assign a PCG graph to an existing actor's component. {actor, graph, generate?}"""
|
|
actor = _find_actor(str(args["actor"]))
|
|
if actor is None:
|
|
return {"ok": False, "error": f"actor not found: {args.get('actor')!r}"}
|
|
comp = _pcg_component(actor)
|
|
if comp is None:
|
|
return {"ok": False, "error": "actor has no PCGComponent"}
|
|
g = _load_graph(args["graph"])
|
|
comp.set_graph(g)
|
|
res = {"ok": True, "actor": actor.get_actor_label(), "graph": g.get_path_name()}
|
|
if args.get("generate", False):
|
|
comp.generate(True)
|
|
res["generated"] = True
|
|
res.update(_graph_output_stats(comp))
|
|
return res
|
|
|
|
|
|
def generate(args: dict) -> dict:
|
|
"""Generate (or regenerate) a PCG actor's component. {actor, force?(True)}"""
|
|
actor = _find_actor(str(args["actor"]))
|
|
if actor is None:
|
|
return {"ok": False, "error": f"actor not found: {args.get('actor')!r}"}
|
|
comp = _pcg_component(actor)
|
|
if comp is None:
|
|
return {"ok": False, "error": "actor has no PCGComponent"}
|
|
comp.generate(bool(args.get("force", True)))
|
|
res = {"ok": True, "actor": actor.get_actor_label(), "generated": True}
|
|
res.update(_graph_output_stats(comp))
|
|
return res
|
|
|
|
|
|
def cleanup(args: dict) -> dict:
|
|
"""Clear generated content on a PCG actor's component. {actor}"""
|
|
actor = _find_actor(str(args["actor"]))
|
|
if actor is None:
|
|
return {"ok": False, "error": f"actor not found: {args.get('actor')!r}"}
|
|
comp = _pcg_component(actor)
|
|
if comp is None:
|
|
return {"ok": False, "error": "actor has no PCGComponent"}
|
|
remove = bool(args.get("remove_components", True))
|
|
try:
|
|
comp.cleanup(remove) # UE5.7: cleanup(remove_components)
|
|
except TypeError:
|
|
comp.cleanup() # older signature fallback
|
|
return {"ok": True, "actor": actor.get_actor_label(), "cleaned": True}
|
|
|
|
|
|
def list_components(args: dict) -> dict:
|
|
"""Enumerate level actors that carry a PCGComponent + their graph/state."""
|
|
found = []
|
|
for a in _ess().get_all_level_actors():
|
|
comp = _pcg_component(a) if a else None
|
|
if comp is None:
|
|
continue
|
|
entry = {"actor": a.get_actor_label(), "path": a.get_path_name(),
|
|
"class": a.get_class().get_name()}
|
|
try:
|
|
gi = comp.get_editor_property("graph_instance")
|
|
g = gi.get_mutable_pcg_graph() if gi is not None else None
|
|
entry["graph"] = g.get_path_name() if g is not None else None
|
|
except Exception:
|
|
entry["graph"] = None
|
|
try:
|
|
entry["generated"] = bool(comp.get_editor_property("generated"))
|
|
except Exception:
|
|
pass
|
|
found.append(entry)
|
|
return {"ok": True, "count": len(found), "components": found}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# scaffolding presets — build a working graph end-to-end in one call
|
|
# --------------------------------------------------------------------------- #
|
|
_TEMPLATES = {
|
|
"surface_scatter": {
|
|
"desc": "Input(surface) -> Surface Sampler -> Transform Points -> Static Mesh Spawner -> Output",
|
|
"nodes": [
|
|
{"id": "sampler", "type": "PCGSurfaceSamplerSettings", "x": 0, "y": 0},
|
|
{"id": "transform", "type": "PCGTransformPointsSettings", "x": 1, "y": 0},
|
|
{"id": "spawner", "type": "PCGStaticMeshSpawnerSettings", "x": 2, "y": 0},
|
|
],
|
|
"edges": [
|
|
{"from": ["input"], "to": ["sampler"]},
|
|
{"from": ["sampler"], "to": ["transform"]},
|
|
{"from": ["transform"], "to": ["spawner"]},
|
|
],
|
|
"outputs": [{"from": ["spawner"]}],
|
|
},
|
|
"points_grid_spawn": {
|
|
"desc": "Create Points Grid -> Static Mesh Spawner -> Output",
|
|
"nodes": [
|
|
{"id": "grid", "type": "PCGCreatePointsGridSettings", "x": 0, "y": 0},
|
|
{"id": "spawner", "type": "PCGStaticMeshSpawnerSettings", "x": 1, "y": 0},
|
|
],
|
|
"edges": [{"from": ["grid"], "to": ["spawner"]}],
|
|
"outputs": [{"from": ["spawner"]}],
|
|
},
|
|
"density_filter_scatter": {
|
|
"desc": "Input -> Surface Sampler -> Density Filter -> Static Mesh Spawner -> Output",
|
|
"nodes": [
|
|
{"id": "sampler", "type": "PCGSurfaceSamplerSettings", "x": 0, "y": 0},
|
|
{"id": "filter", "type": "PCGDensityFilterSettings", "x": 1, "y": 0},
|
|
{"id": "spawner", "type": "PCGStaticMeshSpawnerSettings", "x": 2, "y": 0},
|
|
],
|
|
"edges": [
|
|
{"from": ["input"], "to": ["sampler"]},
|
|
{"from": ["sampler"], "to": ["filter"]},
|
|
{"from": ["filter"], "to": ["spawner"]},
|
|
],
|
|
"outputs": [{"from": ["spawner"]}],
|
|
},
|
|
}
|
|
|
|
|
|
def scaffold(args: dict) -> dict:
|
|
"""Create (or populate) a PCG graph from a named template, fully wired.
|
|
{path|asset_path, template, create?(True), save?(True)}.
|
|
template='__list__' lists available templates."""
|
|
template = str(args.get("template") or "").strip()
|
|
if template in ("", "__list__", "list"):
|
|
return {"ok": True, "templates": {k: v["desc"] for k, v in _TEMPLATES.items()}}
|
|
tpl = _TEMPLATES.get(template)
|
|
if tpl is None:
|
|
return {"ok": False, "error": f"unknown template {template!r}",
|
|
"available": list(_TEMPLATES)}
|
|
path = args.get("path") or args.get("asset_path")
|
|
if not path:
|
|
return {"ok": False, "error": "path (or asset_path) required"}
|
|
asset_path = path
|
|
if args.get("create", True) and not unreal.EditorAssetLibrary.does_asset_exist(path):
|
|
created = create_graph({"path": path})
|
|
asset_path = created.get("asset", path)
|
|
res = apply_graph({"asset_path": asset_path, "clear_existing": args.get("clear_existing", False),
|
|
"nodes": tpl["nodes"], "edges": tpl["edges"], "outputs": tpl["outputs"],
|
|
"auto_layout": True, "save": args.get("save", True)})
|
|
res["template"] = template
|
|
res["asset"] = asset_path
|
|
return res
|
|
|
|
|
|
def _vec(v, default=(0, 0, 0)):
|
|
if v is None:
|
|
v = default
|
|
if isinstance(v, str):
|
|
try:
|
|
v = json.loads(v)
|
|
except Exception:
|
|
v = default
|
|
return unreal.Vector(float(v[0]), float(v[1]), float(v[2]))
|
|
|
|
|
|
def _rot(v, default=(0, 0, 0)):
|
|
if v is None:
|
|
v = default
|
|
if isinstance(v, str):
|
|
try:
|
|
v = json.loads(v)
|
|
except Exception:
|
|
v = default
|
|
return unreal.Rotator(float(v[0]), float(v[1]), float(v[2]))
|
|
|
|
|
|
OPS = {
|
|
"api": api,
|
|
"create_graph": create_graph,
|
|
"list_node_types": list_node_types,
|
|
"read_graph": read_graph,
|
|
"add_node": add_node,
|
|
"delete_node": delete_node,
|
|
"connect": connect,
|
|
"disconnect": disconnect,
|
|
"list_node_properties": list_node_properties,
|
|
"set_node_property": set_node_property,
|
|
"move_nodes": move_nodes,
|
|
"save": save,
|
|
"open_asset": open_asset,
|
|
# declarative authoring + layout
|
|
"apply_graph": apply_graph,
|
|
"auto_layout": auto_layout,
|
|
"tidy_graph": auto_layout,
|
|
"layout": auto_layout,
|
|
# parameters + subgraph
|
|
"list_params": list_params,
|
|
"set_subgraph": set_subgraph,
|
|
# level / runtime
|
|
"spawn_volume": spawn_volume,
|
|
"set_actor_graph": set_actor_graph,
|
|
"generate": generate,
|
|
"cleanup": cleanup,
|
|
"list_components": list_components,
|
|
# presets
|
|
"scaffold": scaffold,
|
|
}
|
|
|
|
|
|
def entrypoint(payload_json: str) -> str:
|
|
try:
|
|
payload = json.loads(payload_json)
|
|
except Exception as exc: # noqa: BLE001
|
|
return _tee({"ok": False, "error": f"bad json: {exc}"}, {})
|
|
|
|
fn = OPS.get(payload.get("op"))
|
|
if not fn:
|
|
return _tee({"ok": False, "error": f"unknown op {payload.get('op')!r}",
|
|
"available": list(OPS)}, payload)
|
|
try:
|
|
return _tee(fn(payload), payload)
|
|
except Exception as exc: # noqa: BLE001
|
|
return _tee({"ok": False, "error": str(exc),
|
|
"traceback": traceback.format_exc()}, payload)
|
|
|
|
|
|
def _tee(result: dict, payload: dict) -> str:
|
|
text = json.dumps(result)
|
|
try:
|
|
if unreal is not None:
|
|
saved_dir = unreal.Paths.project_saved_dir() + "MCP/"
|
|
os.makedirs(saved_dir, exist_ok=True)
|
|
rid = payload.get("__rid") or "last"
|
|
with open(saved_dir + f"{rid}.json", "w", encoding="utf-8") as handle:
|
|
handle.write(text)
|
|
with open(saved_dir + "last.json", "w", encoding="utf-8") as handle:
|
|
handle.write(text)
|
|
unreal.log("[MCP-PCG] " + text[:1500])
|
|
except Exception:
|
|
pass
|
|
return text
|