"""In-memory node-tree snapshot helpers.
These functions serialise a live :class:`Node` subtree to a plain dict and
reconstruct it again. They are the engine's transient snapshot primitive --
used for clipboard copy/paste, editor autosave envelopes, play-mode
serialise/restore, and hot-reload state preservation. They are *not* a disk
serialisation format: scenes on disk are Python source via
:mod:`simvx.core.scene_io`.
Private to the engine -- consumers import via this private module path.
"""
from __future__ import annotations
import copy
import dataclasses
import os
import sys
from enum import Enum
from pathlib import Path
import numpy as np
from .audio import AudioPlayer, AudioPlayer2D, AudioPlayer3D
from .descriptors import declared_default
from .math.types import Quat, Vec2, Vec3
from .node import Node
from .nodes_2d.node2d import Node2D
from .nodes_3d.mesh import MeshInstance3D
from .nodes_3d.node3d import Node3D
# ---------------------------------------------------------------------------
# Type codec -- tagged dicts for JSON-safe encoding
# ---------------------------------------------------------------------------
_TYPE_TAGS = {
Vec2: "__vec2__",
Vec3: "__vec3__",
Quat: "__quat__",
}
_TAG_TO_TYPE = {tag: cls for cls, tag in _TYPE_TAGS.items()}
def _encode_type(v):
"""Encode a Vec2/Vec3/Quat as a tagged dict."""
if isinstance(v, Quat):
return {"__quat__": [float(v.w), float(v.x), float(v.y), float(v.z)]}
if isinstance(v, Vec3):
return {"__vec3__": [float(c) for c in v]}
if isinstance(v, Vec2):
return {"__vec2__": [float(c) for c in v]}
return v
def _decode_type(d):
"""Decode a tagged dict back to Vec2/Vec3/Quat."""
if not isinstance(d, dict):
return d
for tag, cls in _TAG_TO_TYPE.items():
if tag in d:
return cls(*d[tag])
return d
# Spatial defaults per base class for skip-if-default logic
_SPATIAL_DEFAULTS_2D = {
"position": Vec2(),
"rotation": 0.0,
"scale": Vec2(1, 1),
}
_SPATIAL_DEFAULTS_3D = {
"position": Vec3(),
"rotation": Quat(),
"scale": Vec3(1, 1, 1),
}
def _type_eq(a, b) -> bool:
"""Compare two Vec/Quat values (or floats) for approximate equality."""
if isinstance(a, int | float) and isinstance(b, int | float):
return abs(a - b) < 1e-9
if isinstance(a, Quat) and isinstance(b, Quat):
return all(abs(getattr(a, c) - getattr(b, c)) < 1e-9 for c in "wxyz")
if isinstance(a, Vec3) and isinstance(b, Vec3):
return all(abs(av - bv) < 1e-9 for av, bv in zip(a, b, strict=True))
if isinstance(a, Vec2) and isinstance(b, Vec2):
return all(abs(av - bv) < 1e-9 for av, bv in zip(a, b, strict=True))
return a == b
def _serialize_material(mat) -> dict:
"""Serialise a Material to a plain dict (including texture URIs)."""
d = {
"colour": list(mat.colour),
"metallic": mat.metallic,
"roughness": mat.roughness,
"wireframe": mat.wireframe,
"blend": mat.blend,
"double_sided": mat.double_sided,
"unlit": mat.unlit,
}
if mat.blend == "cutoff":
d["alpha_cutoff"] = mat.alpha_cutoff
if mat.uv_offset != (0.0, 0.0) or mat.uv_scale != (1.0, 1.0) or mat.uv_rotation != 0.0:
d["uv_offset"] = list(mat.uv_offset)
d["uv_scale"] = list(mat.uv_scale)
d["uv_rotation"] = mat.uv_rotation
if getattr(mat, "wetness_affected", False):
d["wetness_affected"] = True
if not getattr(mat, "receives_decals", True):
d["receives_decals"] = False
if mat.albedo_uri:
d["albedo_uri"] = mat.albedo_uri
if mat.normal_uri:
d["normal_uri"] = mat.normal_uri
if mat.metallic_roughness_uri:
d["metallic_roughness_uri"] = mat.metallic_roughness_uri
if mat.emissive_uri:
d["emissive_uri"] = mat.emissive_uri
if mat.ao_uri:
d["ao_uri"] = mat.ao_uri
return d
def _serialize_mesh_spec(spec: dict) -> dict:
"""Serialise a Mesh.factory_spec to a JSON-safe dict.
Form 1 (procedural primitives): ``{"factory": "cube", "kwargs": {...}}``
Form 2 (OBJ from filesystem): ``{"factory": "obj", "path": "..."}``
Form 3 (OBJ from package): ``{"factory": "obj", "resource": [pkg, name]}``
Any recorded mutators ride along as ``{"mutators": [...]}``.
"""
from .resource import Resource
name = spec["name"]
d: dict
if name == "obj":
source = spec["source"]
if isinstance(source, Resource):
d = {"factory": "obj", "resource": [source.package, source.name]}
else:
d = {"factory": "obj", "path": str(os.fspath(source))}
else:
d = {"factory": name, "kwargs": dict(spec.get("kwargs", {}))}
if spec.get("mutators"):
d["mutators"] = list(spec["mutators"])
return d
def _deserialize_mesh_spec(d, scene_dir: str = ""):
"""Reconstruct a Mesh from its serialised factory spec.
Backwards-tolerant: if *d* is a string we treat it as an OBJ path
(legacy snapshots from before the URI removal).
"""
# The whitelist comes from the decorator that does the recording, read here
# rather than restated, so the two halves cannot list different methods.
from .graphics.mesh import REPLAYABLE_MUTATORS, Mesh
from .resource import Resource
if isinstance(d, str):
path = d
if scene_dir and not Path(path).is_absolute():
path = str(Path(scene_dir) / path)
return Mesh.from_obj(path)
factory = d["factory"]
if factory == "obj":
if "resource" in d:
pkg, name = d["resource"]
mesh = Mesh.from_obj(Resource(pkg, name))
else:
path = d["path"]
if scene_dir and not Path(path).is_absolute():
path = str(Path(scene_dir) / path)
mesh = Mesh.from_obj(path)
else:
method = getattr(Mesh, factory, None)
if method is None:
raise ValueError(f"Unknown mesh factory: {factory!r}")
mesh = method(**d.get("kwargs", {}))
for mutator in d.get("mutators", ()):
if mutator not in REPLAYABLE_MUTATORS:
raise ValueError(f"Unknown mesh mutator: {mutator!r}")
getattr(mesh, mutator)()
return mesh
def _serialize_audio_source(stream) -> dict | str:
"""Serialise an AudioClip's source to a JSON-safe value."""
from .resource import Resource
source = stream.source
if isinstance(source, Resource):
return {"resource": [source.package, source.name]}
if isinstance(source, (str, os.PathLike)):
return os.fspath(source)
# Traversable / synthetic: fall back to the resolved path.
return stream.path
def _deserialize_audio_source(d, scene_dir: str = ""):
"""Convert serialised audio-source value back into something AudioClip accepts."""
from .resource import Resource
if isinstance(d, dict):
if "resource" in d:
pkg, name = d["resource"]
return Resource(pkg, name)
raise ValueError(f"Unknown audio source dict: {d!r}")
if isinstance(d, str):
if scene_dir and d and not Path(d).is_absolute() and "/" in d:
return str(Path(scene_dir) / d)
return d
raise ValueError(f"Unknown audio source: {d!r}")
# ---------------------------------------------------------------------------
# Snapshot independence
# ---------------------------------------------------------------------------
[docs]
def independent_copy(value):
"""A value one node can mutate without reaching into another that holds it.
A snapshot records the live Property values, so a node rebuilt from one has
to be handed copies or the two share a single object: adding an animation
to a pasted ``AnimatedSprite2D`` would add it to the sprite it was copied
from, and the symptom would appear on the node the author did not edit.
What is copied is the authoring state a node **owns**: containers, the
mutable dataclasses inside them, and the numeric arrays ``Vec2``, ``Vec3``
and ``Quat`` are. What is shared is everything a node **refers to**: a
texture, a physics shape (``Shape.build`` memoises per world, so the
sharing is load-bearing), another node, and any frozen dataclass, which
cannot drift apart in the first place.
"""
if value is None or isinstance(value, str | bytes | int | float | complex | Enum):
return value
if isinstance(value, np.ndarray):
return value.copy()
if isinstance(value, list):
return [independent_copy(item) for item in value]
if isinstance(value, dict):
return {key: independent_copy(item) for key, item in value.items()}
if isinstance(value, set):
return {independent_copy(item) for item in value}
if type(value) is tuple:
items = tuple(independent_copy(item) for item in value)
# A tuple is immutable, so it is only worth rebuilding when something
# inside it was not.
return value if all(a is b for a, b in zip(items, value, strict=True)) else items
if dataclasses.is_dataclass(value) and not isinstance(value, type):
if value.__dataclass_params__.frozen:
return value
clone = copy.copy(value)
for field in dataclasses.fields(value):
setattr(clone, field.name, independent_copy(getattr(value, field.name)))
return clone
return value
# ---------------------------------------------------------------------------
# Node tree <-> dict
# ---------------------------------------------------------------------------
def _serialize_node(node: Node) -> dict:
"""Recursively serialise a node and its children."""
if node._scene_template_path:
d = {"__scene__": node._scene_template_path, "name": node.name}
if isinstance(node, Node3D):
defaults = _SPATIAL_DEFAULTS_3D
for attr in ("position", "rotation", "scale"):
val = getattr(node, attr)
if not _type_eq(val, defaults[attr]):
d[attr] = _encode_type(val)
elif isinstance(node, Node2D):
defaults = _SPATIAL_DEFAULTS_2D
for attr in ("position", "scale"):
val = getattr(node, attr)
if not _type_eq(val, defaults[attr]):
d[attr] = _encode_type(val)
if not _type_eq(node.rotation, 0.0):
d["rotation"] = node.rotation
return d
# ``__module__`` is a hint, not a key: two project files may each declare
# ``class Player(Node2D)``, and ``Node._registry`` keeps only the last of
# them. Recording the defining module lets the rebuild pick the right one
# while a snapshot written without it still resolves the old way.
d = {"__type__": type(node).__name__, "__module__": type(node).__module__, "name": node.name}
if isinstance(node, Node3D):
defaults = _SPATIAL_DEFAULTS_3D
for attr in ("position", "rotation", "scale"):
val = getattr(node, attr)
if not _type_eq(val, defaults[attr]):
d[attr] = _encode_type(val)
elif isinstance(node, Node2D):
defaults = _SPATIAL_DEFAULTS_2D
for attr in ("position", "scale"):
val = getattr(node, attr)
if not _type_eq(val, defaults[attr]):
d[attr] = _encode_type(val)
if not _type_eq(node.rotation, 0.0):
d["rotation"] = node.rotation
settings = {}
# Spatial fields (position/rotation/scale) are emitted via the dedicated
# branch above with their own default-detection: skip them here to avoid
# double-emission once promoted to Property descriptors.
_SKIP_SPATIAL = ("position", "rotation", "scale")
spatial_node = isinstance(node, (Node2D, Node3D))
for name, prop in node.get_properties().items():
if spatial_node and name in _SKIP_SPATIAL:
continue
val = getattr(node, name)
if not _is_default(val, declared_default(prop)):
settings[name] = val
if settings:
d["settings"] = settings
if node._groups:
d["groups"] = sorted(node._groups)
if isinstance(node, MeshInstance3D) and node.mesh is not None:
if getattr(node.mesh, "factory_spec", None):
d["mesh"] = _serialize_mesh_spec(node.mesh.factory_spec)
if isinstance(node, MeshInstance3D) and node.material is not None:
d["material"] = _serialize_material(node.material)
if isinstance(node, AudioPlayer | AudioPlayer2D | AudioPlayer3D):
if node.stream is not None:
d["stream"] = _serialize_audio_source(node.stream)
children = [_serialize_node(c) for c in node.children]
if children:
d["children"] = children
return d
def _resolve_node_type(cls_name: str, module_name: str | None) -> type[Node] | None:
"""Find the node class a snapshot names, preferring its own defining module.
``Node._registry`` is keyed on the bare class name, so a project holding
``a/player.py`` and ``b/player.py`` keeps only whichever ``Player`` was
declared last. The snapshot's ``__module__`` hint resolves that when the
module is still loaded, which is the case for every same-process rebuild
(clipboard, node duplication, entering and leaving play mode).
Falls back to the registry when the hint is absent (older snapshots) or its
module is no longer importable, which is no worse than the previous
last-definition-wins behaviour.
"""
if module_name:
module = sys.modules.get(module_name)
candidate = getattr(module, cls_name, None) if module is not None else None
if isinstance(candidate, type) and issubclass(candidate, Node):
return candidate
return Node._registry.get(cls_name)
def _is_default(value, default) -> bool:
"""Whether ``value`` reads as ``default``, tolerating array-valued properties."""
try:
eq = value == default
return bool(eq) if not isinstance(eq, np.ndarray) else bool(eq.all())
except Exception:
return False
def _snapshot_kwargs(d: dict, cls_name: str) -> dict:
"""The constructor arguments a snapshot dict describes."""
kwargs = {"name": d.get("name", cls_name)}
for attr in ("position", "rotation", "scale"):
if attr in d:
kwargs[attr] = _decode_type(d[attr])
# Copied, never handed straight over: the snapshot holds the live objects,
# and a rebuild that aliased them would leave two nodes sharing one value.
for key, val in d.get("settings", {}).items():
kwargs[key] = independent_copy(val)
return kwargs
def _restate_from_snapshot(node: Node, d: dict) -> None:
"""Rewrite an existing node's own fields to match a snapshot.
The counterpart of passing :func:`_snapshot_kwargs` to a constructor, for a
node that is already built. What the snapshot omits was reading as its
declared default when it was written, so it is put back to that default
rather than left holding whatever built the node in the first place.
"""
node.name = d.get("name", type(node).__name__)
if isinstance(node, Node3D):
spatial = _SPATIAL_DEFAULTS_3D
elif isinstance(node, Node2D):
spatial = _SPATIAL_DEFAULTS_2D
else:
spatial = {}
for attr, default in spatial.items():
setattr(node, attr, _decode_type(d[attr]) if attr in d else independent_copy(default))
settings = d.get("settings", {})
for key, val in settings.items():
setattr(node, key, independent_copy(val))
for name, prop in node.get_properties().items():
if name in settings or name in spatial:
continue
default = declared_default(prop)
if not _is_default(getattr(node, name), default):
setattr(node, name, independent_copy(default))
def _reconcile_children(node: Node, children_data, scene_dir: str) -> None:
"""Give ``node`` exactly the children the snapshot lists.
A custom node's ``__init__`` builds part of its own structure, and
:func:`_serialize_node` walks those children like any others, so the
snapshot already describes them. Adding them on top is what made a round
trip grow the tree by one copy of each per cycle.
The snapshot decides which children exist and what state they hold. Where it
names one the constructor also built, that instance is kept rather than
rebuilt: the constructor's own references to it and any signals it connected
are what the node's code then drives, and a fresh object would leave both
pointing at something no longer in the tree.
"""
unclaimed = list(node.children)
ordered: list[Node] = []
for child_data in children_data:
match = None
if "__scene__" not in child_data:
wanted = _resolve_node_type(child_data["__type__"], child_data.get("__module__"))
wanted_name = child_data.get("name")
for candidate in unclaimed:
if type(candidate) is wanted and candidate.name == wanted_name:
match = candidate
break
if match is None:
ordered.append(_deserialize_node(child_data, scene_dir=scene_dir))
else:
unclaimed.remove(match)
_restate_from_snapshot(match, child_data)
_apply_snapshot_payload(match, child_data, scene_dir)
ordered.append(match)
for orphan in unclaimed:
# Never entered a tree, so this is the pure-bookkeeping unlink: running
# the exit path would fire on_exit_tree on a node that never entered.
node._detach_child(orphan)
for child in ordered:
if child.parent is not node:
node.add_child(child)
node.children.move_last(child)
def _apply_snapshot_payload(node: Node, d: dict, scene_dir: str) -> None:
"""Apply the parts of a snapshot that are not constructor arguments."""
for group in d.get("groups", ()):
node.add_to_group(group)
if "material" in d and isinstance(node, MeshInstance3D):
from .graphics.material import Material
mat_data = d["material"].copy()
if "albedo_uri" in mat_data:
mat_data["albedo_map"] = mat_data.pop("albedo_uri")
if "normal_uri" in mat_data:
mat_data["normal_map"] = mat_data.pop("normal_uri")
if "metallic_roughness_uri" in mat_data:
mat_data["metallic_roughness_map"] = mat_data.pop("metallic_roughness_uri")
if "emissive_uri" in mat_data:
mat_data["emissive_map"] = mat_data.pop("emissive_uri")
if "ao_uri" in mat_data:
mat_data["ao_map"] = mat_data.pop("ao_uri")
node.material = Material(**mat_data)
if "mesh" in d and isinstance(node, MeshInstance3D):
node.mesh = _deserialize_mesh_spec(d["mesh"], scene_dir=scene_dir)
if "stream" in d and isinstance(node, AudioPlayer | AudioPlayer2D | AudioPlayer3D):
from .audio import AudioClip
node.stream = AudioClip(_deserialize_audio_source(d["stream"], scene_dir=scene_dir))
# Absent rather than empty: a snapshot that lists no children at all may be
# hand-built or predate child recording, so the constructor's own structure
# is left alone. `_serialize_node` omits the key only for a childless node.
if "children" in d:
_reconcile_children(node, d["children"], scene_dir)
def _deserialize_node(d: dict, scene_dir: str = "") -> Node:
"""Recursively reconstruct a node from a snapshot dict."""
if "__scene__" in d:
# Sub-scene reference: load the linked scene file via scene_io.
from .scene_io.loader import load_scene
scene_path = Path(scene_dir) / d["__scene__"] if scene_dir else Path(d["__scene__"])
sub = load_scene(scene_path)
sub._scene_template_path = str(scene_path)
if "name" in d:
sub.name = d["name"]
for attr in ("position", "rotation", "scale"):
if attr in d:
setattr(sub, attr, _decode_type(d[attr]))
return sub
cls_name = d["__type__"]
cls = _resolve_node_type(cls_name, d.get("__module__"))
if cls is None:
raise ValueError(f"Unknown node type: {cls_name!r}")
node = cls(**_snapshot_kwargs(d, cls_name))
_apply_snapshot_payload(node, d, scene_dir)
# No collider auto-link needed: the new physics bodies (PhysicsBody*,
# CharacterBody*) discover their geometry from the first direct-child
# CollisionShape at enter-tree, so a deserialised body's shape resolves
# itself once it joins a live tree (no `.collision` attribute to set).
return node