# File: _decoder_.py
# Code: Claude Code and Codex
# Review: Ryoichi Ando (ryoichi.ando@zozo.com)
# License: Apache v2.0
import os
import pickle
from dataclasses import dataclass
from typing import Any, Optional
from . import _rust # type: ignore[attr-defined]
from ._asset_ import AssetManager
from ._mesh_ import CachePathUnusableError, MeshManager, _cache_probe
from ._plot_ import PlotManager
from ._scene_ import FixedScene, Scene
from ._session_ import FixedSession, Session
from ._utils_ import Utils
# Per-vertex pull / barycentric weight prune cutoff: a tet vertex is "kept"
# (pinned) only when its mapped weight exceeds this. Shared by the SOLID
# pin-mapping and intent-classification paths so the cutoff lives in one place.
_PIN_WEIGHT_EPS = 1e-4
# Partial-pin SOLID hard/soft split default: a kept surface vertex is hard
# (FixPair) when its full weight is at or above this, soft (PullPair) below.
# Used as the cfg "fix_weight_threshold" fallback.
_DEFAULT_FIX_WEIGHT_THRESHOLD = 0.5
_SCIPY_MISSING_WARNED = False
def _warn_if_scipy_missing(context):
"""Emit a loud one-time warning when SciPy is absent.
The partial-pin SOLID diffusion (``_build_solid_pin_fields`` /
``_build_harmonic_interior_operator``) imports SciPy inside a
``try/except`` that returns ``None`` on any failure, so a runtime
without SciPy does not crash: it silently takes a different
surface-only fallback pin path. That yields a DIFFERENT driven-vertex
set than a SciPy-equipped runtime (observed: a Windows bundle missing
SciPy diverged from the identical Linux scene). Distinguish the
packaging problem (SciPy genuinely absent) from a legitimate solve
failure so the former is visible instead of silently wrong.
"""
global _SCIPY_MISSING_WARNED
if _SCIPY_MISSING_WARNED:
return
import importlib.util
if importlib.util.find_spec("scipy") is not None:
return # SciPy present; the None came from a real solve failure.
_SCIPY_MISSING_WARNED = True
import sys
print(
"WARNING: SciPy is not installed. Partially-pinned SOLID objects fall "
"back to a surface-only pin path that differs from the SciPy-based "
"two-stage Poisson diffusion, producing a different (and "
"platform-inconsistent) driven-vertex set for the same scene. Install "
f"scipy so every platform uses the same pin path (context: {context}).",
file=sys.stderr, flush=True,
)
@dataclass
class ObjectInfo:
"""Per-object metadata recorded by :class:`SceneDecoder.populate_objects`.
Field shape varies by ``type``:
* ``SOLID`` populates ``vert``, ``V``, ``F``, and optionally ``orig_to_sim``.
* ``SHELL`` populates ``vert``, ``V`` (== ``vert``), ``F``.
* ``ROD`` populates ``vert`` only; ``V`` and ``F`` stay ``None``.
* ``SAND`` populates ``vert`` only (a faceless point cloud); ``V``
and ``F`` stay ``None``.
The Rust ``cross_stitch_apply_batch`` consumer reads via dict access,
so :meth:`to_dict` is used at the FFI boundary.
"""
type: str
vert: Any
V: Optional[Any] = None
F: Optional[Any] = None
orig_to_sim: Optional[Any] = None
def to_dict(self) -> dict:
d: dict = {"type": self.type, "vert": self.vert}
if self.V is not None:
d["V"] = self.V
if self.F is not None:
d["F"] = self.F
if self.orig_to_sim is not None:
d["orig_to_sim"] = self.orig_to_sim
return d
[docs]
class BlenderApp:
"""Attach to a project exported by the Blender addon and build a runnable session.
The addon writes ``data.pickle`` and ``param.pickle`` into a
per-project directory under ``~/.local/share/ppf-cts/git-<branch>/<name>/``.
:meth:`open` is the canonical entry point: it decodes those pickles,
populates the scene, applies parameters, and builds a
``FixedSession`` accessible via :attr:`scene` and :attr:`session`.
Example:
Canonical notebook cell (as generated by the Blender addon),
attach to a transferred project and run it::
from frontend import BlenderApp
app = BlenderApp.open("rod-example-3")
app.scene.report()
app.scene.preview()
app.session.run()
app.session.preview()
"""
def __init__(self, name: str, verbose: bool = False, progress_callback=None):
"""Initialize the BlenderApp.
Args:
name (str): The name of the Blender project.
verbose (bool): Enable verbose logging.
progress_callback: Optional callable ``fn(progress: float, info: str)`` invoked during long-running operations. ``progress`` is in ``[0.0, 1.0]``.
"""
self._verbose = verbose
self._name = name
self._progress_callback = progress_callback
# Path math + branch resolution lives in Rust
# (`dec::blender_app_paths`), so the Python and Windows-native
# branches stay aligned with `App.get_data_dirpath` and the
# server's `make_root`.
paths = _rust.blender_app_paths(os.path.abspath(__file__), name)
self._data_dirpath = paths["data_dirpath"]
self._root = paths["root"]
cache_root = paths["cache_root"]
os.makedirs(cache_root, exist_ok=True)
self._asset_manager = AssetManager()
self._mesh_manager = MeshManager(cache_root)
self._scene = None
self._session = None
self._fixed_scene = None
self._fixed_session = None
self._param_decoder = None
def __getstate__(self):
state = self.__dict__.copy()
state["_progress_callback"] = None
return state
def _report_progress(self, progress: float, info: str):
if self._progress_callback is not None:
self._progress_callback(progress, info)
[docs]
@classmethod
def open(
cls,
name: str,
verbose: bool = False,
progress_callback=None,
) -> "BlenderApp":
"""Open a Blender project and build its scene and session.
Behavior:
1. If the Blender addon has uploaded both ``data.pickle`` and
``param.pickle`` under the project root, ``populate().make()``
is invoked to build a fresh scene and session in this process.
2. Otherwise, ``FileNotFoundError`` is raised. The user must
click "Transfer" in the Blender addon first.
After ``open()``, ``app.scene`` and ``app.session`` are the built
``FixedScene`` / ``FixedSession`` objects, ready for report,
preview, run, and stream.
Example:
Notebook cell generated by the Blender addon, attaching to
the transferred project and running it::
from frontend import BlenderApp
app = BlenderApp.open("<project>")
app.scene.preview()
app.session.run()
app.session.preview()
"""
app = cls(name, verbose=verbose, progress_callback=progress_callback)
data_path = os.path.join(app._root, "data.pickle")
param_path = os.path.join(app._root, "param.pickle")
if not (os.path.exists(data_path) and os.path.exists(param_path)):
raise FileNotFoundError(
f"Blender has not uploaded project '{name}' to {app._root}. "
f"In the Blender addon, click 'Transfer' first."
)
# Invariant: whenever data.pickle and param.pickle both exist,
# upload_id.txt must also exist. It's stamped by the Blender
# addon's upload handler at write time and is the identity that
# client build-tracking pins against. If it's missing, the
# project predates upload_id tracking (or someone wrote pickles
# out-of-band); refuse to build and tell the user how to migrate.
upload_id_path = os.path.join(app._root, "upload_id.txt")
if not os.path.exists(upload_id_path):
raise FileNotFoundError(
f"Project '{name}' at {app._root} has data.pickle and "
f"param.pickle but no upload_id.txt. This project predates "
f"upload_id tracking. Re-upload from the Blender addon, "
f"or run `python tools/backfill_upload_ids.py` to stamp a "
f"fresh id on every legacy project."
)
app.populate().make()
return app
[docs]
def populate(self) -> "BlenderApp":
"""Populate the scene with objects decoded from ``data.pickle``.
Also peeks at ``param.pickle`` for per-group fTetWild overrides so
that tetrahedralization can use them at build time; full
parameter application is deferred to :meth:`make`.
Example:
Two-stage construction (useful when you want to inspect the
mutable scene before parameters are applied)::
from frontend import BlenderApp
app = BlenderApp("my-project")
app.populate()
app.make()
app.session.run()
"""
data_path = os.path.join(self._root, "data.pickle")
assert os.path.exists(data_path)
self._report_progress(0.05, "Loading scene data...")
self._asset_manager = AssetManager()
self._scene = Scene("scene", PlotManager(), self._asset_manager)
self._scene_decoder = SceneDecoder(
data_path,
self._asset_manager,
self._mesh_manager,
)
# Decode param.pickle once and stash the decoder so make() can
# reuse it rather than re-reading and re-parsing the same file.
# We peek at the parsed data here for per-object tetrahedralizer
# overrides so populate_objects can pass them into
# tetrahedralize() at build time; full parameter application is
# deferred to make(). The encoder keys the per-group ``ftetwild``
# entry by object UUID (mirroring ``velocity``), so each object
# can pick its own backend (fTetWild or TetGen) and per-field
# overrides.
ftetwild_by_uuid: dict = {}
soft_constraint_by_uuid: dict = {}
solver_fps = None
time_scale = None
param_path = os.path.join(self._root, "param.pickle")
if os.path.exists(param_path):
self._param_decoder = ParamDecoder().set_path(param_path)
# The DATA payload is timing-free (frame offsets / raw animation
# rates); the time base lives HERE. Both keys are required in a
# v2 param payload; fail loud rather than defaulting.
_scene_params = self._param_decoder._data.get("scene")
if not isinstance(_scene_params, dict) or "fps" not in _scene_params:
raise RuntimeError(
"param.pickle is missing scene.fps; "
"re-Transfer from the Blender addon"
)
solver_fps = float(_scene_params["fps"])
if "time_scale" not in self._param_decoder._data:
raise RuntimeError(
"param.pickle is missing time_scale; "
"re-Transfer from the Blender addon"
)
time_scale = float(self._param_decoder._data["time_scale"])
for _entry in self._param_decoder._data.get("group", []):
if not _entry:
continue
_params = _entry[0]
_ftw = _params.get("ftetwild") if isinstance(_params, dict) else None
if isinstance(_ftw, dict):
for _uuid, _kw in _ftw.items():
if isinstance(_kw, dict) and _kw:
ftetwild_by_uuid[_uuid] = _kw
# A STATIC group's pin spring stiffness. Peeked here for the
# same reason as ftetwild: it is consumed while the object's
# pins are being built, which is earlier than apply_to_object.
_soft = (
_params.get("soft-constraint")
if isinstance(_params, dict) else None
)
if isinstance(_soft, dict):
for _uuid, _k in _soft.items():
soft_constraint_by_uuid[_uuid] = float(_k)
# A STATIC collider that participates in a cross-stitch must be
# reachable by the stitch index space, which addresses only the
# dynamic vertex namespace (map_by_name -> concat_vert -> eval_x). A
# non-moving STATIC is otherwise a disjoint contact-only collision
# mesh, so we pre-scan the cross-stitch endpoints here and promote any
# such STATIC into the dynamic all-pinned namespace at populate time
# (it stays kinematically frozen via immovable fixed pins). The
# cross_stitch list is loaded eagerly by ParamDecoder.set_path.
#
# This pre-scan reads the RAW param entries; promotion must happen at
# populate time (build partitions dyn vs static from the finalized pin
# structure), before cross_stitch_apply_batch runs in make(). So a
# STATIC named by an entry that apply_batch later drops (e.g. a SOLID
# source lacking source_points) is still promoted: a benign frozen
# dynamic collider with no surviving stitch, not an error.
stitch_endpoint_uuids: set[str] = set()
if self._param_decoder is not None:
for _cs in self._param_decoder.cross_stitch:
if not isinstance(_cs, dict):
continue
for _key in ("source_uuid", "target_uuid"):
_u = _cs.get(_key)
if _u:
stitch_endpoint_uuids.add(_u)
self._scene_decoder.populate_objects(
self._scene,
verbose=self._verbose,
progress_callback=lambda progress, info: self._report_progress(
0.10 + 0.50 * progress,
info,
),
ftetwild_by_uuid=ftetwild_by_uuid,
soft_constraint_by_uuid=soft_constraint_by_uuid,
stitch_endpoint_uuids=stitch_endpoint_uuids,
solver_fps=solver_fps,
time_scale=time_scale,
)
return self
[docs]
def make(self, preserve_output: bool = False) -> "BlenderApp":
"""Apply ``param.pickle`` to the populated scene and build a runnable session.
Applies per-object parameters, pin configuration, cross-stitch
constraints, and invisible colliders to the scene, then builds the
fixed scene and session. The resulting state is persisted to
``app_state.pickle`` on a best-effort basis.
Args:
preserve_output (bool): When True, keep the solver ``output/``
subtree (saved checkpoints) while re-exporting the scene
input, so a resume can re-decode edited animation without
wiping the states.
Example:
Finish the build after :meth:`populate`, then run the session::
from frontend import BlenderApp
app = BlenderApp("my-project").populate().make()
app.session.run()
app.session.preview()
"""
assert self._scene is not None, "Scene must be populated before making the app"
param_path = os.path.join(self._root, "param.pickle")
assert os.path.exists(param_path)
self._report_progress(0.65, "Applying object parameters...")
# Reuse the decoder populate() already parsed; construct lazily
# for direct make() callers that bypassed populate().
param_decoder = self._param_decoder
if param_decoder is None:
param_decoder = ParamDecoder().set_path(param_path)
param_decoder.apply_to_objects(
self._scene,
verbose=self._verbose,
solid_weight_transfers=(
self._scene_decoder.solid_weight_transfer
if self._scene_decoder is not None
else None
),
)
# The transfers hold a SuperLU factor and a sparse system, neither of
# which belongs in the app-state pickle written at the end of make().
if self._scene_decoder is not None:
self._scene_decoder.release_solid_weight_transfers()
self._report_progress(0.72, "Applying pin configuration...")
param_decoder.apply_pin_config(self._scene, verbose=self._verbose)
if param_decoder.cross_stitch:
self._report_progress(0.78, "Applying cross-stitch constraints...")
# Whole-batch port: Rust validates endpoints, re-projects the
# anchors of SOLID source and/or target sides independently onto
# their tet surfaces, builds each canonical dict, and appends
# directly to ``self._scene._cross_stitch`` so no Python-side
# per-entry append loop survives.
obj_info_dict = {
k: v.to_dict() for k, v in self._scene_decoder._object_info.items()
}
_rust.cross_stitch_apply_batch(
param_decoder.cross_stitch,
obj_info_dict,
self._scene._cross_stitch,
self._verbose,
)
self._report_progress(0.82, "Applying invisible colliders...")
param_decoder.apply_invisible_colliders(self._scene, verbose=self._verbose)
self._report_progress(0.84, "Building scene: preparing objects...")
self._fixed_scene = self._scene.build(
progress_callback=lambda progress, info: self._report_progress(
0.84 + 0.10 * progress,
info,
)
)
self._report_progress(0.94, "Initializing session...")
self._session = Session(
self._name,
self._root,
os.path.dirname(os.path.dirname(__file__)),
self._data_dirpath,
"session",
).init(self._fixed_scene)
self._report_progress(0.95, "Applying session parameters...")
param_decoder.apply_to_session(self._session, verbose=self._verbose)
# apply_to_session above just sets parameter values and is cheap; the
# heavy tail is session.build() (exports the full solver input to disk
# and pickles the session graph) followed by _persist_app_state()
# (pickles the whole app). Label each distinctly so the bar moves
# through them instead of sitting parked at "Applying session
# parameters..." for the entire export/serialize phase.
self._report_progress(0.96, "Exporting solver input...")
self._fixed_session = self._session.build(preserve_output=preserve_output)
self._report_progress(0.99, "Saving build state...")
self._persist_app_state()
self._report_progress(1.0, "Build decode complete.")
return self
def _persist_app_state(self) -> None:
"""Write this BlenderApp to ``{_root}/app_state.pickle`` so that
both the Blender-addon server and future notebook runs can
discover the built state. The Blender-addon server reads it back
through ``server/engine.py:AppState.load`` when it needs to adopt
an externally-built session.
Raises any persistence error (disk full, permission denied,
pickling failure) rather than silently swallowing it: a failed
persist means the server-side state machine won't pick up this
build, and the caller should know.
Format note: this site uses raw ``pickle`` rather than a CBOR
envelope. The payload is the entire ``BlenderApp`` object
(asset / scene / mesh / fixed_session graph), no manager class
on that graph carries a hand-written CBOR schema, and the
consumer lives outside this repo (the Blender-addon server's
``AppState.load``). The producer-side wrappers in
:mod:`frontend._session_` and :mod:`frontend._app_` use CBOR
envelopes; this is the only intentional raw-pickle write in
:mod:`frontend.`
"""
# _process holds a subprocess Popen with non-picklable locks.
if self._fixed_session is not None:
self._fixed_session._process = None
paths = _rust.app_state_persist_paths(self._root)
os.makedirs(self._root, exist_ok=True)
with open(paths["tmp_path"], "wb") as f:
pickle.dump(self, f)
os.replace(paths["tmp_path"], paths["final_path"])
@property
def scene(self) -> FixedScene:
"""The built fixed scene. Use this for ``.report()``,
``.preview()``, etc. The mutable pre-build ``Scene`` remains
accessible as ``_scene`` for advanced callers.
Example:
Open a Blender-authored project and inspect the built scene::
from frontend import BlenderApp
app = BlenderApp.open("rod-example-3")
app.scene.report()
app.scene.preview()
"""
assert self._fixed_scene is not None, (
"Scene is not built yet. Call BlenderApp.open() or "
".populate().make() first."
)
return self._fixed_scene
@property
def session(self) -> FixedSession:
"""The built fixed session. Use this for ``.run()``,
``.preview()``, ``.stream()``, etc.
Example:
Run a Blender-authored session and stream live output::
from frontend import BlenderApp
app = BlenderApp.open("rod-example-3")
app.session.run()
app.session.preview()
app.session.stream()
"""
assert self._fixed_session is not None, (
"Session is not built yet. Call BlenderApp.open() or "
".populate().make() first."
)
return self._fixed_session
class ParamDecoder:
"""Load and apply ``param.pickle`` written by the Blender addon.
Parameters are split across object-level (velocity, fTetWild hints,
per-key ``param.set`` values), pin configuration, cross-stitch
constraints, explicit merge pairs, invisible walls / spheres, and
session-level scene settings. :meth:`set_path` loads the pickle,
then the ``apply_*`` methods dispatch each section.
Example:
Apply a pickle to an existing scene and session (this is what
:meth:`BlenderApp.make` does internally)::
from frontend._decoder_ import ParamDecoder
decoder = ParamDecoder().set_path("/path/to/param.pickle")
decoder.apply_to_objects(scene)
decoder.apply_pin_config(scene)
decoder.apply_invisible_colliders(scene)
fixed_scene = scene.build()
decoder.apply_to_session(session)
"""
def __init__(self):
self._data = None
def set_path(self, filepath: str) -> "ParamDecoder":
"""Load parameter data from a pickle file and cache it on this decoder.
Args:
filepath (str): Path to the pickle file. Must end in ``.pickle``.
Returns:
ParamDecoder: ``self`` for chaining.
Example:
Chain the load with a subsequent apply call::
from frontend._decoder_ import ParamDecoder
decoder = ParamDecoder().set_path("/path/to/param.pickle")
decoder.apply_to_objects(scene)
"""
_rust.validate_pickle_extension(filepath)
from . import _cbor_bridge_ as _cbor
self._data = _cbor.load_param_file(filepath)
_rust.validate_param_top_keys(
"group" in self._data, "scene" in self._data
)
self._pin_config = self._data.get("pin_config", {})
self._cross_stitch = self._data.get("cross_stitch", [])
return self
@property
def cross_stitch(self) -> list:
return self._cross_stitch
def apply_to_objects(
self,
scene: Scene,
verbose: bool = False,
solid_weight_transfers=None,
):
"""Apply the loaded parameter data to the objects in ``scene``.
Call :meth:`set_path` first. Per-object dicts (velocity,
velocity-schedule, collision-windows) are keyed by UUID; other
keys are forwarded to ``obj.param.set``. fTetWild overrides are
consumed at populate-time and skipped here.
Args:
scene (Scene): The scene to which the parameters will be applied.
verbose (bool): Enable verbose logging.
solid_weight_transfers: Optional callable ``uuid -> transfer``
carrying a SOLID object's painted per-Blender-vertex map
weights onto its tetrahedra. Required for any tet object that
carries a spatial material map, whose weights are authored
against the Blender mesh and not the tetrahedralized one.
Example:
Load a pickle and apply per-object parameters to a populated scene::
from frontend._decoder_ import ParamDecoder
decoder = ParamDecoder().set_path("/path/to/param.pickle")
decoder.apply_to_objects(scene, verbose=True)
"""
assert self._data is not None, "Parameter data not set. Call set_path() first."
if verbose:
print("=== Object Parameters ===")
# Keyframe times for every animated material in the scene. Set first so
# the scene carries them before any object's values arrive; the two are
# checked against each other at build, where the per-triangle arrays
# are assembled.
anim_times = self._data.get("param_anim_times")
if anim_times:
scene.set_param_anim_times(list(anim_times))
for group_entry in self._data["group"]:
params, objects = group_entry[0], group_entry[1]
# Third tuple slot holds UUIDs aligned with ``objects``.
# Per-object dicts (velocity / velocity-schedule / collision-
# windows) are keyed on UUID by the encoder.
_rust.validate_param_group_has_uuids(len(group_entry))
obj_uuids = group_entry[2]
for obj_name, obj_uuid in zip(objects, obj_uuids):
_rust.validate_param_object_uuid(obj_name, obj_uuid)
if verbose:
print(f"*** name: {obj_name} (uuid={obj_uuid}) ***")
obj = scene.select(obj_uuid)
obj.param.clear_all()
for key, val in params.items():
if verbose:
print(f" {key}: {val}")
if key == "velocity":
if isinstance(val, dict):
v = val.get(obj_uuid)
if v is not None:
obj.velocity(*v)
else:
obj.velocity(*val)
elif key == "velocity-schedule":
if isinstance(val, dict):
s = val.get(obj_uuid)
if s:
obj.velocity_schedule(s)
else:
obj.velocity_schedule(val)
elif key == "angular-velocity-schedule":
# Per-UUID list of (t, pca_index, speed_rad) spin
# keyframes; the solver resolves the world axis from
# the live geometry. Same dict-vs-value shape as
# velocity-schedule.
if isinstance(val, dict):
s = val.get(obj_uuid)
if s:
obj.angular_velocity_schedule_pca(s)
else:
obj.angular_velocity_schedule_pca(val)
elif key == "angular-velocity-world-schedule":
# Per-UUID list of (t, [wx, wy, wz]) fixed world-axis
# spins (World X/Y/Z / Custom), already swapped to
# solver space and scaled by speed.
if isinstance(val, dict):
s = val.get(obj_uuid)
if s:
obj.angular_velocity_schedule_world(s)
else:
obj.angular_velocity_schedule_world(val)
elif key == "collision-windows":
if isinstance(val, dict):
w = val.get(obj_uuid)
if w:
obj.collision_windows(w)
elif key == "hinge":
# PDRD hinge: pin the body and lock its rotation to a
# principal axis (see Object.hinge). Per-UUID dict like
# velocity; absent / empty means the body stays free.
if isinstance(val, dict):
h = val.get(obj_uuid)
if h is not None:
obj.hinge(int(h))
elif val is not None:
obj.hinge(int(val))
elif key == "lock-translation":
# Lock Translation: normalized world-space axis
# (already swapped to solver space) restricting this
# object's COM to the line through its initial
# position (see Object.lock_translation). Per-UUID
# dict like velocity/hinge; absent means unlocked.
if isinstance(val, dict):
a = val.get(obj_uuid)
if a is not None:
obj.lock_translation(*a)
elif val is not None:
obj.lock_translation(*val)
elif key == "lock-all-translations":
# Lock All Translations: per-UUID bool pinning this
# object's COM to its initial point rather than to a
# line (see Object.lock_all_translations). The FLAG
# carries the enable bit here, since an all-axes lock
# has no direction: the encoder leaves such an object
# out of "lock-translation" entirely, so a UUID with
# no paired axis is the expected shape rather than an
# error. Carrying both IS an error: they are two
# mutually exclusive spellings of one lock, and an
# encoder emitting both has a bug that must not be
# resolved silently in either direction. Resolve the
# paired axis explicitly rather than relying on map
# iteration order.
m = val.get(obj_uuid) if isinstance(val, dict) else val
if m:
axes = params.get("lock-translation")
a = (
axes.get(obj_uuid)
if isinstance(axes, dict)
else axes
)
if a is not None:
raise ValueError(
f"lock-all-translations for {obj_uuid!r} also "
"carries a lock-translation axis; an all-axes "
"lock has no axis"
)
obj.lock_all_translations()
elif key == "lock-rotation":
# Lock Rotation: normalized world-space axis (already
# swapped to solver space) restricting this object's
# best-fit rigid rotation to rotation about that
# axis only (see Object.lock_rotation). Independent
# of "lock-translation" above. Per-UUID dict like
# velocity/hinge; absent means unlocked.
if isinstance(val, dict):
a = val.get(obj_uuid)
if a is not None:
obj.lock_rotation(*a)
elif val is not None:
obj.lock_rotation(*val)
elif key == "lock-rotation-prohibit-axis":
# Lock Rotation mode: True flips the axis set by
# "lock-rotation" above from a whitelist (rotation
# about it is the only freedom) to a blacklist
# (rotation about it is forbidden, the perpendicular
# plane stays free instead). Only present for UUIDs
# that also appear in "lock-rotation". Resolve the
# paired axis explicitly rather than relying on map
# iteration order.
if isinstance(val, dict):
m = val.get(obj_uuid)
if m is not None:
if getattr(obj, "_rotation_lock", None) is None:
axes = params.get("lock-rotation", {})
a = axes.get(obj_uuid) if isinstance(axes, dict) else axes
if a is None:
raise ValueError(
f"lock-rotation-prohibit-axis for {obj_uuid!r} "
"has no matching lock-rotation axis"
)
obj.lock_rotation(*a)
obj.lock_rotation_prohibit_axis(bool(m))
elif val is not None:
obj.lock_rotation_prohibit_axis(bool(val))
elif key == "lock-all-rotations":
# Lock All Rotations: per-UUID bool forbidding net
# rotation about every axis (see
# Object.lock_all_rotations). Read exactly like
# "lock-all-translations" above, including why a
# missing paired axis is expected and why carrying
# both is refused. An all-locked object appears in
# neither "lock-rotation" nor
# "lock-rotation-prohibit-axis": the mode above
# selects between two readings of an axis, and there
# is no axis here for it to read.
m = val.get(obj_uuid) if isinstance(val, dict) else val
if m:
axes = params.get("lock-rotation")
a = (
axes.get(obj_uuid)
if isinstance(axes, dict)
else axes
)
if a is not None:
raise ValueError(
f"lock-all-rotations for {obj_uuid!r} also "
"carries a lock-rotation axis; an all-axes "
"lock has no axis"
)
obj.lock_all_rotations()
elif key in ("ftetwild", "soft-constraint"):
# Consumed at populate-time via the param.pickle peek;
# no per-object ParamHolder slot by design (would
# break concat_tet_param / concat_tri_param key-set
# equality in _scene_.extend_param). "soft-constraint"
# additionally has no solver param to set: it selects
# the KIND of pin the collider gets, which is decided
# while the pins are built.
pass
elif key == "material-maps":
# Spatial maps: weights are per vertex and keyed by
# object uuid, so only this object's array is applied.
# The base is the object's own param (or that frame's
# animated value), so a mapped key may also be animated.
self._apply_material_maps(
obj, obj_uuid, val, solid_weight_transfers, verbose
)
elif key == "param-anim":
# Sampled F-curves on this group's material sliders,
# one value per entry in the scene-wide
# `param_anim_times`. Set on every object in the group,
# matching how the group's static params are applied.
for anim_key, values in val.items():
obj.set_param_anim(anim_key, values)
else:
obj.param.set(key, val)
@staticmethod
def _apply_material_maps(obj, obj_uuid, maps, solid_weight_transfers, verbose):
"""Set one object's spatial maps, carrying a SOLID's onto its tets.
A group can hold several objects and a map may name only some of them,
so an absent uuid leaves the object unmapped. A uuid that IS named with
an empty array is a defect in the payload and raises.
"""
rows = []
for map_key, entry in maps.items():
static = entry.get("weights")
animated = entry.get("weight_frames")
if static is not None and animated is not None:
raise ValueError(
f"the '{map_key}' material map carries both a single "
"weight array and a keyed sequence; it has one or the "
"other"
)
if animated is not None:
times = entry.get("times")
if not times:
raise ValueError(
f"the '{map_key}' material map carries keyed weights "
"with no times"
)
per_object = animated.get(obj_uuid)
if per_object is None:
continue
if len(per_object) != len(times):
raise ValueError(
f"the '{map_key}' material map has "
f"{len(per_object)} weight arrays for object "
f"{obj_uuid!r} but {len(times)} times"
)
frames = [list(w) for w in per_object]
sample_times = [float(t) for t in times]
else:
weights = static or {}
if obj_uuid not in weights:
continue
frames = [list(weights[obj_uuid])]
sample_times = [0.0]
if not all(frames):
raise ValueError(
f"the '{map_key}' material map names object {obj_uuid!r} "
"but carries no weights for it"
)
rows.append((map_key, frames, sample_times, float(entry["target"])))
if not rows:
return
if obj.obj_type != "tet":
for map_key, frames, sample_times, target in rows:
if len(frames) == 1:
obj.set_param_spatial(map_key, frames[0], target)
else:
obj.set_param_spatial_anim(
map_key, frames, sample_times, target
)
return
# A solid's simulated vertices are its TETRAHEDRALIZED ones, which the
# artist never sees and never paints. The weights arrive on the Blender
# mesh and are carried across, one factorization for every map on the
# object.
if solid_weight_transfers is None:
raise RuntimeError(
f"object {obj_uuid!r} is tetrahedralized and carries a "
"spatial material map, but no transfer was supplied to carry "
"the painted weights onto its tetrahedra"
)
import numpy as np
transfer = solid_weight_transfers(obj_uuid)
# The transfer is cached per canonical tet mesh, so its own recorded
# name is whichever instance built it. Name the object being decoded.
named = getattr(obj, "name", None) or obj_uuid
for map_key, frames, _times, _target in rows:
if len(frames) > 1:
raise ValueError(
f"the '{map_key}' material map on "
f"'{named}' is keyed over time, but a "
"tetrahedralized object has no per-element material "
"schedule"
)
if len(frames[0]) != transfer.n_blender:
raise ValueError(
f"the '{map_key}' material map on "
f"'{named}' has {len(frames[0])} weights "
f"but its Blender mesh has {transfer.n_blender} vertices"
)
stacked = np.column_stack(
[np.asarray(frames[0], dtype=np.float64) for _k, frames, _t, _g in rows]
)
carried = transfer.apply(stacked, object_name=named)
for column, (map_key, _frames, _times, target) in enumerate(rows):
values = carried[:, column]
if verbose:
print(
f" material map '{map_key}' on '{named}': "
f"tet weights in [{float(values.min()):.6f}, "
f"{float(values.max()):.6f}]"
)
obj.set_param_spatial(map_key, values, target)
def apply_to_session(self, session: Session, verbose: bool = False):
"""Apply scene-level and dynamic parameters from the loaded data to ``session``.
Call :meth:`set_path` first. Static scene parameters are forwarded to
``session.param.set``; ``inactive-momentum`` is additionally set as a
dynamic hold. Dynamic parameter keyframes from ``dyn_param`` are
applied via ``session.param.dyn(...)``.
Args:
session (Session): The session to which the parameters will be applied.
verbose (bool): Enable verbose logging.
Example:
Apply scene-level and dynamic parameters to a freshly initialized session::
from frontend._decoder_ import ParamDecoder
decoder = ParamDecoder().set_path("/path/to/param.pickle")
session = app.session.create(scene).init(fixed_scene)
decoder.apply_to_session(session)
session.build().start()
"""
assert self._data is not None, "Parameter data not set. Call set_path() first."
if verbose:
print("=== Session Parameters ===")
session.param.clear_all()
for k, v in self._data["scene"].items():
if verbose:
print(f" {k}: {v}")
if k == "inactive-momentum" and v > 0:
inactive_momentum_time = float(v)
session.param.set("inactive-momentum")
session.param.dyn("inactive-momentum").time(inactive_momentum_time).hold().change(False)
else:
session.param.set(k, v)
# Apply dynamic parameters if present
dyn_param = self._data.get("dyn_param", {})
for key, entries in dyn_param.items():
if len(entries) < 2:
continue
if verbose:
print(f" dyn({key}): {len(entries)} keyframes")
# First entry is initial value (already set via static params above).
# Subsequent entries are (time, value, is_hold) tuples.
builder = session.param.dyn(key)
for entry in entries[1:]:
t, v = entry[0], entry[1]
is_hold = entry[2] if len(entry) > 2 else False
if is_hold:
builder.time(t).hold()
else:
if isinstance(v, list) and len(v) == 1:
builder.time(t).change(v[0])
else:
builder.time(t).change(v)
def apply_pin_config(self, scene, verbose: bool = False):
"""Apply saved pin configuration to scene pin holders.
Applies unpin time, pull strength, pin group id, embedded move
keyframes, and explicit pin operations (spin / scale / move_by /
torque) to each pin holder that has a matching config entry.
Call after the scene is populated and pins are created.
Example:
Reapply pin settings after populating objects::
from frontend._decoder_ import ParamDecoder
decoder = ParamDecoder().set_path("/path/to/param.pickle")
decoder.apply_to_objects(scene)
decoder.apply_pin_config(scene, verbose=True)
"""
if getattr(scene, "_pin_config_applied", False):
return
if verbose and self._pin_config:
print("=== Pin Config ===")
try:
for dyn_name, dyn_obj in scene.object_dict.items():
# pin_config keyed by UUID; objects are registered by UUID
obj_cfg = self._pin_config.get(dyn_name, {})
if not obj_cfg:
continue
# Collapse the single all-vertex holder (built by
# _apply_pin_mapping) into one holder per pin_group_id, so a
# group's ops/animation apply once to the whole group. Without
# this a keyframed N-vertex pin became N one-vertex holders,
# each carrying every keyframe op -> N*M solver pin files.
self._regroup_pin_holders(dyn_obj, obj_cfg)
self._split_solid_holder_by_threshold(dyn_obj, obj_cfg, verbose)
for pin_holder in dyn_obj.pin_list:
# For solid objects, use stored Blender indices for config lookup
lookup_indices = getattr(pin_holder._data, '_blender_pin_indices', None) or pin_holder.index
# Resolve ONE cfg for this holder. Prefer a cfg that carries a
# captured deformation (``embedded_move_index`` / its
# ``rest_shape_track`` flag) over a plain anchor cfg: a SOLID
# holder can span both a captured pin and a non-captured anchor
# (e.g. a fixed pin-root) in one merged surface mapping, and the
# first stored vert is often the anchor. Taking it would drop
# the embedded ops AND the rest-shape track for the whole
# holder. Falls back to the first non-None cfg when none is
# captured (unchanged for pure-anchor / single-intent holders).
# One field is deliberately not read off this cfg: the
# intersection allowance, which ``_apply_pin_cfg_entry``
# reduces over every contributing pin instead, so a holder
# spanning an allowing and a non-allowing pin does not
# inherit the allowance from whichever cfg wins here.
chosen_vi = None
chosen_cfg = None
for vi in lookup_indices:
cfg = obj_cfg.get(vi)
if cfg is None:
continue
if chosen_cfg is None:
chosen_vi, chosen_cfg = vi, cfg
if "embedded_move_index" in cfg or cfg.get("rest_shape_track"):
chosen_vi, chosen_cfg = vi, cfg
break
if chosen_cfg is not None:
self._apply_pin_cfg_entry(
pin_holder, dyn_name, chosen_vi, chosen_cfg, obj_cfg, verbose,
)
finally:
# Sparse LU objects are build-time accelerators and are not
# picklable. Captured motion has been materialized into operations
# above, so no frozen scene or app-state snapshot needs them.
for dyn_obj in scene.object_dict.values():
for pin_holder in dyn_obj.pin_list:
for attr in (
"_solid_pin",
"_solid_frame_map",
"_solid_surface_tri",
"_harmonic",
):
if hasattr(pin_holder._data, attr):
delattr(pin_holder._data, attr)
scene._pin_config_applied = True
@staticmethod
def _regroup_pin_holders(dyn_obj, obj_cfg):
"""Rebuild ``dyn_obj``'s pin holders as one holder per pin group.
``_apply_pin_mapping`` registers a single holder spanning every
pinned vertex of an object. An object with several distinct pin
vertex groups (e.g. separate ``left`` / ``right`` pins) needs one
holder per group so each group's config is applied to exactly its
vertices. Grouping key is ``pin_group_id`` from ``obj_cfg``.
Skips SOLID surface-mapped holders: those carry
``_blender_pin_indices`` and a sim-vertex set the regroup would
lose, and ``_apply_pin_mapping`` already builds them correctly.
"""
holders = list(dyn_obj.pin_list)
if not holders:
return
if any(getattr(h._data, "_blender_pin_indices", None)
for h in holders):
return
all_idx = [int(i) for h in holders for i in h.index]
gid_order: list = []
gid_verts: dict = {}
for vi in all_idx:
cfg = obj_cfg.get(vi)
# Verts with no config entry (plain hold-fixed pins) group
# under one sentinel key so they stay a single holder too.
key = cfg.get("pin_group_id") if cfg else "__plain__"
if key not in gid_verts:
gid_verts[key] = []
gid_order.append(key)
gid_verts[key].append(vi)
# Already one-holder-per-group: nothing to do.
if len(holders) == len(gid_order):
return
dyn_obj.pin_list.clear()
for key in gid_order:
dyn_obj.pin(gid_verts[key])
def _split_solid_holder_by_threshold(self, dyn_obj, obj_cfg, verbose=False):
"""Split a partial-pin SOLID holder into a hard (FixPair) sub-holder
and a soft (PullPair) sub-holder.
Runs for every hard-intent partial Poisson holder (no ``pull_strength``
in cfg; carries ``_solid_pin`` + ``_solid_full_w`` + ``_solid_surf_mask``).
This split used to be a correctness workaround: an interior fix pin was
a zero-diagonal CG nan, because the solver assembled the fix barrier
over surface verts only and gated off inertia for fix pins. That is no
longer true -- a fix pin is an exact Dirichlet BC whose diagonal block
becomes the identity, so an interior fix pin is now well posed.
The split is kept as what it also always was: an AUTHORING control. A
partially pinned SOLID gets a hard core and a soft skirt, and the
per-pin ``fix_weight_threshold`` (cfg, default
``_DEFAULT_FIX_WEIGHT_THRESHOLD``) sets where the boundary falls. 0
makes every surface driven vert hard and the interior soft; higher
values soften the low-weight surface skirt. Retiring the split would
silently stiffen every saved partial-pin SOLID scene, so it is a
separate, separately announced change.
Each sub-holder reuses the shared sparse solve maps with its own
full-axis ``keep`` mask; the move-op builder slices
``positions[:, keep, :]`` so the masks must live on that full axis.
The helper owns the pull calls because a hard-intent cfg carries no
``pull_strength``, so ``_apply_pin_cfg_entry``'s pull block never fires
for these holders. Idempotent via ``_solid_split_done``. Never pins an
empty index list.
"""
import numpy as np
def _carry(sub, src):
sub._data._blender_pin_indices = list(
getattr(src, "_blender_pin_indices", []) or []
)
sub._data._tet_V = getattr(src, "_tet_V", None)
sub._data._blender_vert = getattr(src, "_blender_vert", None)
sub._data._solid_split_done = True
for holder in list(dyn_obj.pin_list):
d = holder._data
if getattr(d, "_solid_split_done", False):
continue
# cfg lookup: first stored Blender pin index that has a cfg entry
# (mirrors the per-holder loop in apply_pin_config).
lookup = getattr(d, "_blender_pin_indices", None) or holder.index
cfg = None
for vi in lookup:
c = obj_cfg.get(vi)
if c is not None:
cfg = c
break
if cfg is None:
continue
# Mixed-intent full-pin harmonic holder: a later HARD pin (e.g.
# pin-root) overwrote the pull cfg on a subset of verts via
# last-wins, so some surface verts are hard (no pull_strength) and
# some are pull. The single harmonic holder would otherwise take
# the first vertex's intent (pull) for ALL verts, leaving the hard
# verts never fixed. Extract those hard SURFACE verts into a
# FixPair holder (held at rest) so they are rigidly fixed; the
# original holder stays the pull holder (its pull surface +
# interior keep following the captured target). Hard verts are
# surface-only (the hard-core / soft-skirt split, see
# _split_solid_pin_holder), and a static hard pin has no captured
# track so it holds at rest
# (the pull holder already pulls those verts toward rest too, so
# there is no conflict; the FixPair just makes them rigid).
harmonic = getattr(d, "_harmonic", None)
simw = getattr(d, "_sim_blender_weights", None)
if harmonic is not None and simw is not None:
n_surf = int(harmonic[0])
surf_sim = list(holder.index[:n_surf])
hard_sim, hard_blender, pull_blender = [], set(), set()
for j in range(min(n_surf, len(simw))):
corners = simw[j] or []
hard_w = sum(w for b, w in corners
if "pull_strength" not in obj_cfg.get(int(b), {}))
pull_w = sum(w for b, w in corners
if "pull_strength" in obj_cfg.get(int(b), {}))
for b, w in corners:
if "pull_strength" in obj_cfg.get(int(b), {}):
pull_blender.add(int(b))
if corners and hard_w > pull_w:
hard_sim.append(int(surf_sim[j]))
for b, w in corners:
if "pull_strength" not in obj_cfg.get(int(b), {}):
hard_blender.add(int(b))
d._solid_split_done = True
if hard_sim and len(hard_sim) < n_surf:
fh = dyn_obj.pin(hard_sim)
fh._data._blender_pin_indices = sorted(hard_blender)
fh._data._tet_V = getattr(d, "_tet_V", None)
fh._data._blender_vert = getattr(d, "_blender_vert", None)
fh._data._solid_split_done = True
# _apply_pin_cfg_entry finds the hard pin's cfg (no
# pull_strength, no captured track) => stationary FixPair
# at rest. The original holder stays the pull holder, but
# the per-holder loop resolves its intent from the FIRST
# stored blender vert, which may now be a hard/root vert
# whose cfg carries neither pull_strength nor a captured
# move (so both would be dropped, freezing the body).
# Re-point its cfg lookup at the pull verts so pull +
# the captured target reliably apply. The captured-move
# builder uses _sim_blender_weights, not this list, so
# narrowing it is safe.
if pull_blender:
d._blender_pin_indices = sorted(pull_blender)
if verbose:
print(f" solid harmonic mixed: {len(hard_sim)} hard "
f"surf -> FixPair(rest); "
f"{n_surf - len(hard_sim)} pull surf + interior")
continue
# Gates: only a hard-intent (no pull_strength) partial Poisson
# SOLID holder is split. The toggle does NOT gate whether the
# split runs (interior fix pins always nan); it only sets thr.
if "pull_strength" in cfg:
continue # pure-pull intent never hardens
# Torque pins are merged by pin_group_id in the solver; do not
# split them into two holders that would share one id.
if any(op.get("type") == "torque"
for op in cfg.get("operations", [])):
continue
sp = getattr(d, "_solid_pin", None)
fw = getattr(d, "_solid_full_w", None)
df = getattr(d, "_solid_driven_full", None)
sm = getattr(d, "_solid_surf_mask", None)
if sp is None or fw is None or df is None or sm is None:
continue # full_pin / harmonic / SHELL / ROD: no partial fields
# Per-pin threshold (default _DEFAULT_FIX_WEIGHT_THRESHOLD;
# 0 => every surface driven hard).
thr = float(
cfg.get("fix_weight_threshold", _DEFAULT_FIX_WEIGHT_THRESHOLD)
)
keep = np.asarray(sp["keep"]) # full axis, bool
full_w = np.asarray(fw) # full axis
df_arr = np.asarray(df)
surf_mask = np.asarray(sm) # full axis, bool
# Hard FixPairs are SURFACE-ONLY. This is the hard-core /
# soft-skirt authoring split, not a solver limitation: an interior
# fix pin is well posed now that a fix pin is an exact Dirichlet BC
# (its diagonal block becomes the identity). Interior high-weight
# verts fall through to soft pull.
# Intent-aware hardening: a pull-intent surface vert (its Blender
# corners are dominated by pull pins) must NEVER harden. Such verts
# share this merged holder only because a hard pin-root overlaps
# the pull region; freezing them strands the captured target so the
# body sits at initial geometry instead of following the deformer.
# Classify each driven surface vert by its stored Blender corners
# (mirrors the full-pin harmonic mixed-intent path above).
simw = getattr(d, "_sim_blender_weights", None)
frame_map = getattr(d, "_solid_frame_map", None)
n_surf_driven = int(surf_mask.sum())
# Three-way intent classification of each driven SURFACE vert by
# its stored Blender corners:
# * static anchor (pin-root): corners carry NEITHER pull_strength
# NOR a captured track -> a stationary FixPair held at rest,
# NEVER driven by the captured diffusion. (Otherwise the
# least-squares map extrapolates the moving pin's motion onto the
# anchor region and it drifts.)
# * pull follow: pull-strength corners dominate -> soft PullPair
# toward the captured target;
# * captured-fix follow: a captured corner with no pull (a fixed
# barrier pin) -> hard FixPair that follows the captured target.
# Distinguishing "no pull" from "no pull AND no capture" is what
# lets a FIXED captured pin coexist with a static pin-root: both
# lack pull_strength, so the old pull-only test merged them and the
# anchor followed the pin.
pull_surf = np.zeros(len(df_arr), dtype=bool)
static_surf = np.zeros(len(df_arr), dtype=bool)
# A Poisson weight can diffuse beyond the painted source region.
# Exact motion needs all three source-triangle tracks so its full
# local frame, including normal offset, is reconstructible.
local_surf = np.ones(len(df_arr), dtype=bool)
has_captured_tracks = any(
bool(c.get("pin_anim")) for c in obj_cfg.values()
)
if frame_map is not None and has_captured_tracks:
local_surf[:] = False
frame_tri = np.asarray(frame_map["triangles"], dtype=np.int64)
for j in range(min(n_surf_driven, len(frame_tri))):
tracks = [
obj_cfg.get(int(b), {})
.get("pin_anim", {})
.get(int(b))
for b in frame_tri[j]
]
if any(track is None for track in tracks):
continue
times = [list(track["time"]) for track in tracks]
local_surf[j] = (
len(times[0]) >= 2
and times[1] == times[0]
and times[2] == times[0]
and all(
np.asarray(track["position"]).shape
== (len(times[0]), 3)
for track in tracks
)
)
if simw is not None:
for j in range(min(n_surf_driven, len(simw))):
corners = simw[j] or []
if not corners:
continue
pw = sum(w for b, w in corners
if "pull_strength" in obj_cfg.get(int(b), {}))
static_w = sum(
w for b, w in corners
if "pull_strength" not in obj_cfg.get(int(b), {})
and "embedded_move_index" not in obj_cfg.get(int(b), {})
)
follow_w = sum(w for b, w in corners) - static_w
if static_w > follow_w:
static_surf[j] = True
elif pw > follow_w - pw: # pull dominates the captured part
pull_surf[j] = True
# Static is SURFACE-only (an interior fix pin is a zero-diagonal CG
# nan; pin-root is painted on the surface anyway).
static_keep = keep & surf_mask & static_surf
hard_keep = (keep & (full_w >= thr) & surf_mask
& local_surf & ~pull_surf & ~static_surf)
surface_tri = getattr(d, "_solid_surface_tri", None)
exact_candidates = static_keep | hard_keep
exact_keep = _independent_surface_pin_mask(
df_arr,
exact_candidates,
surface_tri,
full_w + 2.0 * static_keep,
)
static_soft_keep = static_keep & ~exact_keep
static_keep &= exact_keep
hard_keep &= exact_keep
soft_keep = keep & ~hard_keep & ~static_keep
soft_keep &= ~static_soft_keep
static_index = [int(df_arr[k]) for k in range(len(df_arr))
if static_keep[k]]
static_soft_index = [
int(df_arr[k]) for k in range(len(df_arr))
if static_soft_keep[k]
]
hard_index = [int(df_arr[k]) for k in range(len(df_arr))
if hard_keep[k]]
soft_index = [int(df_arr[k]) for k in range(len(df_arr))
if soft_keep[k]]
# Blender-vert partitions for cfg re-pointing. Follow sub-holders
# resolve their cfg from a captured/pull vert (so pull_strength +
# rest_shape_track + the captured ops apply); the static FixPair
# resolves from a pin-root vert (no captured track => held at rest).
bpi = list(getattr(d, "_blender_pin_indices", []) or [])
def _is_static_cfg(vi):
c = obj_cfg.get(int(vi), {})
return ("pull_strength" not in c
and "embedded_move_index" not in c)
follow_blender = sorted(vi for vi in bpi if not _is_static_cfg(vi))
static_blender = sorted(vi for vi in bpi if _is_static_cfg(vi))
# Re-point each follow sub-holder at the cfg of ITS OWN intent. A
# follow vert is either pull (has ``pull_strength``) or captured-fix
# (has ``embedded_move_index`` only). The hard FixPair MUST resolve a
# captured-fix cfg, otherwise _apply_pin_cfg_entry sees a pull vert's
# ``pull_strength`` and overrides its pull(0.0) into a soft pull; the
# soft PullPair MUST resolve a pull cfg so the pull_strength scaling
# applies. Without this split, both holders shared ``follow_blender``
# and the first vertex's intent leaked across (wrong for a SOLID that
# mixes captured-fix and pull pins). Fall back to the combined list
# when a category has no vertex of its own.
pull_blender = [
vi for vi in follow_blender
if "pull_strength" in obj_cfg.get(int(vi), {})
]
fix_blender = [
vi for vi in follow_blender
if "pull_strength" not in obj_cfg.get(int(vi), {})
]
hard_blender = fix_blender or follow_blender
soft_blender = pull_blender or follow_blender
# Rebuild this object's holders from the three categories. Each
# non-empty category becomes its own holder (``pin([])`` is never
# called), replacing the original merged holder.
dyn_obj.pin_list.remove(holder)
d._solid_split_done = True
if static_index:
fa = dyn_obj.pin(static_index)
_carry(fa, d)
if static_blender:
fa._data._blender_pin_indices = static_blender
# No _solid_pin => no captured ops; the pin-root cfg carries no
# embedded move, so this stays a stationary FixPair at rest.
fa.pull(0.0)
if static_soft_index:
sa = dyn_obj.pin(static_soft_index)
_carry(sa, d)
if static_blender:
sa._data._blender_pin_indices = static_blender
sa_w = full_w[static_soft_keep].astype(np.float32)
sa.pull(1.0)
sa.pull_per_vertex(sa_w)
if hard_index:
h = dyn_obj.pin(hard_index)
_carry(h, d)
if frame_map is not None and has_captured_tracks:
frame_tri = np.asarray(
frame_map["triangles"], dtype=np.int64
)
frame_coef = np.asarray(
frame_map["coefs"], dtype=np.float64
)
hard_surface = hard_keep[:n_surf_driven]
h._data._solid_frame_map = {
"triangles": frame_tri[hard_surface],
"coefs": frame_coef[hard_surface],
}
else:
h._data._solid_pin = {
"surface_map": sp["surface_map"],
"interior_map": sp["interior_map"],
"motion_cache": sp["motion_cache"],
"keep": hard_keep,
"n_input": sp["n_input"],
"rest_full": sp.get("rest_full"),
}
if hard_blender:
h._data._blender_pin_indices = hard_blender
h.pull(0.0) # FixPair (exact Dirichlet fix); captured ops make it kinematic
if soft_index:
s = dyn_obj.pin(soft_index)
_carry(s, d)
s._data._solid_pin = {
"surface_map": sp["surface_map"],
"interior_map": sp["interior_map"],
"motion_cache": sp["motion_cache"],
"keep": soft_keep,
"n_input": sp["n_input"],
"rest_full": sp.get("rest_full"),
}
s_w = full_w[soft_keep].astype(np.float32)
# Compacted per-vertex weight for the cfg pass's pull scaling.
s._data._solid_pin_weights = s_w
if soft_blender:
s._data._blender_pin_indices = soft_blender
s.pull(1.0)
s.pull_per_vertex(s_w)
if verbose:
print(f" solid split (capture-aware) @ thr={thr:.3f}: "
f"{len(static_index)} static-rest / "
f"{len(static_soft_index)} static-soft / "
f"{len(hard_index)} hard-follow / "
f"{len(soft_index)} soft-follow")
def _apply_pin_cfg_entry(self, pin_holder, dyn_name, vi, cfg, obj_cfg,
verbose):
"""Apply a single pin-config entry to ``pin_holder``."""
if "unpin_time" in cfg:
pin_holder.unpin(cfg["unpin_time"])
if verbose:
print(f" {dyn_name}[{vi}]: unpin_time={cfg['unpin_time']}")
if "pull_strength" in cfg:
pin_holder.pull(cfg["pull_strength"])
# Partial-pin SOLID: scale the diffused [0,1] per-vertex weight
# field by pull_strength so each tet vertex is pulled with
# strength pull_strength * weight (the scalar pull() above is
# kept as the soft-vs-hard classification marker).
sp_weights = getattr(pin_holder._data, "_solid_pin_weights", None)
if sp_weights is not None:
import numpy as np
pin_holder.pull_per_vertex(
float(cfg["pull_strength"]) * np.asarray(sp_weights)
)
if "pin_group_id" in cfg:
# pin_group_id is a mirrored field, but the decode-time override
# writes only the canonical _data; the Rust validator mirror is
# not updated and may go stale here. That is harmless today
# because no consumer (export, scene builder) reads the mirror's
# group id, only _data. If export is ever taught to read the
# mirror, route this through a setter that updates both.
pin_holder._data.pin_group_id = cfg["pin_group_id"]
if verbose:
print(f" {dyn_name}[{vi}]: pin_group_id={cfg['pin_group_id']}")
# Per-pin intersection allowance. Read from EVERY cfg the holder's
# vertices resolve to, not from the single chosen ``cfg``, because a
# holder can span more than one Blender pin: ``_regroup_pin_holders``
# leaves SOLID surface-mapped holders alone, and
# ``_split_solid_holder_by_threshold`` builds sub-holders whose stored
# Blender indices mix a captured pin with a static anchor. One holder
# carries one flag, so a mixed holder resolves to False, which is the
# same unanimity the solver applies per vertex (a vertex is exempt only
# when every pin covering it asks for it, and an unpinned vertex never
# is). Granting the flag on a mixed holder would instead exempt
# elements held by a pin that never asked; splitting the holder by flag
# would cut the pin grouping that the surface mapping and the threshold
# split are built around. A vertex whose cfg is absent belongs to a pin
# that emitted no settings at all, so it is a pin that did not ask.
# Applies to both pin modes and needs no operations, so it is resolved
# before the ops-only early return below.
lookup_indices = (
getattr(pin_holder._data, "_blender_pin_indices", None)
or pin_holder.index
)
allow_intersection = bool(lookup_indices) and all(
(obj_cfg.get(v) or {}).get("allow_intersection", False)
for v in lookup_indices
)
pin_holder._data.allow_intersection = allow_intersection
if verbose and allow_intersection:
print(f" {dyn_name}[{vi}]: allow_intersection=True")
if "operations" not in cfg and "embedded_move_index" not in cfg:
return
embedded_ops = self._build_embedded_move_ops(pin_holder, obj_cfg)
embedded_move_index = cfg.get("embedded_move_index", -1)
explicit_operations = cfg.get("operations", [])
# Validate: torque cannot be mixed with kinematic ops
_rust.validate_pin_op_types(
[str(o.get("type", "")) for o in explicit_operations],
dyn_name,
)
# Clear existing operations
pin_holder._data.operations = []
# Re-add in the correct order
explicit_idx = 0
embedded_inserted = False
total = len(explicit_operations) + (1 if embedded_move_index >= 0 else 0)
if total == 0 and embedded_ops:
# Only embedded move, no explicit ops
pin_holder._data.operations.extend(embedded_ops)
embedded_inserted = True
for pos in range(total):
if pos == embedded_move_index and not embedded_inserted:
# Insert embedded move operations at this position
pin_holder._data.operations.extend(embedded_ops)
embedded_inserted = True
else:
if explicit_idx < len(explicit_operations):
op = explicit_operations[explicit_idx]
explicit_idx += 1
self._dispatch_pin_op(pin_holder, op)
# If embedded move wasn't inserted yet (index beyond end), append
if not embedded_inserted and embedded_move_index >= 0:
pin_holder._data.operations.extend(embedded_ops)
# Flag a captured deformation so the scene builder emits a time-varying
# rest shape from this holder's target trajectory. Independent of the
# pin constraint type: a pull pin (soft) and a fixed pin (hard FixPair
# barrier) BOTH track the rest shape, so the gate is NOT conditioned on
# ``pull_strength``. The encoder sets ``rest_shape_track`` only for
# has_captured_anim SOLID/SHELL pins, so explicit Move/Spin/fcurve-
# keyframed pins never trip it.
if cfg.get("rest_shape_track") and embedded_ops:
pin_holder._data.rest_shape_track = True
if verbose:
print(f" {dyn_name}[{vi}]: operations={len(pin_holder.operations)}")
@staticmethod
def _build_embedded_move_ops(pin_holder, obj_cfg):
"""Build embedded ``MoveByOperation`` segments from ``pin_anim``.
SHELL / ROD: the holder's sim indices ARE Blender vertex indices,
so each pinned vertex carries its own single-entry ``pin_anim``
track (``{vertex_index: PinAnim}``) and consecutive keyframes
become per-vertex ``MoveByOperation`` deltas. The pin deforms โ it
is not rigidly translated.
SOLID: the holder spans simulation surface vertices that do not
line up with Blender vertices (fTetWild remeshes the surface), so
the holder carries ``_sim_blender_weights`` and the transfer is
delegated to :meth:`_build_solid_embedded_move_ops`.
A partial-pin SOLID holder carries BOTH ``_solid_pin`` (the two-stage
Poisson operators with a ``keep`` mask compacting positions to the
driven subset = ``holder.index``) AND ``_sim_blender_weights`` (a
full-surface inverse map the threshold split uses for intent
classification). ``_solid_pin`` must win: its builder slices to the
driven subset, whereas ``_build_solid_embedded_move_ops`` emits one
delta per surface vertex (full surface, no ``keep``), so taking the
``_sim_blender_weights`` branch here would hand a full-surface-length
delta to a driven-subset-length holder ("delta length N mismatch with
vertex M"). Full-pin holders never set ``_solid_pin``, so they still
reach the harmonic surface+interior builder below.
"""
import numpy as np
from ._scene_pin_ import MoveByOperation
solid_pin = getattr(pin_holder._data, "_solid_pin", None)
if solid_pin is not None:
return ParamDecoder._build_solid_poisson_move_ops(solid_pin, obj_cfg)
solid_frame_map = getattr(
pin_holder._data, "_solid_frame_map", None
)
if solid_frame_map is not None:
return ParamDecoder._build_solid_frame_move_ops(
solid_frame_map, obj_cfg
)
sim_weights = getattr(pin_holder._data, "_sim_blender_weights", None)
if sim_weights is not None:
harmonic = getattr(pin_holder._data, "_harmonic", None)
return ParamDecoder._build_solid_embedded_move_ops(
sim_weights, obj_cfg, harmonic,
)
index = list(pin_holder.index)
n_pin_verts = len(index)
# Each vertex's own track lives in its own cfg entry, keyed by
# that same vertex index. Gather them in pin_holder.index order.
tracks = []
times = None
for v in index:
cfg = obj_cfg.get(v)
track = cfg.get("pin_anim", {}).get(v) if cfg else None
tracks.append(track)
if track is not None and times is None:
times = list(track["time"])
if times is None or len(times) < 2:
return []
n_frames = len(times)
# Per-vertex absolute-position tracks, aligned to the
# pin_holder.index order. A vertex with no track keeps a zero
# track (zero delta โ unaffected by the embedded move).
positions = np.zeros((n_frames, n_pin_verts, 3), dtype=np.float64)
for j, track in enumerate(tracks):
if track is not None:
pos = np.asarray(track["position"], dtype=np.float64)
if pos.shape == (n_frames, 3):
positions[:, j, :] = pos
embedded_ops: list = []
for k in range(n_frames - 1):
embedded_ops.append(MoveByOperation(
delta=np.ascontiguousarray(positions[k + 1] - positions[k]),
t_start=times[k],
t_end=times[k + 1],
transition="linear",
))
return embedded_ops
@staticmethod
def _build_solid_frame_move_ops(frame_map, obj_cfg):
"""Reconstruct hard SOLID pin motion in each source triangle frame."""
import numpy as np
triangles = np.asarray(frame_map["triangles"], dtype=np.int64)
coefs = np.asarray(frame_map["coefs"], dtype=np.float64)
if triangles.shape != (len(coefs), 3) or coefs.shape[1:] != (3,):
raise ValueError(
"SOLID pin frame map must contain aligned (N,3) arrays"
)
tracks = {}
times = None
for vertex in np.unique(triangles):
cfg = obj_cfg.get(int(vertex), {})
track = cfg.get("pin_anim", {}).get(int(vertex))
if track is None:
raise ValueError(
f"SOLID hard pin source vertex {int(vertex)} has no captured track"
)
track_times = list(track["time"])
positions = np.asarray(track["position"], dtype=np.float64)
if positions.shape != (len(track_times), 3):
raise ValueError(
f"SOLID hard pin source vertex {int(vertex)} has "
f"track shape {positions.shape}, expected ({len(track_times)}, 3)"
)
if times is None:
times = track_times
elif track_times != times:
raise ValueError("SOLID hard pin source tracks use different times")
tracks[int(vertex)] = positions
if times is None or len(times) < 2:
return []
tri_positions = np.stack(
[tracks[int(vertex)] for vertex in triangles.reshape(-1)],
axis=0,
).reshape(len(triangles), 3, len(times), 3).transpose(2, 0, 1, 3)
x0 = tri_positions[:, :, 0]
b1 = tri_positions[:, :, 1] - x0
b2 = tri_positions[:, :, 2] - x0
normal = np.cross(b1, b2)
normal_sq = np.einsum("fnc,fnc->fn", normal, normal)
valid_normal = normal_sq >= 1e-20
unit_normal = np.zeros_like(normal)
unit_normal[valid_normal] = (
normal[valid_normal] / np.sqrt(normal_sq[valid_normal])[:, None]
)
positions = (
x0
+ coefs[None, :, 0, None] * b1
+ coefs[None, :, 1, None] * b2
+ coefs[None, :, 2, None] * unit_normal
)
return ParamDecoder._positions_to_move_ops(positions, times)
@staticmethod
def _build_solid_embedded_move_ops(sim_weights, obj_cfg, harmonic=None):
"""Build SOLID capture-deformation MoveBy segments.
``sim_weights`` lists, per pinned sim SURFACE vertex (in
``holder.index`` order), the ``(blender_index, weight)`` pairs of
the Blender pins that embed onto it; each surface vertex follows the
weight-averaged displacement of those pins.
``harmonic`` (``(n_surface, sparse_map)`` or ``None``) drives the tet
interior when the whole surface is pinned. The sparse map solves the
Laplace equation with the surface as its Dirichlet boundary.
Consecutive deltas move each interior vertex by the harmonic extension
of the surface displacement. With both surface and interior
prescribed the SOLID is fully kinematic, so its elastic interior
cannot buckle into self-intersection.
``holder.index`` is ordered ``surface_ids + interior_ids`` to match.
"""
import numpy as np
from ._scene_pin_ import MoveByOperation
n_surf = len(sim_weights)
# Resolve each contributing Blender vertex's captured track once.
blender_tracks: dict = {}
times = None
for pairs in sim_weights:
for b, _w in pairs:
if b in blender_tracks:
continue
cfg = obj_cfg.get(b)
track = cfg.get("pin_anim", {}).get(b) if cfg else None
blender_tracks[b] = track
if track is not None and times is None:
times = list(track["time"])
if times is None or len(times) < 2:
return []
n_frames = len(times)
# Surface vertex track = ฮฃ wยทpos(b) / ฮฃ w over the embedding Blender
# pins. Consecutive deltas cancel the absolute offset, leaving the
# weight-averaged displacement the sim vertex follows.
surf_pos = np.zeros((n_frames, n_surf, 3), dtype=np.float64)
for j, pairs in enumerate(sim_weights):
acc = np.zeros((n_frames, 3), dtype=np.float64)
wsum = 0.0
for b, w in pairs:
track = blender_tracks.get(b)
if track is None:
continue
pos = np.asarray(track["position"], dtype=np.float64)
if pos.shape == (n_frames, 3):
acc += w * pos
wsum += w
if wsum > 0.0:
surf_pos[:, j, :] = acc / wsum
if harmonic is not None:
_, M = harmonic
# Interior = harmonic extension of the surface positions. The
# sparse map solves only these frame-coordinate right-hand sides.
interior_pos = _apply_sparse_frame_map(M, surf_pos)
positions = np.concatenate([surf_pos, interior_pos], axis=1)
else:
positions = surf_pos
embedded_ops: list = []
for k in range(n_frames - 1):
embedded_ops.append(MoveByOperation(
delta=np.ascontiguousarray(positions[k + 1] - positions[k]),
t_start=times[k],
t_end=times[k + 1],
transition="linear",
))
return embedded_ops
@staticmethod
def _rigid_fit(P, Q):
"""Best rigid ``R, t`` (Kabsch) mapping rest points ``P`` onto captured
``Q`` (unweighted). Falls back to a pure translation (``R = I``) for a
degenerate set (fewer than 3 points) so a rank-deficient fit never adds
a bogus rotation. ``R`` maps ``P`` to ``Q``: ``Q ~= P @ R.T + t``.
"""
import numpy as np
if P.shape[0] == 0:
return np.eye(3), np.zeros(3)
cp = P.mean(axis=0)
cq = Q.mean(axis=0)
if P.shape[0] < 3:
return np.eye(3), cq - cp
H = (P - cp).T @ (Q - cq)
U, _s, Vt = np.linalg.svd(H)
D = np.eye(3)
if np.linalg.det(Vt.T @ U.T) < 0.0:
D[2, 2] = -1.0
R = Vt.T @ D @ U.T
return R, cq - R @ cp
@staticmethod
def _build_solid_poisson_move_ops(solid_pin, obj_cfg):
"""Build MoveBy segments for a partially-pinned SOLID via the
two-stage Poisson operators.
``solid_pin`` carries matrix-free sparse maps for the surface
least-squares solve and optional interior harmonic solve, plus a
``keep`` mask selecting the driven tet verts (holder.index order =
surface + interior). Only pinned input verts have captured tracks; the
others sit at zero and are removed by the surface map's weighted RHS.
Rigid-aware diffusion. Both sparse maps are linear, so diffusing the
absolute captured positions directly does NOT reproduce a rigid input
motion: a least-squares fit of a rotation over a partial pin set has no
partition-of-unity / affine precision, so it extrapolates non-rigidly
(the linear-blend-skinning shrinkage) and injects a spurious bend that
grows with the rotation angle into BOTH the pulled targets and the
captured rest shape. So per frame, factor the rigid motion ``R_g, t_g``
out of the pinned inputs, diffuse only the residual, and recombine
``rigid(rest) + diffuse(residual)``. A rigid capture has zero residual,
so every driven tet vert lands exactly on ``R_g rest + t_g``. When
the surface map does reproduce rigids this is identical to direct
diffusion, so only the spurious component changes.
"""
import numpy as np
surface_map = solid_pin["surface_map"]
interior_map = solid_pin["interior_map"]
keep = np.asarray(solid_pin["keep"])
n_input = int(solid_pin["n_input"])
motion_cache = solid_pin["motion_cache"]
cached = motion_cache.get("value")
if cached is not None:
positions, times = cached
return ParamDecoder._positions_to_move_ops(
positions[:, keep, :], times,
)
# Gather captured per-Blender-vertex tracks (input space).
tracks: dict = {}
times = None
for b, cfg in obj_cfg.items():
track = cfg.get("pin_anim", {}).get(b) if cfg else None
if track is not None:
tracks[int(b)] = np.asarray(track["position"], dtype=np.float64)
if times is None:
times = list(track["time"])
if times is None or len(times) < 2:
return []
n_frames = len(times)
d_frame = np.zeros((n_frames, n_input, 3), dtype=np.float64)
for b, pos in tracks.items():
if 0 <= b < n_input and pos.shape == (n_frames, 3):
d_frame[:, b, :] = pos
def diffuse_frames(vec_in):
# vec_in: (frames,n_input,3) -> (frames,n_surf[+n_int],3).
surf = _apply_sparse_frame_map(surface_map, vec_in)
if interior_map is None:
return surf
interior = _apply_sparse_frame_map(interior_map, surf)
return np.concatenate([surf, interior], axis=1)
# Rigid reference = the solver's TRUE tet rest on the driven_full axis.
# The MoveBy deltas accumulate onto rest0[holder.index] downstream (the
# pull target and the captured rest shape), so the rigid part must carry
# that exact rest. Falling back to the diffused rest for payloads
# without ``rest_full`` leaves a small rest/LS-fit mismatch that the
# rotation scales back into a bend, so prefer the true rest.
pinned = sorted(b for b in tracks if 0 <= b < n_input)
rest_in = d_frame[0]
rest_full = solid_pin.get("rest_full")
rest_full = (
np.asarray(rest_full, dtype=np.float64)
if rest_full is not None else diffuse_frames(rest_in[None, :, :])[0]
)
P0 = rest_in[pinned]
residual = np.zeros_like(d_frame)
rigid_positions = np.empty(
(n_frames, rest_full.shape[0], 3), dtype=np.float64
)
for f in range(n_frames):
Rg, tg = ParamDecoder._rigid_fit(P0, d_frame[f, pinned])
residual[f, pinned] = d_frame[f, pinned] - (P0 @ Rg.T + tg)
rigid_positions[f] = rest_full @ Rg.T + tg
positions = rigid_positions + diffuse_frames(residual)
motion_cache["value"] = (positions, times)
return ParamDecoder._positions_to_move_ops(
positions[:, keep, :], times,
)
@staticmethod
def _positions_to_move_ops(positions, times):
"""Convert absolute per-frame positions into linear MoveBy segments."""
import numpy as np
from ._scene_pin_ import MoveByOperation
return [
MoveByOperation(
delta=np.ascontiguousarray(positions[k + 1] - positions[k]),
t_start=times[k],
t_end=times[k + 1],
transition="linear",
)
for k in range(len(times) - 1)
]
def _dispatch_pin_op(self, pin_holder, op):
"""Dispatch a single explicit pin op to its per-type handler."""
t_start = op.get("t_start", 0)
t_end = op.get("t_end", 1)
transition = op.get("transition", "linear")
if transition != "linear":
pin_holder.interp(transition)
op_type = op["type"]
if op_type == "spin":
self._apply_pin_op_spin(pin_holder, op, t_start, t_end)
elif op_type == "scale":
self._apply_pin_op_scale(pin_holder, op, t_start, t_end)
elif op_type == "move_by":
self._apply_pin_op_move_by(pin_holder, op, t_start, t_end)
elif op_type == "torque":
self._apply_pin_op_torque(pin_holder, op, t_start, t_end)
if transition != "linear":
pin_holder.interp("linear")
@staticmethod
def _apply_pin_op_spin(pin_holder, op, t_start, t_end):
center_mode = op.get("center_mode", "absolute")
pin_holder.spin(
center=op.get("center"),
axis=op.get("axis", [0, 1, 0]),
angular_velocity=op.get("angular_velocity", 360),
t_start=t_start,
t_end=t_end,
center_mode=center_mode,
)
@staticmethod
def _apply_pin_op_scale(pin_holder, op, t_start, t_end):
center_mode = op.get("center_mode", "absolute")
pin_holder.scale(
op.get("factor", 1.0),
t_start=t_start,
t_end=t_end,
center=op.get("center"),
center_mode=center_mode,
)
@staticmethod
def _apply_pin_op_move_by(pin_holder, op, t_start, t_end):
pin_holder.move_by(
op.get("delta", [0, 0, 0]),
t_start=t_start,
t_end=t_end,
)
@staticmethod
def _apply_pin_op_torque(pin_holder, op, t_start, t_end):
blender_hint = int(op.get("hint_vertex", -1))
# Translate Blender hint vertex to nearest tet vertex by position
# (robust to re-tetrahedralization). Closest-vertex search
# lives in Rust (`dec::closest_vertex_index`).
tet_V = getattr(pin_holder._data, '_tet_V', None)
blender_vert = getattr(pin_holder._data, '_blender_vert', None)
if tet_V is not None and blender_vert is not None and blender_hint >= 0:
import numpy as np
hint_pos = np.ascontiguousarray(
np.asarray(blender_vert[blender_hint], dtype=np.float64)
)
tet_arr = np.ascontiguousarray(
np.asarray(tet_V, dtype=np.float64)
)
sim_hint = int(_rust.closest_vertex_index(tet_arr, hint_pos))
else:
sim_hint = blender_hint
pin_holder.torque(
magnitude=op.get("magnitude", 1.0),
axis_component=int(op.get("axis_component", 2)),
hint_vertex=sim_hint,
t_start=t_start,
t_end=t_end,
)
def apply_invisible_colliders(self, scene, verbose: bool = False):
"""Create invisible wall and sphere colliders on the scene from the loaded pickle data.
Must be called BEFORE ``scene.build()``.
Example:
Add the pickle's invisible walls and spheres just before building::
from frontend._decoder_ import ParamDecoder
decoder = ParamDecoder().set_path("/path/to/param.pickle")
decoder.apply_invisible_colliders(scene)
fixed_scene = scene.build()
"""
ic = self._data.get("invisible_colliders", {})
if not ic:
return
if verbose:
print("=== Invisible Colliders ===")
for w in ic.get("walls", []):
wall = scene.add.invisible.wall(w["position"], w["normal"])
_apply_common_collider_params(wall, w, "wall")
keyframes = w.get("keyframes", [])
for kf in keyframes[1:]:
wall.move_to(kf["position"], kf["time"])
if verbose:
print(f" Wall: pos={w['position']}, normal={w['normal']}, kf={len(keyframes)}")
for s in ic.get("spheres", []):
sphere = scene.add.invisible.sphere(s["position"], s["radius"])
if s.get("hemisphere", False):
sphere.hemisphere()
if s.get("invert", False):
sphere.invert()
_apply_common_collider_params(sphere, s, "sphere")
keyframes = s.get("keyframes", [])
for kf in keyframes[1:]:
sphere.transform_to(kf["position"], kf["radius"], kf["time"])
if verbose:
print(f" Sphere: pos={s['position']}, r={s['radius']}, inv={s.get('invert')}, hemi={s.get('hemisphere')}, kf={len(keyframes)}")
def _apply_common_collider_params(collider, spec, kind):
"""Set the shared contact-gap / friction / active-duration / thickness
defaults on an invisible wall or sphere collider from ``spec``.
``kind`` is the "wall"/"sphere" label passed to the thickness validator.
"""
collider.param.set("contact-gap", spec.get("contact_gap", 1e-3))
collider.param.set("friction", spec.get("friction", 0.0))
collider.param.set("active-duration", spec.get("active_duration", -1.0))
thickness = _rust.validate_invisible_collider_thickness(
kind, float(spec.get("thickness", 1.0))
)
collider.param.set("thickness", thickness)
def _graph_laplacian_from_edges(e0, e1, n):
"""Binary-adjacency graph Laplacian ``L = diag(deg) - A`` (CSR) over the
undirected edges ``(e0[i], e1[i])``. Edges are symmetrized and the
per-element multiplicity is collapsed to a binary adjacency so ``L`` is
the standard graph Laplacian. ``n`` is the vertex count."""
import numpy as np
import scipy.sparse as sp
rows = np.concatenate([e0, e1])
cols = np.concatenate([e1, e0])
adj = sp.coo_matrix(
(np.ones(rows.shape[0], dtype=np.float64), (rows, cols)),
shape=(n, n),
).tocsr()
adj.data[:] = 1.0
deg = np.asarray(adj.sum(axis=1)).ravel()
return (sp.diags(deg) - adj).tocsr()
def _independent_surface_pin_mask(
driven_full, candidates, surface_tri, priority,
):
"""Select exact pins with no surface edge joining two selected vertices."""
import numpy as np
candidates = np.asarray(candidates, dtype=bool)
selected = np.zeros_like(candidates)
if not candidates.any():
return selected
if surface_tri is None:
return candidates.copy()
driven = np.asarray(driven_full, dtype=np.int64)
priority = np.asarray(priority, dtype=np.float64)
candidate_axes = np.flatnonzero(candidates)
axis_by_vertex = {
int(driven[axis]): int(axis) for axis in candidate_axes
}
neighbors = {vertex: set() for vertex in axis_by_vertex}
for face in np.asarray(surface_tri, dtype=np.int64).reshape(-1, 3):
present = [int(v) for v in face if int(v) in axis_by_vertex]
for i, vertex in enumerate(present):
neighbors[vertex].update(present[:i])
neighbors[vertex].update(present[i + 1:])
blocked = set()
order = sorted(
candidate_axes,
key=lambda axis: (-priority[axis], int(driven[axis])),
)
for axis in order:
vertex = int(driven[axis])
if vertex in blocked:
continue
selected[axis] = True
blocked.add(vertex)
blocked.update(neighbors[vertex])
return selected
class _SparseLinearMap:
"""Matrix-free ``A^-1 B`` backed by one sparse LU factorization."""
def __init__(self, matrix, rhs_map):
self._matrix = matrix.tocsc()
self._rhs_map = rhs_map.tocsr()
self._factor = None
self._factorize()
def _factorize(self):
import scipy.sparse.linalg as spla
self._factor = spla.splu(self._matrix)
def apply(self, values):
import numpy as np
if self._factor is None:
self._factorize()
rhs = np.asarray(self._rhs_map @ values, dtype=np.float64)
result = np.asarray(self._factor.solve(rhs), dtype=np.float64)
if not np.all(np.isfinite(result)):
raise ValueError("sparse pin diffusion produced non-finite values")
return result
def __getstate__(self):
# SuperLU itself cannot be pickled. Keeping the sparse system makes a
# scene serializable even during the short populate-to-configure
# window; apply_pin_config normally removes the map before snapshots.
return {"matrix": self._matrix, "rhs_map": self._rhs_map}
def __setstate__(self, state):
self._matrix = state["matrix"]
self._rhs_map = state["rhs_map"]
self._factor = None
def _apply_sparse_frame_map(linear_map, values):
"""Apply a sparse map to ``(frames, input, xyz)`` values in one solve."""
import numpy as np
values = np.asarray(values, dtype=np.float64)
if values.ndim != 3 or values.shape[2] != 3:
raise ValueError(
f"expected frame values with shape (frames, vertices, 3), got {values.shape}"
)
n_frames, n_input, _ = values.shape
rhs = values.transpose(1, 0, 2).reshape(n_input, n_frames * 3)
mapped = linear_map.apply(rhs)
return mapped.reshape(mapped.shape[0], n_frames, 3).transpose(1, 0, 2)
def _harmonic_interior_operator_strict(n_verts, tets, surf_ids, interior_ids):
"""Sparse map from surface values to a tet mesh's interior values.
Solves the discrete Laplace equation with the surface vertices held as
Dirichlet boundary conditions: partition the graph Laplacian
``L = D - A`` (``A`` = tet-edge adjacency) into interior (``I``) and
surface (``S``) blocks and solve ``L_II u_I = -L_IS u_S``. The returned
map factors ``L_II`` once and applies only the right-hand sides the caller
needs. It never materializes the dense ``-L_II^-1 L_IS`` operator.
Each interior value is a convex combination of the surface values, so a
boundary field inside [0, 1] extends to an interior field inside [0, 1].
Raises:
RuntimeError: SciPy is unavailable, or ``L_II`` is singular because an
interior component has no edge path to the surface.
ValueError: the tet array is not ``(n, 4)``, or the surface or interior
index set is empty.
"""
try:
import warnings
import numpy as np
import scipy.sparse.linalg as spla
except Exception as exc:
raise RuntimeError(
"SciPy is required to extend a field into a tetrahedral mesh's "
"interior. Install the frontend dependencies with warmup.py "
"(warmup.bat on Windows)."
) from exc
T = np.asarray(tets, dtype=np.int64)
if T.ndim != 2 or T.shape[1] != 4:
raise ValueError(
f"expected a (n, 4) tet array, got shape {tuple(T.shape)}"
)
inter = np.asarray(interior_ids, dtype=np.int64)
surf = np.asarray(surf_ids, dtype=np.int64)
if surf.size == 0:
raise ValueError("the tet mesh has no surface vertices to extend from")
if inter.size == 0:
raise ValueError("the tet mesh has no interior vertices to extend into")
pairs = ((0, 1), (0, 2), (0, 3), (1, 2), (1, 3), (2, 3))
e0 = np.concatenate([T[:, a] for a, _ in pairs])
e1 = np.concatenate([T[:, b] for _, b in pairs])
L = _graph_laplacian_from_edges(e0, e1, n_verts)
L_II = L[inter][:, inter].tocsc()
L_IS = L[inter][:, surf].tocsc()
with warnings.catch_warnings():
warnings.simplefilter("ignore", spla.MatrixRankWarning)
try:
return _SparseLinearMap(L_II, -L_IS)
except Exception as exc:
raise RuntimeError(
"an interior region of this tetrahedral mesh has no edge path "
"to its surface, so no boundary value determines it"
) from exc
def _build_harmonic_interior_operator(n_verts, tets, surf_ids, interior_ids):
"""`_harmonic_interior_operator_strict`, or None when it cannot be built.
The pin paths degrade to a surface-only pin set, which is the behavior
every partial-pin SOLID scene was authored against. A material value has no
correct degraded answer, so the map transfer calls the strict builder.
"""
try:
return _harmonic_interior_operator_strict(
n_verts, tets, surf_ids, interior_ids
)
except Exception:
_warn_if_scipy_missing("_build_harmonic_interior_operator")
return None
def _surface_graph_laplacian(tris, n_surf):
"""Graph Laplacian L = D - A over the edges of a triangle mesh (compact
surface index space). Binary adjacency, robust to fTetWild slivers (a
cotangent Laplacian can go non-finite / negative on degenerate tris)."""
import numpy as np
T = np.asarray(tris, dtype=np.int64).reshape(-1, 3)
e0 = np.concatenate([T[:, 0], T[:, 1], T[:, 2]])
e1 = np.concatenate([T[:, 1], T[:, 2], T[:, 0]])
return _graph_laplacian_from_edges(e0, e1, n_surf)
class _SolidWeightTransfer:
"""Carries a per-Blender-vertex weight field onto a tet mesh's vertices.
Two stages, both convex combinations of painted values:
* every tet SURFACE vertex takes the weights of the Blender triangle
closest to it, combined with clipped and renormalized frame
coefficients;
* every tet INTERIOR vertex takes the Dirichlet Laplace extension of the
tet surface values.
A weight field inside [0, 1] therefore reaches the tet vertices inside
[0, 1] in exact arithmetic. `tolerance` is the extension operator's own
measured partition-of-unity residual, which bounds the rounding the second
stage can introduce; a value outside the interval by more than that is a
defect and raises rather than being clipped out of sight.
"""
def __init__(
self,
object_name,
n_blender,
n_tet,
surf_ids,
interior_ids,
surface_map,
interior_map,
tolerance,
):
self.object_name = object_name
self.n_blender = int(n_blender)
self.n_tet = int(n_tet)
self._surf_ids = surf_ids
self._interior_ids = interior_ids
self._surface_map = surface_map
self._interior_map = interior_map
self.tolerance = float(tolerance)
def apply(self, weights, object_name=None):
"""`weights` per Blender vertex mapped to one value per tet vertex.
`object_name` names the CALLER, because one transfer is shared by every
instance of a canonical tet mesh and the name it was built from is
whichever instance came first.
Accepts a single field of shape ``(n_blender,)`` or a stack of them of
shape ``(n_blender, k)``, which shares the one factorization across
every map on the object.
"""
import numpy as np
w = np.asarray(weights, dtype=np.float64)
one_field = w.ndim == 1
if one_field:
w = w.reshape(-1, 1)
named = object_name or self.object_name
if w.ndim != 2 or w.shape[0] != self.n_blender:
raise ValueError(
f"a material map on '{named}' has "
f"{w.shape[0]} weights but its Blender mesh has "
f"{self.n_blender} vertices"
)
out = np.zeros((self.n_tet, w.shape[1]), dtype=np.float64)
surface = np.asarray(self._surface_map @ w, dtype=np.float64)
out[self._surf_ids] = surface
if self._interior_map is not None:
out[self._interior_ids] = self._interior_map.apply(surface)
lo, hi = float(out.min()), float(out.max())
if lo < -self.tolerance or hi > 1.0 + self.tolerance:
raise ValueError(
f"carrying a material map onto the tetrahedra of "
f"'{named}' produced weights in [{lo}, {hi}], "
f"outside [0, 1] by more than the extension operator's "
f"residual of {self.tolerance}"
)
np.clip(out, 0.0, 1.0, out=out)
return out.ravel() if one_field else out
def _build_solid_weight_transfer(tet_mesh, F_arr, object_name):
"""The `_SolidWeightTransfer` for one tetrahedralized object.
Raises rather than degrading: a pin set has a defensible surface-only
reading, and a material value does not.
"""
import numpy as np
import scipy.sparse as sp
from ._bvh_ import frame_mapping
bl = getattr(tet_mesh, "_pin_blender_surface", None)
if bl is None:
raise RuntimeError(
f"object '{object_name}' carries no record of the Blender surface "
"its tetrahedra were built from, so a material map cannot be "
"carried onto them. Transfer the scene again."
)
bl_verts = np.ascontiguousarray(np.asarray(bl[0], dtype=np.float64))
bl_tris = np.ascontiguousarray(np.asarray(bl[1], dtype=np.int64)).reshape(-1, 3)
if bl_tris.shape[0] == 0:
raise RuntimeError(
f"object '{object_name}' has no Blender triangles, so there is no "
"surface to read its material map weights from"
)
V_local = np.ascontiguousarray(np.asarray(tet_mesh[0], dtype=np.float64))
n_tet = int(V_local.shape[0])
n_blender = int(bl_verts.shape[0])
sim_surf_ids = np.unique(np.asarray(F_arr, dtype=np.int64).reshape(-1))
n_surf = int(sim_surf_ids.size)
if n_surf == 0:
raise RuntimeError(
f"object '{object_name}' has no tetrahedral surface vertices to "
"carry a material map onto"
)
# Inverse direction: each tet surface vertex takes the Blender triangle
# closest to it. The forward map is not surjective onto tet surface
# vertices, so it would leave some of them with no painted value at all.
tri_idx, coefs = frame_mapping(V_local[sim_surf_ids], bl_verts, bl_tris)
tri_idx = np.asarray(tri_idx, dtype=np.int64)
if (tri_idx < 0).any():
raise RuntimeError(
f"{int((tri_idx < 0).sum())} tetrahedral surface vertices of "
f"'{object_name}' matched no Blender triangle"
)
c = np.asarray(coefs, dtype=np.float64).reshape(-1, 3)
bary = np.stack([1.0 - c[:, 0] - c[:, 1], c[:, 0], c[:, 1]], axis=1)
np.clip(bary, 0.0, None, out=bary)
total = bary.sum(axis=1)
# A degenerate target triangle returns zero coefficients, which names the
# triangle's first vertex. That is still one painted value, so the row
# stays a convex combination.
degenerate = total <= 1e-12
bary[degenerate] = (1.0, 0.0, 0.0)
total[degenerate] = 1.0
bary /= total[:, None]
cols = bl_tris[tri_idx].reshape(-1)
rows = np.repeat(np.arange(n_surf, dtype=np.int64), 3)
surface_map = sp.coo_matrix(
(bary.reshape(-1), (rows, cols)), shape=(n_surf, n_blender)
).tocsr()
interior_ids = np.setdiff1d(np.arange(n_tet, dtype=np.int64), sim_surf_ids)
interior_map = None
residual = 0.0
if interior_ids.size:
interior_map = _harmonic_interior_operator_strict(
n_tet, tet_mesh[2], sim_surf_ids, interior_ids
)
# The extension is exact on a constant field, so what it returns for
# all-ones measures the rounding it introduces on any field.
ones = interior_map.apply(np.ones(n_surf, dtype=np.float64))
residual = float(np.max(np.abs(ones - 1.0)))
tolerance = max(residual, float(np.finfo(np.float32).eps))
return _SolidWeightTransfer(
object_name,
n_blender,
n_tet,
sim_surf_ids,
interior_ids,
surface_map,
interior_map,
tolerance,
)
def _build_solid_pin_fields(
tet_mesh, F_arr, pin_index, alpha_rel=0.1, progress_callback=None,
):
"""Two-stage diffusion for a PARTIALLY-pinned SOLID. Returns a dict with
a per-tet-vertex pull WEIGHT and a per-frame TARGET operator, or ``None``
on failure (caller falls back to surface-only).
Stage 1 (surface, least-squares Poisson): map each input (Blender)
vertex to its closest tet-surface triangle (barycentric), then solve
``(B^T W B + alpha L_s) u = B^T W d`` on the tet surface. The WEIGHT
field uses ``d = 1`` on pinned input / 0 elsewhere with ``W = I``; the
TARGET field uses ``d = captured target`` with ``W = diag(pinned)`` so
only pinned input constrains it. The target matrix is factorized once
and applied only to captured frame data.
Stage 2 (interior): graph-Laplace harmonic extension of the surface
field into the tet interior (reuses :func:`_build_harmonic_interior_operator`).
"""
try:
import warnings
import numpy as np
import scipy.sparse as sp
import scipy.sparse.linalg as spla
from ._bvh_ import frame_mapping
except Exception:
_warn_if_scipy_missing("_build_solid_pin_fields")
return None
try:
bl = getattr(tet_mesh, "_pin_blender_surface", None)
if bl is None:
return None
bl_verts = np.ascontiguousarray(np.asarray(bl[0], dtype=np.float64))
V_local = np.ascontiguousarray(np.asarray(tet_mesh[0], dtype=np.float64))
n_tet = V_local.shape[0]
sim_surf_ids = np.unique(F_arr.reshape(-1))
n_surf = int(sim_surf_ids.size)
n_input = int(bl_verts.shape[0])
if n_surf == 0 or n_input == 0:
return None
# Compact surface index space (dense re-index of sim_surf_ids).
full2cpt = np.full(n_tet, -1, dtype=np.int64)
full2cpt[sim_surf_ids] = np.arange(n_surf)
F_cpt = full2cpt[F_arr.reshape(-1, 3)]
if (F_cpt < 0).any():
return None
surf_verts = V_local[sim_surf_ids]
# Stage 1: closest tet-surface triangle per INPUT vertex (forward map).
tri_idx, coefs = frame_mapping(bl_verts, surf_verts, F_cpt)
rows = np.empty(3 * n_input, dtype=np.int64)
cols = np.empty(3 * n_input, dtype=np.int64)
data = np.empty(3 * n_input, dtype=np.float64)
for a in range(n_input):
tri = F_cpt[int(tri_idx[a])]
c = coefs[a]
w = np.array([1.0 - c[0] - c[1], c[0], c[1]], dtype=np.float64)
w = np.clip(w, 0.0, None)
s = w.sum()
w = (w / s) if s > 1e-12 else np.array([1.0, 0.0, 0.0])
for j in range(3):
rows[3 * a + j] = a
cols[3 * a + j] = int(tri[j])
data[3 * a + j] = w[j]
B = sp.coo_matrix((data, (rows, cols)), shape=(n_input, n_surf)).tocsr()
L_s = _surface_graph_laplacian(F_cpt, n_surf)
BtB = (B.T @ B).tocsc()
diag_btb = BtB.diagonal().mean() if BtB.nnz else 1.0
diag_ls = L_s.diagonal().mean() if L_s.nnz else 1.0
alpha = alpha_rel * diag_btb / max(diag_ls, 1e-12)
pin_mask = np.zeros(n_input, dtype=np.float64)
pinned = np.asarray(sorted({int(i) for i in pin_index}), dtype=np.int64)
pin_mask[pinned[(pinned >= 0) & (pinned < n_input)]] = 1.0
if progress_callback is not None:
progress_callback("Building partial SOLID surface pin map...")
with warnings.catch_warnings():
warnings.simplefilter("ignore", spla.MatrixRankWarning)
# Weight field (W = I): diffuse the binary pinned mask.
A_w = (BtB + alpha * L_s).tocsc()
w_surf = spla.spsolve(A_w, B.T @ pin_mask)
# Target operator (W = diag(pin_mask)): only pinned input
# constrains it; eps*I removes the constant nullspace on surface
# components with no pinned input vertex.
Wd = sp.diags(pin_mask)
BtW = (B.T @ Wd).tocsr()
A_t = (BtW @ B + alpha * L_s + 1e-8 * sp.eye(n_surf)).tocsc()
surface_map = _SparseLinearMap(A_t, BtW)
w_surf = np.clip(np.asarray(w_surf, dtype=np.float64), 0.0, 1.0)
if w_surf.shape[0] != n_surf:
return None
if not np.all(np.isfinite(w_surf)):
return None
# Stage 2: interior harmonic extension of weight + (per-frame) target.
surf_ids = [int(s) for s in sim_surf_ids]
surf_set = set(surf_ids)
interior_ids = [v for v in range(n_tet) if v not in surf_set]
interior_map = None
interior_w = None
if interior_ids:
if progress_callback is not None:
progress_callback("Building partial SOLID interior pin map...")
interior_map = _build_harmonic_interior_operator(
n_tet, tet_mesh[2], surf_ids, interior_ids,
)
if interior_map is not None:
try:
interior_w = np.clip(
interior_map.apply(w_surf), 0.0, 1.0
)
except ValueError:
interior_map = None
return {
"surf_ids": surf_ids,
"interior_ids": interior_ids if interior_map is not None else [],
"w_surf": w_surf,
"interior_w": interior_w,
"surface_map": surface_map,
"interior_map": interior_map,
"motion_cache": {},
"n_input": n_input,
}
except Exception:
return None
class SceneDecoder:
"""Decode ``data.pickle`` written by the Blender addon into scene objects.
Handles canonical mesh deduplication, per-instance transforms, cached
tetrahedralization for SOLID groups, pin registration with
Blender-to-sim surface mapping, UV assignment for SHELL groups, and
rod / static groups. Used internally by :class:`BlenderApp`.
Example:
Drive the decoder directly to populate a scene (this is what
:meth:`BlenderApp.populate` does internally)::
from frontend._asset_ import AssetManager
from frontend._mesh_ import MeshManager
from frontend._decoder_ import SceneDecoder
assets = AssetManager()
meshes = MeshManager("/tmp/cache")
decoder = SceneDecoder("/path/to/data.pickle", assets, meshes)
decoder.populate_objects(scene, verbose=True)
"""
def __init__(
self, filepath: str, asset_manager: AssetManager, mesh_manager: MeshManager
):
_rust.validate_pickle_extension(filepath)
# See ``frontend/_cbor_bridge_.py`` for the byte-sniff contract.
from . import _cbor_bridge_ as _cbor
self._data = _cbor.load_scene_file(filepath)
self._asset = asset_manager
self._mesh = mesh_manager
self._object_info: dict[str, ObjectInfo] = {} # uuid -> ObjectInfo for stitch generation
# What a SOLID needs to carry a painted map onto its tetrahedra, and
# the built transfers. Recorded per object, cached per canonical ASSET,
# so two instances of one tet mesh share a factorization.
self._solid_weight_inputs: dict[str, tuple] = {}
self._solid_weight_transfers: dict[str, object] = {}
@staticmethod
def _tetra_cache_name(tri_mesh, ftw_kwargs) -> str:
"""Cache filename component naming the tetrahedralization of
*tri_mesh* under *ftw_kwargs*.
Composed by the same Rust helper ``TriMesh.tetrahedralize`` calls,
so the plan reports on the file the build reads and writes. The
kwargs are stringified the way ``tetrahedralize`` stringifies its
own, since both sides receive the same decoded values. The decoder
drives ``tetrahedralize`` with keyword arguments only, so there are
no positional arguments to carry.
"""
pairs = [(str(k), str(v)) for k, v in (ftw_kwargs or {}).items()]
name, _cache_key = _rust.mesh_tetra_cache_key(tri_mesh.hash, [], pairs)
return name
def _summarize_tetra_jobs(self, tetra_jobs: list[dict]) -> str:
return _rust.summarize_tetra_jobs(tetra_jobs)
@staticmethod
def _apply_transform(local_vert, transform):
"""Apply a 4x4 transform to local-space vertices and return world-space vertices as float32."""
import numpy as np
mat = np.ascontiguousarray(np.asarray(transform, dtype=np.float64))
v = np.ascontiguousarray(np.asarray(local_vert, dtype=np.float64))
return _rust.apply_transform_4x4(v, mat)
def _resolve_object_mesh(self, obj, name, obj_uuid, mesh_ref, canonical_meshes):
"""Resolve local-space mesh data for an object during the second
pass. Returns ``(local_vert, face, edge, uv, vert)`` where
``vert`` is the world-space vertices used by ``ObjectInfo`` and
the pin-mapping pass.
"""
if mesh_ref and mesh_ref in canonical_meshes:
ref = canonical_meshes[mesh_ref]
local_vert = ref["vert"]
face = ref["face"]
edge = ref.get("edge")
uv = ref.get("uv")
elif "vert" in obj:
local_vert = obj["vert"] # Local if transform present, world if legacy
face = obj.get("face")
edge = obj.get("edge")
uv = obj.get("uv", None)
else:
_rust.validate_object_has_mesh(name, obj_uuid, False)
raise AssertionError("unreachable") # pragma: no cover
# World-space vertices (for object_info used by pin mapping etc.)
vert = obj.get("_resolved_vert", local_vert)
return local_vert, face, edge, uv, vert
@staticmethod
def _log_object_mesh(name, vert, face, edge, uv):
"""Verbose-mode per-object summary printed during the second pass."""
if edge is not None:
print(
f" * name: {name}, vert: {vert.shape}, edge: {edge.shape}, uv: {len(uv) if uv is not None else 'None'}"
)
else:
print(
f" * name: {name}, vert: {vert.shape}, face: {face.shape if face is not None else 'None'}, uv: {len(uv) if uv is not None else 'None'}"
)
def populate_objects(self, scene: Scene, verbose: bool = False, progress_callback=None, ftetwild_by_uuid: dict | None = None, soft_constraint_by_uuid: dict | None = None, stitch_endpoint_uuids: set | None = None, solver_fps: float | None = None, time_scale: float | None = None) -> Scene:
"""Populate ``scene`` with objects from the decoder's pickle data.
Handles STATIC, SOLID, SHELL, ROD, PDRD, and SAND groups, including canonical
mesh deduplication by UUID, per-instance transforms, cached
tetrahedralization (with per-UUID fTetWild overrides), pin
registration with Blender-to-sim surface mapping, and stitches.
Args:
scene (Scene): The scene to populate.
verbose (bool): Enable verbose logging.
progress_callback: Optional callable ``fn(progress: float, info: str)`` invoked during loading. ``progress`` is in ``[0.0, 1.0]``.
ftetwild_by_uuid (dict | None): Optional mapping of object UUID to fTetWild keyword arguments used during tetrahedralization.
soft_constraint_by_uuid (dict | None): Optional mapping of STATIC object UUID to the spring stiffness holding its vertices to their animated positions. A UUID absent from the mapping keeps exact (Dirichlet) pins.
Returns:
Scene: The populated scene.
Example:
Populate a fresh scene from a Blender pickle::
from frontend._asset_ import AssetManager
from frontend._mesh_ import MeshManager
from frontend._decoder_ import SceneDecoder
decoder = SceneDecoder(
"/path/to/data.pickle", AssetManager(), MeshManager("/tmp/cache"),
)
decoder.populate_objects(scene, verbose=True)
fixed_scene = scene.build()
"""
plan = self._plan_build(progress_callback, ftetwild_by_uuid)
object_entries = plan["object_entries"]
canonical_meshes = plan["canonical_meshes"]
canonical_asset_name = plan["canonical_asset_name"]
tetra_jobs = plan["tetra_jobs"]
progress = plan["progress"]
report = progress["report"]
report(f"Build plan: {self._summarize_tetra_jobs(tetra_jobs)}")
progress["completed"] += 0.5
object_iter = iter(object_entries)
for group in self._data:
objects = group.get("object", None)
assert objects is not None, "Object data not found in the group."
group_type = group.get("type")
if verbose:
print(f"--- new group: {group_type} ---")
for obj in objects:
entry = next(object_iter)
name = obj.get("name", "")
obj_uuid = obj.get("uuid", "")
_rust.validate_scene_object_identity(name, obj_uuid)
mesh_ref = obj.get("_mesh_ref_resolved")
transform = obj.get("transform")
local_vert, face, edge, uv, vert = self._resolve_object_mesh(
obj, name, obj_uuid, mesh_ref, canonical_meshes,
)
report(f"Loading {group_type}: {name}...")
if verbose:
self._log_object_mesh(name, vert, face, edge, uv)
tet_mesh = None
V = None
F = None
if group_type == "STATIC":
_obj = self._populate_static(
scene, obj, name, obj_uuid, local_vert, face, transform, vert,
is_stitch_endpoint=bool(
stitch_endpoint_uuids and obj_uuid in stitch_endpoint_uuids
),
soft_stiffness=(
(soft_constraint_by_uuid or {}).get(obj_uuid)
),
verbose=verbose,
solver_fps=solver_fps,
time_scale=time_scale,
)
elif group_type == "SOLID":
_obj, tet_mesh, V, F = self._populate_solid(
scene, obj, entry, name, obj_uuid, vert,
object_entries, tetra_jobs, ftetwild_by_uuid, progress,
)
# Recorded here rather than inside `_populate_solid`, whose
# Blender-surface block is skipped for an object reusing an
# earlier object's tetrahedralization.
reuse_from = entry.get("tetra_reuse_from")
self._solid_weight_inputs[obj_uuid] = (
object_entries[reuse_from]["name"]
if reuse_from is not None
else name,
tet_mesh,
F,
name,
)
elif group_type == "SHELL":
_obj = self._populate_shell(
scene, obj_uuid, mesh_ref, canonical_meshes,
canonical_asset_name, name, local_vert, face, transform, vert, uv,
)
elif group_type == "ROD":
_obj = self._populate_rod(
scene, name, obj_uuid, local_vert, edge, transform, vert,
)
elif group_type == "PDRD":
_obj = self._populate_pdrd(
scene, obj_uuid, mesh_ref, canonical_meshes,
canonical_asset_name, name, local_vert, face, transform, vert,
)
elif group_type == "SAND":
_obj = self._populate_sand(
scene, name, obj_uuid, local_vert, transform, vert,
)
else:
_rust.validate_group_type(group_type)
_obj = None
if _obj is not None:
_obj._statistics_uuid = obj_uuid
_obj._statistics_name = name
_obj._statistics_type = group_type
# Per-object bending reference rest angle (SHELL): the scene
# object carries a `bend_rest_vert` local buffer aligned 1:1
# with its own vertices. Transform it to world space (matching
# `vert`) and hand it to the scene object so the assembler can
# scatter it into the concatenated reference-vertex array.
if _obj is not None and obj.get("bend_rest_vert") is not None:
import numpy as np
brv = obj["bend_rest_vert"]
brv_world = (
self._apply_transform(brv, transform)
if transform is not None
else np.ascontiguousarray(brv, dtype=np.float32)
)
_obj.set_bend_rest_vert(brv_world)
self._apply_pin_mapping(
obj, _obj, group_type, vert, V, F, tet_mesh, verbose,
progress_callback=report,
)
self._apply_stitch(obj, _obj, name, verbose)
progress["completed"] += entry["base_weight"]
report(f"Loaded {group_type}: {name}")
return scene
def _plan_build(self, progress_callback, ftetwild_by_uuid=None):
"""First pass: build canonical mesh table, resolve mesh refs, and
size the tetra-job list. Returns a dict with ``object_entries``,
``canonical_meshes``, ``canonical_asset_name``, ``tetra_jobs``,
and a ``progress`` sub-dict carrying mutable counters plus the
``report`` callable used by the second pass.
``ftetwild_by_uuid`` carries the same per-object tetrahedralizer
settings the second pass hands to ``tetrahedralize``. The plan
needs them because they are part of the cache key: an object's
cache file is named for its mesh and its settings together, so
both the "cached / new" label and the dedup decision are wrong
without them.
"""
planning_steps = sum(len(group.get("object", [])) for group in self._data)
planning_weight = max(1.0, 0.35 * planning_steps)
# Preallocate ``object_entries`` to ``planning_steps`` slots so
# the planning loop fills slots by index instead of growing the
# list one ``append`` at a time. Each slot is later overwritten
# with a fully-populated entry dict.
object_entries: list = [None] * planning_steps
# The ``progress`` dict carries mutable counters across the
# planning loop, the per-group dispatchers, and ``report``.
progress = {"completed": 0.0, "total": 0.0}
object_work = 0.0
# First-pass tetra-job counter; the dedup-aware rebuild below
# is what actually feeds ``_summarize_tetra_jobs`` so this just
# stamps a provisional ``tetra_index``.
tetra_idx_first_pass = 0
entry_idx = 0
# Mesh deduplication: canonical mesh data keyed by UUID
canonical_meshes: dict = {}
def report(info: str, total_work_override: float | None = None):
if progress_callback is not None:
total = progress["total"] if total_work_override is None else total_work_override
progress_callback(
progress["completed"] / total if total > 0 else 1.0,
info,
)
progress["report"] = report
# First pass: resolve mesh_ref and build canonical mesh table.
# canonical_meshes is keyed by UUID (encoder sends mesh_ref as UUID).
# Also keep asset_name (Blender name) for asset registration.
canonical_asset_name: dict = {}
for group in self._data:
objects = group.get("object", None)
assert objects is not None, "Object data not found in the group."
for obj in objects:
name = obj.get("name", "")
obj_uuid = obj.get("uuid", "")
_rust.validate_scene_object_identity(name, obj_uuid)
if "mesh_ref" not in obj and "vert" in obj:
# Canonical mesh: store local data for referenced instances
canonical_meshes[obj_uuid] = {
"vert": obj.get("vert"),
"face": obj.get("face"),
"edge": obj.get("edge"),
"uv": obj.get("uv", None),
"stitch": obj.get("stitch", None),
}
canonical_asset_name[obj_uuid] = name
planning_increment = planning_weight / planning_steps if planning_steps > 0 else 0.0
report("Scanning build plan...", planning_weight)
for group in self._data:
objects = group.get("object", None)
assert objects is not None, "Object data not found in the group."
group_type = group.get("type")
for obj in objects:
name = obj.get("name")
obj_uuid = obj.get("uuid", "")
report(
f"Scanning build plan: {name}...",
planning_weight,
)
# Resolve mesh data: either from obj directly or via mesh_ref
mesh_ref = obj.get("mesh_ref")
transform = obj.get("transform")
if mesh_ref is not None:
_rust.validate_mesh_ref_known(
name, str(mesh_ref), mesh_ref in canonical_meshes
)
if mesh_ref is not None and mesh_ref in canonical_meshes:
# Duplicate: get local mesh from canonical, apply transform
ref = canonical_meshes[mesh_ref]
local_vert = ref["vert"]
face = ref["face"]
vert = self._apply_transform(local_vert, transform) if transform is not None else local_vert
obj["_resolved_vert"] = vert
obj["_resolved_face"] = face
obj["_resolved_uv"] = ref.get("uv")
obj["_resolved_stitch"] = ref.get("stitch")
obj["_resolved_edge"] = ref.get("edge")
obj["_mesh_ref_resolved"] = mesh_ref
elif transform is not None and "vert" in obj:
# Canonical with transform: apply transform to local vertices
vert = self._apply_transform(obj["vert"], transform)
obj["_resolved_vert"] = vert
obj["_resolved_face"] = obj.get("face")
obj["_resolved_uv"] = obj.get("uv")
obj["_resolved_stitch"] = obj.get("stitch")
obj["_resolved_edge"] = obj.get("edge")
obj["_mesh_ref_resolved"] = None
else:
# Legacy format: vertices already in world space
obj["_resolved_vert"] = obj.get("vert")
obj["_resolved_face"] = obj.get("face")
obj["_resolved_uv"] = obj.get("uv")
obj["_resolved_stitch"] = obj.get("stitch")
obj["_resolved_edge"] = obj.get("edge")
obj["_mesh_ref_resolved"] = None
entry = {
"group_type": group_type,
"obj": obj,
"name": name,
"base_weight": 1.0,
"tetra_weight": 0.0,
"tetra_index": None,
"tetra_cached": False,
"tetra_cache_name": None,
"tri_mesh": None,
}
if group_type == "SOLID":
report(
f"Checking tetra cache: {name}...",
planning_weight,
)
# For dedup: use LOCAL vertices for tet (same hash for duplicates)
if mesh_ref and mesh_ref in canonical_meshes:
tet_vert = canonical_meshes[mesh_ref]["vert"]
tet_face = canonical_meshes[mesh_ref]["face"]
elif "vert" in obj and obj.get("transform") is not None:
tet_vert = obj["vert"] # Local-space
tet_face = obj.get("face")
else:
tet_vert = obj["_resolved_vert"] # Legacy: world-space
tet_face = obj["_resolved_face"]
tri_mesh = self._mesh.create.tri(tet_vert, tet_face)
tetra_cache_name = self._tetra_cache_name(
tri_mesh, (ftetwild_by_uuid or {}).get(obj_uuid)
)
try:
cached = _cache_probe(tri_mesh.cache_path(tetra_cache_name))
except CachePathUnusableError as exc:
# The plan's only products are a progress weight and
# a label, so an unreadable cache location is
# reported and the object is planned as uncached.
# The build reaches the same path moments later
# through ``tetrahedralize``, which raises there.
report(
f"Cannot read the tetra cache for {name}: {exc}",
planning_weight,
)
cached = False
entry["tri_mesh"] = tri_mesh
entry["tetra_cached"] = cached
entry["tetra_cache_name"] = tetra_cache_name
tetra_idx_first_pass += 1
entry["tetra_index"] = tetra_idx_first_pass
entry["tetra_weight"] = 3.0 if cached else 8.0
object_work += entry["tetra_weight"]
object_entries[entry_idx] = entry
entry_idx += 1
object_work += entry["base_weight"]
progress["completed"] += planning_increment
progress["total"] = planning_weight + 0.5 + object_work
# Dedup-by-cache-key + tetra_jobs rebuild + per-entry tetra_index
# reassignment all run in a single Rust pass. Two SOLIDs share a
# tetrahedralization exactly when they address the same cache
# file, so the key is each entry's ``tetra_cache_name``. The Rust
# call returns the rebuilt jobs list and the work_delta that gets
# folded into ``object_work``.
tetra_jobs, _work_delta = _rust.dedup_and_rebuild_tetra_jobs(object_entries)
object_work += _work_delta
progress["total"] = planning_weight + 0.5 + object_work
return {
"object_entries": object_entries,
"canonical_meshes": canonical_meshes,
"canonical_asset_name": canonical_asset_name,
"tetra_jobs": tetra_jobs,
"progress": progress,
}
def _populate_static(self, scene, obj, name, obj_uuid, local_vert, face, transform, vert, is_stitch_endpoint=False, soft_stiffness=None, verbose=False, solver_fps=None, time_scale=None):
"""STATIC group dispatcher: rest-pose mesh, transform-keyframe
animation, UI-assigned static ops, or per-vertex deformation
cache. Returns the Scene Object for downstream pin / stitch
passes, or ``None`` for a non-stitched rest-pose collider (no
further pin work needed).
``vert`` is the world-space surface vertices recorded in the
``ObjectInfo`` so a STATIC can be a cross-stitch endpoint (it is
SHELL-like: 1:1 indices, never re-projected). ``is_stitch_endpoint``
marks a STATIC that appears in a cross-stitch: a non-moving such
STATIC is PROMOTED into the dynamic all-pinned namespace (instead of
the disjoint collision-mesh pool) so the stitch index can reach its
surface, while immovable fixed pins keep it frozen at rest.
"""
import numpy as np
transform_anim = obj.get("transform_animation", None)
static_ops = obj.get("static_ops", []) or []
static_deform = obj.get("static_deform_animation", None)
_rust.validate_static_anim_xor_ops(
name,
transform_anim is not None,
bool(static_ops),
static_deform is not None,
)
if (transform_anim is not None or static_ops
or static_deform is not None) and (
solver_fps is None or time_scale is None):
# The wire carries frame offsets and raw animation rates; only
# the Param payload can supply the time base. Fail loud rather
# than guessing a rate.
raise RuntimeError(
f"STATIC '{name}' carries animation; decoding requires the "
"Param payload fps/time_scale (param.pickle). A standalone "
"SceneDecoder cannot decode animated STATICs."
)
if verbose:
print(
f" > transform_animation: "
f"{'YES' if transform_anim else 'NO'}, "
f"static_ops: {len(static_ops)}, "
f"static_deform_animation: "
f"{'YES' if static_deform else 'NO'}"
)
def _setup_pin_shell():
"""Common setup for animated static objects: a
zero-stiffness shell whose vertices are driven by
pin operations. Returns the Object plus its
rest-frame translation (``(obj, rest_t)``).
"""
self._asset.add.tri(name, local_vert, face)
_o = scene.add(name, obj_uuid)
if transform is not None:
_o.mat4x4(transform)
_o.param.set("density", 0.0)
_o.param.set("young-mod", 0.0)
_o.param.set("poiss-rat", 0.0)
_o.param.set("bend", 0.0)
# Mark for preview suppression: the shell's pins
# are implementation detail, not user-chosen.
_o._is_static_moving = True
# Register stitch metadata so a STATIC pin-shell can be a
# cross-stitch endpoint. type="STATIC" (not "SOLID") tells the
# decoder to keep its barycentric slots verbatim: a STATIC is
# never re-tetrahedralized, so its surface indices map 1:1 to
# the solver (SHELL-like). Harmless for non-stitched statics
# (the decoder only looks up endpoints named in a stitch).
self._object_info[obj_uuid] = ObjectInfo(
type="STATIC", vert=vert, V=vert, F=face,
)
rest_t = (
np.asarray(transform, dtype=np.float64)[:3, 3]
if transform is not None
else np.zeros(3, dtype=np.float64)
)
return _o, rest_t
def _driven_pin(_o):
"""Pin holder for a collider that moves. Every one of its vertices is
prescribed by the user's animation, and an ordinary (non-pull) pin is
now an exact Dirichlet boundary condition, so no marking is needed:
the solver eliminates these DOF and the collider tracks its keyframes
exactly instead of being pushed off them by contact.
With soft constraints the same vertices are held by Hookean springs
instead: the DOF stay in the system and the pin contributes
``force = k * (target - x)`` with Hessian ``k * I``, so contact can
push the collider off its animated path where it pushes harder than
``k``. That is what a group asks for when its geometry closes onto
cloth harder than the cloth can escape.
"""
_p = _o.pin()
if soft_stiffness is not None:
_k = float(soft_stiffness)
if not _k > 0.0:
# `pull_w == 0` is the solver's fix/pull discriminator, so
# this would hand back an exact pin under a soft label.
raise ValueError(
f"STATIC '{name}' soft-constraint stiffness {_k} must "
"be strictly positive; zero is how the solver spells "
"an exact pin."
)
_p.pull(_k)
return _p
if transform_anim is not None:
# Case 1: Blender keyframes drive the pose. The simulator
# enforces the pin as a soft constraint, so its output
# vertices drift slightly from the input keyframes. The
# dynamics were resolved against the drifted positions, so
# we must display those โ not the keyframes โ to keep the
# collider visually consistent with the cloth. Include in
# output PC2.
_obj, rest_translation = _setup_pin_shell()
_driven_pin(_obj).transform_keyframes(
local_vert=local_vert,
# Frame offsets on the wire; seconds derived from the Param
# payload's fps (the Time-Scaled solver rate).
times=[float(o) / solver_fps
for o in transform_anim["frame_offset"]],
translations=transform_anim["translation"],
quaternions=transform_anim["quaternion"],
scales=transform_anim["scale"],
segments=transform_anim.get("segments", []),
rest_translation=rest_translation,
)
return _obj
if static_ops:
# Case 2: UI-assigned move/spin/scale ops. Blender has no
# fcurves, so the remote sim is the source of truth (include
# in output so PC2 can play it back).
_obj, _ = _setup_pin_shell()
pin = _driven_pin(_obj)
for op in static_ops:
# Frame offsets on the wire -> seconds via the Param fps.
t_start = float(op["frame_offset_start"]) / solver_fps
t_end = float(op["frame_offset_end"]) / solver_fps
transition = op.get("transition", "linear")
if op["op_type"] == "MOVE_BY":
pin.move_by(
list(op["delta"]),
t_start=t_start,
t_end=t_end,
transition=transition,
)
elif op["op_type"] == "SPIN":
# Center is always the object origin: in the
# pin-shell's local op frame that's (0,0,0).
pin.spin(
center=[0.0, 0.0, 0.0],
axis=list(op["axis"]),
# Wire carries the RAW authored degrees per
# ANIMATION second; time_scale converts to the
# solver rate (deg per solver second).
angular_velocity=(
float(op["angular_velocity_anim"]) * time_scale),
t_start=t_start,
t_end=t_end,
center_mode="absolute",
)
elif op["op_type"] == "SCALE":
pin.scale(
scale=float(op["factor"]),
t_start=t_start,
t_end=t_end,
center=[0.0, 0.0, 0.0],
center_mode="absolute",
)
else:
_rust.validate_static_op_type(op["op_type"])
return _obj
if static_deform is not None:
# Case 3: per-frame depsgraph-baked vertex stream from the
# Capture Deformation operator. local_vert is already
# frame_start's depsgraph-evaluated mesh in solver world
# space (the encoder swapped it in), and transform is
# identity. The pin shell starts at vert_frames[0] and we
# add one MoveByOperation per consecutive frame pair to
# drive every vertex through the recorded trajectory.
#
# The pin is a soft constraint, so the simulator output
# drifts from the captured cache. The cloth was resolved
# against the drifted positions, so include the shell in
# output PC2 and let MESH_CACHE overwrite the depsgraph-
# driven mesh on display.
vert_frames = np.ascontiguousarray(
static_deform["vert_frames"], dtype=np.float64,
)
if vert_frames.ndim != 3 or vert_frames.shape[2] != 3:
raise ValueError(
f"static_deform_animation['vert_frames'] for "
f"'{name}' must be (n_frames, n_verts, 3); got "
f"shape {vert_frames.shape}"
)
n_frames = vert_frames.shape[0]
n_verts = vert_frames.shape[1]
# Row i IS frame offset i (no time array on the wire), so the
# row times derive from the Param fps by construction and this
# channel cannot desync from the rest of the schedule.
times = [k / solver_fps for k in range(n_frames)]
if n_verts != len(local_vert):
raise ValueError(
f"static_deform_animation for '{name}': "
f"cache has {n_verts} vertices but mesh has "
f"{len(local_vert)}"
)
_obj, _ = _setup_pin_shell()
pin = _driven_pin(_obj)
# Successive MoveBy segments compose: at t=times[k],
# pin pos = local_vert + sum(deltas up to k) = vert_frames[k]
# (because the encoder set local_vert == vert_frames[0]).
for k in range(n_frames - 1):
delta = np.ascontiguousarray(
vert_frames[k + 1] - vert_frames[k], dtype=np.float64,
)
pin.move_by(
delta,
t_start=float(times[k]),
t_end=float(times[k + 1]),
transition="linear",
)
return _obj
# Case 4: rest-pose static (never moves).
if is_stitch_endpoint or soft_stiffness is not None:
# A soft constraint promotes for the same structural reason a
# cross-stitch endpoint does: the thing being softened is the pin,
# and a non-promoted rest-pose static has none (it is a disjoint
# contact-only collision mesh, never solved). Leaving it in that
# pool would make the checkbox a silent no-op. Promotion costs a
# solved body, which is the price of a collider that can yield.
# Promote into the dynamic all-pinned namespace so the cross-stitch
# index can address this collider's surface. It is built as the
# same zero-stiffness pin-shell the animated cases use, with EVERY
# vertex held by an immovable fixed pin (an all-vertex pin carrying
# no operations) so it stays kinematically frozen at its rest pose:
# a fixed vertex is prescribed exactly (not softly), so there is no
# drift. The _force_dynamic flag routes it into dyn_objects at
# build (an all-pinned, no-op object would otherwise classify as
# static via scene_all_vertices_pinned and fall back to the
# unreachable collision-mesh pool). ObjectInfo is registered by
# _setup_pin_shell; positive per-vertex mass and the density/young
# the solver asserts come from apply_to_objects' clear_all() tri
# defaults at make() time (the STATIC encoder prunes those keys).
_obj, _ = _setup_pin_shell()
# Carries no operations, so its pins never move: every vertex is
# held at its rest pose, exactly when the pin is a fix and by a
# spring of the group's stiffness when soft constraints are on.
_driven_pin(_obj)
_obj._force_dynamic = True
# A promoted STATIC is a cross-stitch TARGET (a collider), never a
# CIPC stitch SOURCE. The encoder auto-detects intra-mesh
# loose-edge stitches for any mesh (mesh.py:detect_stitch_edges),
# so suppress that here: now that this object is returned non-None
# (previously a rest-pose static returned None), the post-dispatch
# _apply_stitch would otherwise attach a stitch between this
# collider's own (all-fixed) vertices. The cross-stitch itself is
# unaffected (it flows through result["cross_stitch"], not obj).
if obj.get("_resolved_stitch", obj.get("stitch")) is not None and verbose:
print(f" > suppressing intra-mesh stitch on promoted static {name}")
obj["_resolved_stitch"] = None
obj.pop("stitch", None)
return _obj
# Non-stitched rest-pose static: a disjoint contact-only collision
# mesh (cheap, never solved). No ObjectInfo / promotion needed.
self._asset.add.tri(name, local_vert, face)
_static_obj = scene.add(name, obj_uuid)
_static_obj._statistics_uuid = obj_uuid
_static_obj._statistics_name = name
_static_obj._statistics_type = "STATIC"
if transform is not None:
_static_obj.mat4x4(transform)
_static_obj.pin()
return None
def solid_weight_transfer(self, obj_uuid: str):
"""The `_SolidWeightTransfer` for one tetrahedralized object.
Built on first use and cached by canonical ASSET name, so two
instances of one tetrahedral mesh share a single factorization.
"""
inputs = self._solid_weight_inputs.get(obj_uuid)
if inputs is None:
raise RuntimeError(
f"object {obj_uuid!r} carries a spatial material map but was "
"not recorded as a tetrahedralized object during populate"
)
asset_name, tet_mesh, F_arr, name = inputs
transfer = self._solid_weight_transfers.get(asset_name)
if transfer is None:
transfer = _build_solid_weight_transfer(tet_mesh, F_arr, name)
self._solid_weight_transfers[asset_name] = transfer
return transfer
def release_solid_weight_transfers(self) -> None:
"""Drop the built transfers and the inputs they were built from.
Each holds a sparse system and a SuperLU factor, which are wanted only
while the maps are being applied.
"""
self._solid_weight_inputs.clear()
self._solid_weight_transfers.clear()
def _populate_solid(
self,
scene,
obj,
entry,
name,
obj_uuid,
vert,
object_entries,
tetra_jobs,
ftetwild_by_uuid,
progress,
):
"""SOLID group dispatcher: tetrahedralize (or reuse a canonical
result), register the tet asset, add the scene instance, and
record an ``ObjectInfo`` entry. Returns ``(_obj, tet_mesh, V, F)``
for the pin-mapping pass.
"""
report = progress["report"]
reuse_from = entry.get("tetra_reuse_from")
transform = obj.get("transform")
if reuse_from is not None:
# Reuse tet + asset from canonical mesh
src = object_entries[reuse_from]
tet_mesh = src["_tet_mesh_result"]
asset_name = src["name"]
report(f"Reusing tetrahedralization from {asset_name} for {name}")
else:
prefix = (
f"Tetrahedralizing {name} "
f"({entry['tetra_index']}/{len(tetra_jobs)}, "
f"{'cached' if entry['tetra_cached'] else 'new'})"
)
report(f"{prefix}...")
ftw_kwargs = (
(ftetwild_by_uuid or {}).get(obj_uuid, {}) or {}
)
try:
tet_mesh = entry["tri_mesh"].tetrahedralize(
status_callback=lambda detail, prefix=prefix: report(
f"{prefix}: {detail}"
),
**ftw_kwargs,
)
except ValueError as e:
# Prepend the object name so the addon UI shows which
# SOLID mesh failed (the underlying message already
# explains the cause and suggests SHELL).
raise ValueError(f"{name}: {e}") from e
entry["_tet_mesh_result"] = tet_mesh
progress["completed"] += entry["tetra_weight"]
report(
f"Finished tetrahedralizing {name} "
f"({entry['tetra_index']}/{len(tetra_jobs)})"
)
asset_name = name
V_local, F, T = tet_mesh
# Register asset with local-space vertices
self._asset.add.tet(asset_name, V_local, F, T)
V_local, F, T = tet_mesh
# Add scene instance with per-instance transform
_obj = scene.add(asset_name, obj_uuid)
if transform is not None:
_obj.mat4x4(transform)
# Compute world-space V for object_info
if transform is not None:
V = self._apply_transform(V_local, transform)
else:
V = V_local
solid_info = ObjectInfo(type="SOLID", vert=vert, V=V, F=F)
if tet_mesh.has_surface_mapping():
import numpy as np
tri_indices, coefs = tet_mesh.surface_map
# Pass local-space tet vertices: the surface-map coefs were
# computed in local space, so the in-Rust ``x0 + c1*b1 + c2*b2 +
# c3*nฬ`` reconstruction (used to pick the closest of three
# triangle corners) only matches the original Blender position
# when V is in the same space as the coefs. Under non-uniform
# world scale, mixing world V with local coefs shifts the bp by
# tens of centimeters and can pick the wrong corner.
orig_to_sim = _rust.solid_orig_to_sim(
np.ascontiguousarray(np.asarray(tri_indices, dtype=np.int64)),
np.ascontiguousarray(np.asarray(coefs, dtype=np.float64)),
np.ascontiguousarray(np.asarray(F, dtype=np.int64)),
np.ascontiguousarray(np.asarray(V_local, dtype=np.float64)),
)
solid_info.orig_to_sim = orig_to_sim
scene.set_surface_map(obj_uuid, tri_indices, coefs, F)
# Stash the source Blender surface (local verts + triangles)
# that produced the surface map, so _apply_pin_mapping can build
# the INVERSE map (each sim surface vertex -> closest Blender
# triangle) and drive every sim surface vertex. The forward
# Blender->sim map is not surjective onto sim vertices, so using
# it alone leaves some sim surface vertices unpinned. Vertex
# indices in tri_mesh match the Blender pin index space (the
# forward surface map is already keyed by Blender vertex).
if not hasattr(tet_mesh, "_pin_blender_surface"):
src_tri = entry.get("tri_mesh")
if src_tri is None and reuse_from is not None:
src_tri = object_entries[reuse_from].get("tri_mesh")
if src_tri is not None:
tet_mesh._pin_blender_surface = (
np.ascontiguousarray(
np.asarray(src_tri[0], dtype=np.float64)
),
np.ascontiguousarray(
np.asarray(src_tri[1], dtype=np.int64)
),
)
self._object_info[obj_uuid] = solid_info
return _obj, tet_mesh, V, F
def _populate_shell(
self, scene, obj_uuid, mesh_ref, canonical_meshes, canonical_asset_name,
name, local_vert, face, transform, vert, uv,
):
"""SHELL group dispatcher: register (or reuse) the tri asset, add
the scene instance, and record an ``ObjectInfo`` entry. Returns
the Scene Object for the pin / stitch passes.
"""
if mesh_ref and mesh_ref in canonical_meshes:
asset_name = canonical_asset_name[mesh_ref]
else:
asset_name = name
self._asset.add.tri(asset_name, local_vert, face)
_obj = scene.add(asset_name, obj_uuid)
if transform is not None:
_obj.mat4x4(transform)
self._object_info[obj_uuid] = ObjectInfo(
type="SHELL", vert=vert, V=vert, F=face,
)
if uv is not None:
assert len(uv) == len(face), "UV length must match face length."
_obj.set_uv(uv)
return _obj
def _populate_pdrd(
self, scene, obj_uuid, mesh_ref, canonical_meshes, canonical_asset_name,
name, local_vert, face, transform, vert,
):
"""PDRD group dispatcher: register (or reuse) the tri asset (no
tetrahedralization), add the scene instance, flag it as PDRD,
and record an ``ObjectInfo`` entry. Returns the Scene Object
for the pin / stitch passes.
"""
if mesh_ref and mesh_ref in canonical_meshes:
asset_name = canonical_asset_name[mesh_ref]
else:
asset_name = name
self._asset.add.tri(asset_name, local_vert, face)
_obj = scene.add(asset_name, obj_uuid)
_obj.as_pdrd()
if transform is not None:
_obj.mat4x4(transform)
self._object_info[obj_uuid] = ObjectInfo(
type="PDRD", vert=vert, V=vert, F=face,
)
return _obj
def _populate_rod(self, scene, name, obj_uuid, local_vert, edge, transform, vert):
"""ROD group dispatcher: register the rod asset, add the scene
instance, and record an ``ObjectInfo`` entry. Returns the Scene
Object for the pin / stitch passes.
"""
_rust.validate_rod_has_edges(name, edge is not None)
self._asset.add.rod(name, local_vert, edge)
_obj = scene.add(name, obj_uuid)
if transform is not None:
_obj.mat4x4(transform)
self._object_info[obj_uuid] = ObjectInfo(type="ROD", vert=vert)
return _obj
def _populate_sand(self, scene, name, obj_uuid, local_vert, transform, vert):
"""SAND group dispatcher: register the positions-only points
asset (no connectivity), add the scene instance, and record an
``ObjectInfo`` entry. A SAND object is a faceless cloud of loose
vertices (one grain per vertex), so there is no surface/tri asset
and no PDRD flag. Returns the Scene Object for the pin / stitch
passes.
"""
self._asset.add.points(name, local_vert)
_obj = scene.add(name, obj_uuid)
if transform is not None:
_obj.mat4x4(transform)
self._object_info[obj_uuid] = ObjectInfo(type="SAND", vert=vert)
return _obj
def _apply_pin_mapping(
self, obj, _obj, group_type, vert, V, F, tet_mesh, verbose,
progress_callback=None,
):
"""Pin-mapping pass: register pins on ``_obj`` from ``obj['pin']``,
building the Blender-to-sim surface transfer for SOLID groups
with a surface mapping and falling back to direct per-index pins
otherwise.
"""
if _obj is None or "pin" not in obj:
return
pin_index = obj["pin"]
if verbose:
print(f" > pin: {len(pin_index)}")
if group_type == "SOLID" and tet_mesh is not None and tet_mesh.has_surface_mapping():
import numpy as np
from ._bvh_ import frame_mapping
# Drive sim surface vertices from the captured per-Blender-vertex
# deformation. fTetWild resamples the surface, so a Blender vertex
# does not coincide with a sim vertex. The forward map (Blender
# vertex -> its sim triangle) is NOT surjective onto sim vertices:
# some sim surface vertices fall in no pinned Blender vertex's
# triangle, stay unpinned, and float while neighbors follow the
# deformer -> distortion and solver self-intersection.
#
# Build the INVERSE map instead: for EACH sim surface vertex, its
# closest Blender surface triangle with in-plane barycentric
# weights (1-c1-c2, c1, c2; the normal offset c3 is irrelevant to
# a surface-displacement transfer). The sim vertex follows the
# weight-blended displacement of that triangle's PINNED corners.
# A sim vertex whose closest triangle has no pinned corner stays
# free, so partial and full pins use one code path (full pinning
# is just the case where every corner is pinned, hence every sim
# surface vertex is covered).
pinned = {int(i) for i in pin_index}
F_arr = np.ascontiguousarray(np.asarray(F, dtype=np.int64))
sim_surf_ids = np.unique(F_arr.reshape(-1))
bl = getattr(tet_mesh, "_pin_blender_surface", None)
n_blender = int(np.asarray(bl[0]).shape[0]) if bl is not None else 0
# Full pin (every Blender surface vertex pinned) keeps the
# verified path below unchanged. A PARTIAL pin instead diffuses
# a per-vertex pull weight + target via the two-stage Poisson so
# the pinned region follows while the rest stays free, with a
# graceful transition.
full_pin = bl is not None and n_blender > 0 and len(pinned) == n_blender
handled = False
if not full_pin and bl is not None:
fields = _build_solid_pin_fields(
tet_mesh,
F_arr,
pin_index,
progress_callback=progress_callback,
)
if fields is not None:
w_surf = fields["w_surf"]
interior_w = fields["interior_w"]
surf_ids = fields["surf_ids"]
interior_ids = fields["interior_ids"]
if interior_w is not None and interior_ids:
full_w = np.concatenate([w_surf, interior_w])
driven_full = surf_ids + interior_ids
else:
full_w = w_surf
driven_full = surf_ids
keep = full_w > _PIN_WEIGHT_EPS
driven = [int(driven_full[k]) for k in range(len(driven_full))
if keep[k]]
if driven:
holder = _obj.pin(driven)
holder._data._blender_pin_indices = list(pin_index)
holder._data._tet_V = V
holder._data._blender_vert = vert
# [0,1] per-driven-vertex weight; _apply_pin_cfg_entry
# scales it by pull_strength and exports pin-pullw.
holder._data._solid_pin_weights = (
full_w[keep].astype(np.float32)
)
holder._data._solid_pin = {
"surface_map": fields["surface_map"],
"interior_map": fields["interior_map"],
"motion_cache": fields["motion_cache"],
"keep": keep,
"n_input": fields["n_input"],
# True tet rest positions on the driven_full axis
# (surf_ids + interior_ids), in the SOLVER/world
# frame (V = transform @ V_local, the same space as
# the captured tracks and the solver rest verts; NOT
# tet_mesh[0], which is the untransformed local
# frame). The rigid-aware move-op builder carries
# THESE rigidly, not the S_t-diffused rest: the
# MoveBy deltas land on the solver's tet rest, so a
# rigid capture must rotate that exact rest or the
# rest/LS-fit mismatch reappears as a rotation-scaled
# bend.
"rest_full": np.asarray(
V, dtype=np.float64
)[driven_full],
}
# Full-axis arrays (length n_surf+n_interior) the
# fix_weight_threshold split needs. ``keep`` and
# ``_solid_pin_weights`` are the compacted view;
# _build_solid_poisson_move_ops slices positions[:, keep]
# over the FULL axis, so the split masks live there too.
holder._data._solid_full_w = full_w
holder._data._solid_driven_full = list(driven_full)
# Surface mask over the full axis. driven_full is
# surf_ids + interior_ids, so the first len(surf_ids)
# entries are the surface verts (solver tet-index <
# surface_vert_count). Hard FixPairs are kept surface
# only as the hard-core / soft-skirt authoring split;
# the old "interior fix pin is a zero-diagonal CG nan"
# reason no longer applies (a fix pin is an exact
# Dirichlet BC whose diagonal block is the identity).
# The threshold keeps interior high-weight verts as
# soft pull.
holder._data._solid_surf_mask = (
np.arange(len(driven_full)) < len(surf_ids)
)
holder._data._solid_surface_tri = F_arr
# Per-surface-vertex Blender corners (inverse map),
# aligned to surf_ids / the leading driven_full axis.
# The fix_weight_threshold split uses this to tell
# pull-intent surface verts from hard-intent ones: a
# pull pin overlapping a hard pin-root shares this one
# merged Poisson holder, and hardening its (often
# weight-1.0) verts would freeze them at initial
# geometry and drop the captured target. Mirrors the
# full-pin harmonic mixed-intent path.
try:
_blv = np.ascontiguousarray(
np.asarray(vert, dtype=np.float64)
)
_blt = np.ascontiguousarray(
np.asarray(bl[1], dtype=np.int64)
)
_Vloc = np.ascontiguousarray(
np.asarray(V, dtype=np.float64)
)
_itri, _icoef = frame_mapping(
_Vloc[sim_surf_ids], _blv, _blt
)
_sbw = []
for _k in range(sim_surf_ids.size):
_tri = _blt[int(_itri[_k])]
_c = _icoef[_k]
_w = (1.0 - _c[0] - _c[1], _c[0], _c[1])
_sbw.append([
(int(_tri[_j]), float(_w[_j]))
for _j in range(3)
if _w[_j] > _PIN_WEIGHT_EPS
and int(_tri[_j]) in pinned
])
holder._data._sim_blender_weights = _sbw
holder._data._solid_frame_map = {
"triangles": _blt[_itri],
"coefs": _icoef,
}
except Exception:
holder._data._sim_blender_weights = None
holder._data._solid_frame_map = None
handled = True
if verbose:
kw = full_w[keep]
print(f" > pin (poisson partial): "
f"{int(keep.sum())} driven tet verts from "
f"{len(pinned)} blender pins (weight "
f"[{kw.min():.2f},{kw.max():.2f}])")
if handled:
return
# sim vertex -> list of (blender_index, weight)
support: dict = {}
if bl is not None and sim_surf_ids.size:
bl_verts, bl_tris = bl[0], bl[1]
V_local = np.ascontiguousarray(
np.asarray(tet_mesh[0], dtype=np.float64)
)
inv_tri, inv_coefs = frame_mapping(
V_local[sim_surf_ids], bl_verts, bl_tris,
)
for k in range(sim_surf_ids.size):
tri = bl_tris[int(inv_tri[k])]
c = inv_coefs[k]
w = (1.0 - c[0] - c[1], c[0], c[1])
pairs = [
(int(tri[j]), float(w[j]))
for j in range(3)
if w[j] > _PIN_WEIGHT_EPS and int(tri[j]) in pinned
]
if not pairs:
# Projection landed outside the triangle / on a
# near-zero corner: fall back to the highest-weight
# pinned corner. No pinned corner -> sim vertex free.
cand = [
(int(tri[j]), float(w[j]))
for j in range(3)
if int(tri[j]) in pinned
]
if cand:
pairs = [max(cand, key=lambda t: t[1])]
if pairs:
support[int(sim_surf_ids[k])] = pairs
else:
# Defensive fallback (source surface unavailable): forward
# map. May under-cover the sim surface.
tri_indices_pin, coefs_pin = tet_mesh.surface_map
for i in pin_index:
ti = int(tri_indices_pin[int(i)])
tri = F_arr[ti]
c = coefs_pin[int(i)]
w = (1.0 - c[0] - c[1], c[0], c[1])
sv = [int(tri[j]) for j in range(3) if w[j] > _PIN_WEIGHT_EPS]
if not sv:
sv = [int(tri[int(np.argmax(w))])]
for s in sv:
support.setdefault(s, []).append((int(i), 1.0))
if support:
surf_ids = sorted(support.keys())
# When the WHOLE sim surface is driven, it forms a complete
# Dirichlet boundary: solve Laplace to drive the tet interior
# by the harmonic extension of the surface displacement, so
# the SOLID is fully kinematic (like a STATIC shell) and its
# free elastic interior cannot buckle into self-intersection.
# A partially-pinned surface is not a complete boundary, so
# the interior stays free (soft) there.
n_tet_verts = int(np.asarray(tet_mesh[0]).shape[0])
full_surface = len(surf_ids) == int(sim_surf_ids.size)
harmonic_M = None
interior_ids: list = []
if full_surface and n_tet_verts > len(surf_ids):
surf_set = set(surf_ids)
interior_ids = [
v for v in range(n_tet_verts) if v not in surf_set
]
if progress_callback is not None:
progress_callback(
"Building fully pinned SOLID interior map..."
)
harmonic_M = _build_harmonic_interior_operator(
n_tet_verts, tet_mesh[2], surf_ids, interior_ids,
)
if harmonic_M is not None:
driven = surf_ids + interior_ids
holder = _obj.pin(driven)
holder._data._blender_pin_indices = list(pin_index)
holder._data._tet_V = V
holder._data._blender_vert = vert
holder._data._sim_blender_weights = [
support[s] for s in surf_ids
]
# Interior (holder.index[len(surf_ids):]) follows
# M @ surface; surface keeps its blender transfer.
holder._data._harmonic = (len(surf_ids), harmonic_M)
if verbose:
print(f" > pin (harmonic solid): {len(surf_ids)} "
f"surface + {len(interior_ids)} interior = "
f"{len(driven)} tet verts (Laplace interior fill)")
else:
holder = _obj.pin(surf_ids)
holder._data._blender_pin_indices = list(pin_index)
holder._data._tet_V = V
holder._data._blender_vert = vert
holder._data._sim_blender_weights = [
support[s] for s in surf_ids
]
if verbose:
why = ("partial surface (interior left elastic)"
if not full_surface
else "harmonic solve unavailable/failed")
print(f" > pin (inverse mapped): {len(surf_ids)}/"
f"{int(sim_surf_ids.size)} sim surface verts from "
f"{len(pin_index)} blender pins [{why}]")
else:
# One holder for the whole pinned set. apply_pin_config
# later splits it per pin_group_id when the object has
# multiple distinct pin vertex groups. (A per-vertex
# _obj.pin([i]) loop here made N holders, and a keyframed
# pin then wrote N x M operation files for the solver.)
if pin_index:
_obj.pin(list(pin_index))
def _apply_stitch(self, obj, _obj, name, verbose):
"""Stitch pass: register a stitch asset and attach it to ``_obj``
when a stitch was resolved during planning.
"""
if _obj is None:
return
resolved_stitch = obj.get("_resolved_stitch", obj.get("stitch"))
if resolved_stitch is None:
return
stitch_data = resolved_stitch
stitch_name = f"{name}_stitch"
if verbose:
print(f" > stitch: {len(stitch_data[0])} edges")
self._asset.add.stitch(stitch_name, stitch_data)
_obj.stitch(stitch_name)