Fixed usd_replicate() authoring environment grid positions on nested prims (e.g. cameras), overwriting their local transforms.

source/isaaclab/changelog.d/fix-incorrect-camera-xform-in-usd-replicate.rst

@@ -0,0 +1,6 @@
+Fixed
+^^^^^
+
+* Fixed :func:`~isaaclab.cloner.usd.usd_replicate` authoring environment grid
+  positions on nested replicated prims (e.g. cameras), overwriting their local
+  transforms.

source/isaaclab/isaaclab/cloner/__init__.pyi

@@ -17,6 +17,7 @@ __all__ = [
     "REPLICATION_QUEUE",
     "replicate",
     "resolve_clone_plan_source",
+    "split_clone_template",
     "queue_usd_replication",
     "sequential",
     "UsdReplicateContext",
@@ -34,6 +35,7 @@ from .cloner_utils import (
     iter_clone_plan_matches,
     make_clone_plan,
     resolve_clone_plan_source,
+    split_clone_template,
 )
 from .replicate_session import (
     REPLICATION_QUEUE,

source/isaaclab/isaaclab/cloner/clone_plan.py

@@ -61,9 +61,10 @@ def from_env_0(
             device: Torch device for the mask and env id buffers.
             positions: Optional per-env world positions [m], shape ``[num_clones, 3]``.
         """
+        from .cloner_utils import split_clone_template  # noqa: PLC0415
         from .replicate_session import REPLICATION_QUEUE  # noqa: PLC0415
 
-        prefix, _, _ = destination.partition("{}")
+        prefix, _ = split_clone_template(destination)
         cfg_rows: dict[int, tuple[int, ...]] = {
             id(cfg): (0,) for cfg, _ in REPLICATION_QUEUE if cfg.prim_path.startswith(prefix)
         }

source/isaaclab/isaaclab/cloner/cloner_utils.py

@@ -26,6 +26,27 @@
 logger = logging.getLogger(__name__)
 
 
+def split_clone_template(destination_template: str) -> tuple[str, str]:
+    """Split a clone destination template around its clone slot.
+
+    The ``"{}"`` slot represents one concrete environment/instance path segment.
+
+    Args:
+        destination_template: Destination path template with ``"{}"`` for the instance id.
+
+    Returns:
+        The ``(prefix, suffix)`` around the clone slot.
+
+    Raises:
+        ValueError: If ``destination_template`` does not contain a clone slot.
+    """
+    destination_template = destination_template.rstrip("/") or "/"
+    prefix, slot, suffix = destination_template.partition("{}")
+    if slot != "{}":
+        raise ValueError(f"Clone destination template must contain '{{}}': {destination_template!r}.")
+    return prefix, suffix
+
+
 def get_suffix(path_expr: str, destination_template: str) -> str | None:
     """Return the part of ``path_expr`` below a destination template's env-instance root.
 
@@ -47,7 +68,7 @@ def get_suffix(path_expr: str, destination_template: str) -> str | None:
         >>> get_suffix("/World/scenes/env_3/sub/Robot/base", tmpl) is None
         True
     """
-    pattern = re.compile(r"[^/]+".join(re.escape(part) for part in destination_template.split("{}")))
+    pattern = re.compile(r"[^/]+".join(re.escape(part) for part in split_clone_template(destination_template)))
     match = pattern.match(path_expr)
     if match is None:
         return None

source/isaaclab/isaaclab/cloner/usd.py

@@ -13,6 +13,7 @@
 from pxr import Gf, Sdf, Usd, UsdGeom, Vt
 
 from ._fabric_notices import disabled_fabric_change_notifies
+from .cloner_utils import split_clone_template
 from .replicate_session import REPLICATION_QUEUE
 
 
@@ -55,8 +56,10 @@ def queue(
             source: Source prim path.
             destination: Destination path template with ``"{}"`` for env id.
             env_ids: Environment ids selected for this source row.
-            positions: Optional per-environment world positions [m].
-            quaternions: Optional per-environment orientations in xyzw order.
+            positions: Optional per-environment world positions [m]. Authored only for
+                instance-root destination templates (for example, ``.../env_{}``).
+            quaternions: Optional per-environment orientations in xyzw order. Authored only
+                for instance-root destination templates (for example, ``.../env_{}``).
         """
         self._queue.append((source, destination, env_ids, positions, quaternions))
 
@@ -77,8 +80,10 @@ def queue_mapping(
             destinations: Destination path templates with ``"{}"`` for env id.
             env_ids: Environment indices.
             mask: Optional per-source or shared mask.
-            positions: Optional per-environment world positions [m].
-            quaternions: Optional per-environment orientations in xyzw order.
+            positions: Optional per-environment world positions [m]. Authored only for
+                instance-root destination templates (for example, ``.../env_{}``).
+            quaternions: Optional per-environment orientations in xyzw order. Authored only
+                for instance-root destination templates (for example, ``.../env_{}``).
         """
         for i, source in enumerate(sources):
             self.queue(
@@ -115,14 +120,18 @@ def dp_depth(template: str) -> int:
         for depth in sorted(depth_to_items.keys()):
             with Sdf.ChangeBlock():
                 for src, tmpl, target_envs, positions, quaternions in depth_to_items[depth]:
+                    _, clone_suffix = split_clone_template(tmpl)
+                    is_instance_root = clone_suffix == ""
+
                     for wid in target_envs.tolist():
                         wid = int(wid)
                         dp = tmpl.format(wid)
                         Sdf.CreatePrimInLayer(rl, dp)
                         if src != dp:
                             Sdf.CopySpec(rl, Sdf.Path(src), rl, Sdf.Path(dp))
 
-                        if positions is not None or quaternions is not None:
+                        # Author positions/quaternions for instance roots only.
+                        if is_instance_root and (positions is not None or quaternions is not None):
                             ps = rl.GetPrimAtPath(dp)
                             op_names = []
                             if positions is not None:
@@ -177,8 +186,10 @@ def usd_replicate(
         destinations: Destination formattable templates with ``"{}"`` for env index.
         env_ids: Environment indices.
         mask: Optional per-source or shared mask. ``None`` selects all.
-        positions: Optional positions [m], shape ``[E, 3]``, authored as ``xformOp:translate``.
-        quaternions: Optional orientations in xyzw order, shape ``[E, 4]``, authored as ``xformOp:orient``.
+        positions: Optional positions [m], shape ``[E, 3]``. Authored as ``xformOp:translate`` only
+            for env-instance root destinations (``.../env_{}``).
+        quaternions: Optional orientations in xyzw order, shape ``[E, 4]``. Authored as
+            ``xformOp:orient`` only for env-instance root destinations (``.../env_{}``).
     """
     ctx = UsdReplicateContext(stage)
     ctx.queue_mapping(sources, destinations, env_ids, mask, positions=positions, quaternions=quaternions)

source/isaaclab/test/sim/test_cloner.py

@@ -34,6 +34,7 @@
     replicate,
     resolve_clone_plan_source,
     sequential,
+    split_clone_template,
     usd_replicate,
 )
 from isaaclab.sim import build_simulation_context
@@ -56,6 +57,14 @@ def _drain_replication_queue():
     REPLICATION_QUEUE.clear()
 
 
+def test_split_clone_template():
+    """Split clone destination templates around their clone slot."""
+    assert split_clone_template("/World/envs/env_{}/Robot") == ("/World/envs/env_", "/Robot")
+    assert split_clone_template("/World/scenes/{}/") == ("/World/scenes/", "")
+    with pytest.raises(ValueError, match="must contain"):
+        split_clone_template("/World/envs/env_0/Robot")
+
+
 def test_usd_replicate_with_positions_and_mask(sim):
     """Replicate sources to selected envs and author translate ops from positions."""
     # Prepare sources under /World/template
@@ -120,6 +129,48 @@ def test_usd_replicate_context_queue_and_replicate(sim):
     assert stage.GetPrimAtPath("/World/envs/env_1/A").IsValid()
 
 
+def test_usd_replicate_nested_asset_preserves_local_offset_with_positions(sim):
+    """Grid positions are authored on env roots but not on nested replicated assets."""
+    camera_offset = (0.57, -0.8, 0.5)
+    num_envs = 2
+    env_ids = torch.arange(num_envs, dtype=torch.long)
+    positions, _ = grid_transforms(num_envs, 3.0, device=sim.cfg.device)
+
+    sim_utils.create_prim("/World/envs", "Xform")
+    sim_utils.create_prim("/World/envs/env_0", "Xform")
+    sim_utils.create_prim("/World/envs/env_0/Camera", "Camera", translation=camera_offset)
+
+    stage = sim_utils.get_current_stage()
+    usd_replicate(
+        stage,
+        sources=["/World/envs/env_0"],
+        destinations=["/World/envs/env_{}"],
+        env_ids=env_ids,
+        positions=positions,
+    )
+    usd_replicate(
+        stage,
+        sources=["/World/envs/env_0/Camera"],
+        destinations=["/World/envs/env_{}/Camera"],
+        env_ids=env_ids,
+        positions=positions,
+    )
+
+    for env_idx in range(num_envs):
+        env_prim = stage.GetPrimAtPath(f"/World/envs/env_{env_idx}")
+        assert env_prim.IsValid()
+        env_translate = env_prim.GetAttribute("xformOp:translate").Get()
+        assert env_translate is not None
+        expected_env_pos = positions[env_idx].tolist()
+        assert (env_translate[0], env_translate[1], env_translate[2]) == pytest.approx(expected_env_pos)
+
+        camera_prim = stage.GetPrimAtPath(f"/World/envs/env_{env_idx}/Camera")
+        assert camera_prim.IsValid()
+        camera_translate = camera_prim.GetAttribute("xformOp:translate").Get()
+        assert camera_translate is not None
+        assert (camera_translate[0], camera_translate[1], camera_translate[2]) == pytest.approx(camera_offset)
+
+
 def test_disabled_fabric_change_notifies_noops_when_usdrt_unavailable(monkeypatch):
     """Fabric notice suspension no-ops when Carbonite bindings exist but ``usdrt`` does not."""
     import builtins

source/isaaclab_newton/isaaclab_newton/sensors/ray_caster/ray_caster.py

@@ -15,7 +15,7 @@
 from pxr import UsdPhysics
 
 import isaaclab.sim as sim_utils
-from isaaclab.cloner import resolve_clone_plan_source
+from isaaclab.cloner import resolve_clone_plan_source, split_clone_template
 from isaaclab.sensors.ray_caster.base_ray_caster import BaseRayCaster
 from isaaclab.sensors.ray_caster.kernels import (
     ALIGNMENT_BASE,
@@ -117,9 +117,7 @@ def _register_sites_for_expr(self, prim_expr: str) -> list[str]:
         plan = sim_utils.SimulationContext.instance().get_clone_plan()
         if plan is not None:
             for destination_template in plan.destinations:
-                if "{}" not in destination_template:
-                    continue
-                destination_prefix, _ = destination_template.split("{}", 1)
+                destination_prefix, _ = split_clone_template(destination_template)
                 if attach_expr.startswith(destination_prefix) and "/" not in attach_expr[len(destination_prefix) :]:
                     return [NewtonManager.cl_register_site(None, identity, per_world=True)]
 

source/isaaclab_newton/isaaclab_newton/sim/views/newton_site_frame_view.py

@@ -14,7 +14,7 @@
 from pxr import UsdPhysics
 
 import isaaclab.sim as sim_utils
-from isaaclab.cloner.cloner_utils import get_suffix, iter_clone_plan_matches
+from isaaclab.cloner.cloner_utils import get_suffix, iter_clone_plan_matches, split_clone_template
 from isaaclab.physics import PhysicsEvent
 from isaaclab.sim.views.base_frame_view import BaseFrameView
 from isaaclab.utils.string import resolve_matching_names
@@ -323,8 +323,8 @@ def _resolve_source_prim(
 
         ref_path = source_root
         if source_root is not None and destination_template is not None:
-            instance_template = destination_template.partition("{}")[0] + "{}"
-            source_suffix = get_suffix(source_root, instance_template)
+            template_prefix, _ = split_clone_template(destination_template)
+            source_suffix = get_suffix(source_root, template_prefix + "{}")
             if source_suffix is not None:
                 ref_path = source_root[: -len(source_suffix)] if source_suffix else source_root
         ref_prim = stage.GetPrimAtPath(ref_path) if ref_path is not None else None

source/isaaclab_ovphysx/isaaclab_ovphysx/cloner/replicate.py

@@ -27,6 +27,7 @@
 
 from pxr import Sdf, Usd
 
+from isaaclab.cloner.cloner_utils import split_clone_template
 from isaaclab.cloner.replicate_session import REPLICATION_QUEUE
 
 
@@ -90,8 +91,8 @@ def queue_mapping(
         for i, src in enumerate(sources):
             active_env_ids = _select_env_ids(env_ids, mapping, i).tolist()
 
-            pre, _, suf = destinations[i].partition("{}")
             self_env_id: int | None = None
+            pre, suf = split_clone_template(destinations[i])
             candidate = src.removeprefix(pre).removesuffix(suf)
             if candidate.isdigit():
                 self_env_id = int(candidate)

source/isaaclab_physx/isaaclab_physx/cloner/replicate.py

@@ -13,6 +13,7 @@
 from omni.physx import get_physx_replicator_interface
 from pxr import Sdf, Usd, UsdUtils
 
+from isaaclab.cloner.cloner_utils import split_clone_template
 from isaaclab.cloner.replicate_session import REPLICATION_QUEUE
 
 
@@ -81,7 +82,7 @@ def queue_mapping(
         for i, src in enumerate(sources):
             worlds = _select_env_ids(env_ids, mapping, i).tolist()
             if exclude_self_replication:
-                pre, _, suf = destinations[i].partition("{}")
+                pre, suf = split_clone_template(destinations[i])
                 self_id = src.removeprefix(pre).removesuffix(suf)
                 if self_id.isdigit():
                     filtered = [w for w in worlds if w != int(self_id)]