#!/usr/bin/env python3
# Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.
#!/usr/bin/env python3
"""Load and identify serialized kdFlex models for the kdpy launcher.
The launcher accepts one ``--from-model`` path without requiring the user to
name its serialized root type. This module owns that probing policy and
normalizes every successful load into :class:`LoadedModel`, allowing the rest
of kdpy to work with one small interface.
"""
from __future__ import annotations
import gc
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
import click
import Karana.Core as kc
import Karana.Dynamics as kd
from Karana.Dynamics.SOADyn_types import (
StatePropagatorDS,
SubGraphDS,
SubTreeDS,
)
from Karana.KUtils.BasicPrefab import BasicPrefabDS
from Karana.KUtils.Sim import Sim, SimDS
[docs]
@dataclass(frozen=True)
class LoadedModel:
"""Normalize a successfully loaded root model for launcher orchestration.
``state_target`` is the object reset from ``--from-state``. It may be a
subtree even when ``sim.mb`` is the complete multibody. ``model_target`` is
retained until output serialization so kdpy writes the same kind of model
that it read. The caller owns ``sim`` and therefore controls the lifetime of
every native object reachable through the other fields.
"""
sim: Sim
model_type: str
state_target: kd.SubTree
model_target: Any
def _loadSim(model_file: Path) -> tuple[Sim, kd.SubTree, Any]:
"""Load a Sim model into a standalone simulation."""
# Keep an independent serialized template for output. Sim construction
# resolves raw IDs and runtime contexts in the object it consumes, making
# that instance unsafe to write back as a portable prefab.
# Separate the portable output template before toSim resolves serialized IDs
# into process-local native references.
sim_ds, meta = SimDS.fromFile(model_file, get_meta=True)
output_ds = sim_ds.model_copy(deep=True)
sim = sim_ds.toSim()
# Retain the original file metadata independently of the runtime model.
sim.kdpy_model_meta = meta
return sim, sim.mb, output_ds
def _loadBasicPrefab(model_file: Path) -> tuple[Sim, kd.SubTree, Any]:
"""Load a BasicPrefab model into a standalone simulation."""
# Keep an independent serialized template for output. Sim construction
# resolves raw IDs and runtime contexts in the object it consumes, making
# that instance unsafe to write back as a portable prefab.
# Preserve an untouched serialization target because prefab construction
# binds context fields and node references into the consumed specification.
prefab_ds, meta = BasicPrefabDS.fromFile(model_file, get_meta=True)
output_ds = prefab_ds.model_copy(deep=True)
sim, _prefab = Sim.fromBasicPrefabDS(prefab_ds)
# Retain the original file metadata independently of the runtime model.
sim.kdpy_model_meta = meta
return sim, sim.mb, output_ds
def _loadStatePropagator(model_file: Path) -> tuple[Sim, kd.SubTree, Any]:
"""Load a StatePropagator and expose its bound subtree as state target."""
# Sim constructs the empty multibody needed by a standalone propagator file.
# A standalone propagator supplies its own subtree; the empty Sim supplies
# only the remaining process containers and cleanup ownership.
specification, meta = StatePropagatorDS.fromFile(model_file, get_meta=True)
sim = Sim(None, specification)
sim.kdpy_model_meta = meta
return sim, sim.sp.getSubTree(), sim.sp
def _loadSubGraph(model_file: Path) -> tuple[Sim, kd.SubTree, Any]:
"""Load a SubGraph model as the simulation multibody."""
# Sim converts the serialized graph and creates its default propagator.
# Passing the graph as Sim's multibody source makes the live multibody both
# the state target and the model serialization target.
specification, meta = SubGraphDS.fromFile(model_file, get_meta=True)
sim = Sim(specification, None)
sim.kdpy_model_meta = meta
return sim, sim.mb, sim.mb
def _loadSubTree(model_file: Path) -> tuple[Sim, kd.SubTree, Any]:
"""Attach a serialized SubTree below a new Sim's virtual root."""
# A SubTree needs an owning Multibody, which the empty Sim supplies.
# Create the owning multibody first; a serialized subtree cannot become live
# until it has both a parent body and the multibody's FrameContainer.
sim = Sim()
subtree_ds, meta = SubTreeDS.fromFile(model_file, get_meta=True)
sim.kdpy_model_meta = meta
subtree = subtree_ds.toSubTree(sim.mb, sim.mb.virtualRoot())
return sim, subtree, subtree
# Probe more-specific formats before SubTree because SubGraph extends SubTree.
MODEL_LOADERS: tuple[tuple[str, Callable[[Path], tuple[Sim, kd.SubTree, Any]]], ...] = (
("Sim", _loadSim),
("BasicPrefab", _loadBasicPrefab),
("StatePropagator", _loadStatePropagator),
("SubGraph", _loadSubGraph),
("SubTree", _loadSubTree),
)
[docs]
def loadModel(model_file: Path) -> LoadedModel:
"""Load the first supported kdFlex model type that accepts the file.
Each kdFlex ``fromFile`` call performs schema validation. A schema mismatch
is therefore the dispatch signal rather than a fatal error. Failed probes
can allocate native objects and register cleanup callbacks, so each attempt
is treated as a small transaction: remember the callback boundary, try the
loader, and unwind only callbacks introduced by that attempt on failure.
"""
# Accumulate concise probe diagnostics without retaining exception objects,
# whose tracebacks could keep failed native wrappers alive.
failures: list[str] = []
at_exit_fns = kc.BaseContainer.singleton().at_exit_fns
# Probe serially because native deserializers share object-identity maps.
for model_type, loader in MODEL_LOADERS:
# The callback boundary is the transaction marker for this probe. A
# successful loader transfers those callbacks to the returned Sim.
callback_count = len(at_exit_fns)
try:
# Deserialization validation determines whether this loader matches.
sim, state_target, model_target = loader(model_file)
# Broader schemas must not reinterpret an already accepted model.
return LoadedModel(sim, model_type, state_target, model_target)
except Exception as error: # Expected while probing incompatible schemas.
# Keep only text: an exception traceback retains loader and dispatch
# frames, including any Sim returned by a later successful probe.
failures.append(f"{model_type}: {type(error).__name__}: {error}")
# A failed loader may register native-object cleanup before schema
# validation finishes. Run only callbacks added by this attempt.
while len(at_exit_fns) > callback_count:
cleanup = at_exit_fns.pop(len(at_exit_fns) - 1)
try:
cleanup()
except Exception as cleanup_error: # Continue releasing objects.
failures.append(
f"{model_type} cleanup: {type(cleanup_error).__name__}: {cleanup_error}"
)
# Collect deserialization and traceback cycles before trying a
# different schema while their native modules remain available.
gc.collect()
# Preserve concise diagnostics while identifying every attempted format.
attempted = ", ".join(model_type for model_type, _loader in MODEL_LOADERS)
details = "; ".join(failures)
raise click.ClickException(
f"Could not load '{model_file}' as any supported model type ({attempted}). {details}"
)
