feat(tracking): v0.1.2 双手轨迹回放与 GUI 预览

支持 L20 右手模型身份与严格浮动控制覆盖层,新增完整轨迹受限回放、时间拉伸及可选 GUI 显示。

验证:88 项 CPU/USD 回归、Ruff、格式与独立暂存审查通过。历史 80 倍降速回放完成 2x32000 步;整合后 GUI E2E 和完整 pre-commit 未执行,相关边界见 L20_TRACKING.md。

右手 USD、示教数据、媒体和日志未纳入提交;资产存储及许可仍待确认。保留原控制与安全阈值。
This commit is contained in:
2026-09-14 13:57:57 +08:00
parent 057f4c2cf6
commit 8e7ab5fc76
20 changed files with 2218 additions and 62 deletions
+2 -2
View File
@@ -1,7 +1,7 @@
[package]
# Semantic Versioning is used: https://semver.org/
version = "0.1.1"
version = "0.1.2"
# Description
category = "isaaclab"
@@ -42,4 +42,4 @@ name = "dex_workbench.ui_extension_example"
# after this extension is installed with ./isaaclab.sh --install.
# List package names only; version ranges, extras, and platform markers
# come from this extension's setup.py metadata.
# pip_upgrade_dependencies = ["example_package"]
# pip_upgrade_dependencies = ["example_package"]
@@ -52,7 +52,26 @@ def dependencies(source):
return [{"path": name, "sha256": sha} for name, sha in entries], digest.hexdigest()
def inspect(source, source_urdf=None):
def require_static_endpoint(stage, path):
"""Only an identity default Xform can stand for the importer's static world."""
prim = stage.GetPrimAtPath(path)
_require(prim == stage.GetDefaultPrim() and prim.IsA(UsdGeom.Xform), "Unsupported static anchor endpoint")
while prim and not prim.IsPseudoRoot():
_require(
not prim.HasAPI(UsdPhysics.RigidBodyAPI) and not prim.HasAPI(UsdPhysics.ArticulationRootAPI),
"Static endpoint has rigid/articulation API",
)
_require(not any(a.GetTimeSamples() for a in prim.GetAttributes()), "Sampled static endpoint")
xform = UsdGeom.Xformable(prim)
_require(not xform.GetResetXformStack(), "Static endpoint reset stack prohibited")
_require(
Gf.IsClose(UsdGeom.XformCache().GetLocalToWorldTransform(prim), Gf.Matrix4d(1), 1e-12),
"Static endpoint must have identity world transform",
)
prim = prim.GetParent()
def inspect(source, source_urdf=None, hand_side=None):
source = Path(source).resolve(strict=True)
deps, bundle_sha = dependencies(source)
stage = Usd.Stage.Open(str(source))
@@ -86,7 +105,9 @@ def inspect(source, source_urdf=None):
b0 = [str(p) for p in joint.GetBody0Rel().GetTargets()]
b1 = [str(p) for p in joint.GetBody1Rel().GetTargets()]
_require(len(b1) == 1 and b1[0] in body_paths, f"Unresolved/non-body body1: {prim.GetPath()}")
if not b0:
if not b0 or (len(b0) == 1 and b0[0] not in body_paths):
if b0:
require_static_endpoint(stage, b0[0])
_require(prim.IsA(UsdPhysics.FixedJoint), "Unsupported non-fixed world joint")
world_joints.append({"path": str(prim.GetPath()), "body1": b1[0]})
continue
@@ -187,7 +208,14 @@ def inspect(source, source_urdf=None):
coupling_status = "MIMIC_SCHEMA_PRESENT_RUNTIME_UNVERIFIED"
elif source_info["mimic"]:
coupling_status = "MISSING_MIMIC_SCHEMA"
identity = {}
if hand_side is not None:
from .identity import manifest_side
identity = {"hand_side": hand_side}
manifest_side(dict(identity, source_urdf=source_info))
return {
**identity,
"manifest_version": "l20_asset_manifest_v1",
"asset_sha256": bundle_sha,
"entry_file": source.name,
@@ -238,9 +266,11 @@ def prepare_floating_overlay(source, output):
before = inspect(source)
_require(before["physics_variant"] == "physx", "Only the current PhysX variant is supported")
anchors = before["world_fixed_joints"]
_require(len(anchors) == 1, "Expected one fixed-world anchor")
site = before["articulation_roots"][0]
_require(
len(anchors) == 1 and before["articulation_roots"] == [anchors[0]["path"]],
"Expected a fixed-world articulation root; will not guess topology",
site in (anchors[0]["path"], before["root_body_path"].rsplit("/", 1)[0]),
"Expected fixed anchor or exact root container articulation site",
)
original = Usd.Stage.Open(str(source))
stage = Usd.Stage.CreateInMemory()
@@ -252,19 +282,28 @@ def prepare_floating_overlay(source, output):
anchor = stage.GetPrimAtPath(anchors[0]["path"])
# Snapshot resolved schema fallbacks before removing the API. Preserve them
# explicitly on the inert anchor and the new root, not just authored values.
articulation_site = stage.GetPrimAtPath(site)
articulation_values = []
for attr in anchor.GetAttributes():
if attr.GetName().startswith("physxArticulation:") and attr.Get() is not None:
for attr in articulation_site.GetAttributes():
if (
attr.GetName().startswith("physxArticulation:") or attr.GetName() == "newton:selfCollisionEnabled"
) and attr.Get() is not None:
_require(not attr.GetTimeSamples(), f"Time samples prohibited: {attr.GetPath()}")
articulation_values.append((attr.GetName(), attr.GetTypeName(), attr.Get()))
anchor.RemoveAPI(UsdPhysics.ArticulationRootAPI)
anchor.RemoveAppliedSchema("PhysxArticulationAPI")
source_schemas = articulation_site.GetMetadata("apiSchemas").GetAppliedItems()
articulation_site.RemoveAPI(UsdPhysics.ArticulationRootAPI)
articulation_site.RemoveAppliedSchema("PhysxArticulationAPI")
if "NewtonArticulationRootAPI" in source_schemas:
articulation_site.RemoveAppliedSchema("NewtonArticulationRootAPI")
UsdPhysics.Joint(anchor).GetJointEnabledAttr().Set(False)
root = stage.GetPrimAtPath(before["root_body_path"])
UsdPhysics.ArticulationRootAPI.Apply(root)
root.AddAppliedSchema("PhysxArticulationAPI")
if "PhysxArticulationAPI" in source_schemas:
root.AddAppliedSchema("PhysxArticulationAPI")
if "NewtonArticulationRootAPI" in source_schemas:
root.AddAppliedSchema("NewtonArticulationRootAPI")
for name, dtype, value in articulation_values:
anchor.CreateAttribute(name, dtype).Set(value)
articulation_site.CreateAttribute(name, dtype).Set(value)
root.CreateAttribute(name, dtype).Set(value)
stage.GetRootLayer().customLayerData = {"purpose": "L20 topology-only diagnostic; dynamic replay BLOCKED"}
cache0, cache1 = UsdGeom.XformCache(), UsdGeom.XformCache()
@@ -305,15 +344,18 @@ def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("asset", type=Path)
parser.add_argument("--source-urdf", type=Path, help="Optional local original; inspected, never modified")
parser.add_argument(
"--hand-side", choices=("left", "right"), help="Explicit inspected identity; right requires source URDF"
)
parser.add_argument("--output", type=Path, required=True, help="New JSON manifest; no overwrite")
parser.add_argument("--floating-overlay", type=Path, help="New topology-only USD layer; no overwrite")
args = parser.parse_args()
try:
_require(not args.output.exists(), "Refusing to overwrite manifest")
manifest = inspect(args.asset, args.source_urdf)
manifest = inspect(args.asset, args.source_urdf, args.hand_side)
if args.floating_overlay:
prepare_floating_overlay(args.asset, args.floating_overlay)
manifest = inspect(args.floating_overlay, args.source_urdf)
manifest = inspect(args.floating_overlay, args.source_urdf, args.hand_side)
args.output.parent.mkdir(parents=True, exist_ok=True)
with args.output.open("x", encoding="utf-8") as stream:
json.dump(manifest, stream, indent=2, allow_nan=False)
@@ -1,6 +1,7 @@
"""CPU preparation CLI: validate, synthesize test references, or resample HDF5."""
import argparse
import hashlib
import json
import os
import tempfile
@@ -9,6 +10,7 @@ from pathlib import Path
import h5py
import numpy as np
from .identity import manifest_side
from .trajectory import (
ContractError,
Demonstrations,
@@ -48,6 +50,26 @@ def publish_validated(path, data, manifest):
os.link(temporary, path)
def stretch_time(data, factor):
"""Slow references without altering any pose, joint, valid mask or coordinate metadata."""
require(np.isfinite(factor) and factor >= 1, "Time stretch factor must be finite and >= 1")
episodes = {}
for name, episode in data.episodes.items():
with np.errstate(over="ignore"):
time = episode.time * factor
require(np.isfinite(time).all(), "Time stretch overflow")
episodes[name] = Episode(
time,
episode.wrist_position.copy(),
episode.wrist_quaternion.copy(),
episode.joint_position.copy(),
episode.valid.copy(),
)
metadata = dict(data.metadata)
metadata["source_description"] += f"; uniform time stretch factor={factor:g}; all original samples preserved"
return Demonstrations(metadata, data.joint_names, data.world_from_source.copy(), episodes)
def synthetic(manifest, profile="small"):
"""Named analytic fixtures, not experts. ``range_4s`` has smooth out-and-back endpoints."""
require(profile in ("small", "range_4s"), "Unknown synthetic profile")
@@ -81,7 +103,7 @@ def synthetic(manifest, profile="small"):
metadata = {
"schema_version": "l20_tracking_v1",
"embodiment": "L20",
"hand_side": "left",
"hand_side": manifest_side(manifest),
"asset_sha256": manifest["asset_sha256"],
"root_link": manifest["root_link"],
"metric_scale_provenance": "Analytic SI fixture, not video reconstruction or measurement",
@@ -124,6 +146,13 @@ def main():
resample.add_argument("--manifest", required=True, type=Path)
resample.add_argument("--hz", required=True, type=float)
resample.add_argument("--output", required=True, type=Path)
stretch = commands.add_parser(
"stretch-time", help="Uniform slowdown; no pose/coordinate changes or identity override"
)
stretch.add_argument("input", type=Path)
stretch.add_argument("--manifest", required=True, type=Path)
stretch.add_argument("--factor", required=True, type=float)
stretch.add_argument("--output", required=True, type=Path)
gate = commands.add_parser("replay-check", help="Fails closed until a physical tracking Adapter is implemented")
gate.add_argument("--manifest", required=True, type=Path)
args = parser.parse_args()
@@ -139,6 +168,12 @@ def main():
publish_validated(args.output, data, manifest)
else:
data = load(args.input, manifest)
if args.command == "stretch-time":
data = stretch_time(data, args.factor)
data.metadata["source_description"] += (
f"; stretch input HDF5 SHA256={hashlib.sha256(args.input.read_bytes()).hexdigest()}"
)
publish_validated(args.output, data, manifest)
if args.command == "resample":
require(np.isfinite(args.hz) and 0 < args.hz <= 1000, "hz must be finite and in (0, 1000]")
episodes = {}
@@ -0,0 +1,32 @@
"""Explicit inspected L20 source identities, not inferred hardware motor layouts."""
LEFT_URDF_SHA = "b8ef22e436ab311fb61f87091b72ae717821092471d628d0ea9475b5088daa5d"
RIGHT_URDF_SHA = "aa198155651eb7e815f1c27bc8bd9c902e969e81c1165aed756779db378e715d"
SOURCE_SHA_BY_SIDE = {"left": LEFT_URDF_SHA, "right": RIGHT_URDF_SHA}
def manifest_side(manifest):
"""Legacy side-less manifests are left only; right requires explicit provenance."""
side = manifest.get("hand_side", "left")
if side not in SOURCE_SHA_BY_SIDE:
raise ValueError("hand_side must be left or right")
sha = manifest.get("source_urdf", {}).get("sha256")
if side == "right" and sha != RIGHT_URDF_SHA:
raise ValueError("Right manifest requires inspected right URDF identity")
if sha in SOURCE_SHA_BY_SIDE.values() and sha != SOURCE_SHA_BY_SIDE[side]:
raise ValueError("Manifest hand_side conflicts with source URDF identity")
return side
def model_mimic(side):
return [
{"joint": f"{finger}_dip", "reference": f"{finger}_pip", "multiplier": 0.89, "offset_rad": 0.0}
for finger in ("index", "middle", "pinky", "ring")
] + [
{
"joint": "thumb_ip",
"reference": "thumb_mcp",
"multiplier": 1.03 if side == "right" else 1.02,
"offset_rad": 0.0,
}
]
@@ -12,14 +12,12 @@ from pathlib import Path
from pxr import Gf, Plug, Sdf, Usd, UsdGeom, UsdPhysics
from .asset import _require, dependencies, inspect, prepare_floating_overlay
from .asset import _require, dependencies, inspect, prepare_floating_overlay, require_static_endpoint
from .identity import LEFT_URDF_SHA, SOURCE_SHA_BY_SIDE, manifest_side, model_mimic
SOURCE_URDF_SHA = "b8ef22e436ab311fb61f87091b72ae717821092471d628d0ea9475b5088daa5d"
SOURCE_URDF_SHA = LEFT_URDF_SHA
BACKEND_VERSION = "110.1.13"
MIMIC = [
{"joint": f"{finger}_dip", "reference": f"{finger}_pip", "multiplier": 0.89, "offset_rad": 0.0}
for finger in ("index", "middle", "pinky", "ring")
] + [{"joint": "thumb_ip", "reference": "thumb_mcp", "multiplier": 1.02, "offset_rad": 0.0}]
MIMIC = model_mimic("left") # Existing left callers retain the same constant.
DRIVE_FIELDS = ("stiffness", "damping", "maxForce")
@@ -33,7 +31,8 @@ def schema_context():
"""Resolved schema fallbacks are part of the authoring/inspection contract."""
registry = Usd.SchemaRegistry()
return {
name: bool(registry.FindAppliedAPIPrimDefinition(name)) for name in ("PhysxArticulationAPI", "NewtonMimicAPI")
name: bool(registry.FindAppliedAPIPrimDefinition(name))
for name in ("PhysxArticulationAPI", "NewtonMimicAPI", "NewtonArticulationRootAPI")
}
@@ -71,11 +70,17 @@ def newton_mimic(prim):
def bind_source(source, manifest):
"""Reject stale identity/structural metadata before authoring or accepting an overlay."""
side = manifest_side(manifest)
actual = inspect(source)
actual["hand_side"] = side
for key in ("asset_sha256", "joints", "root_body_path", "root_link", "dependencies", "bodies"):
_require(actual[key] == manifest[key], f"Stale/inconsistent source manifest: {key}")
_require(manifest["source_urdf"]["sha256"] == SOURCE_URDF_SHA, "Unapproved URDF provenance")
_require(manifest["source_urdf"]["mimic"] == MIMIC, "Unexpected L20 mimic equations")
_require(manifest["source_urdf"]["sha256"] == SOURCE_SHA_BY_SIDE[side], "Unapproved URDF provenance")
_require(manifest["source_urdf"]["mimic"] == model_mimic(side), "Unexpected L20 mimic equations")
anchors = actual["world_fixed_joints"]
_require(len(anchors) == 1, "Expected one source world anchor")
expected_site = anchors[0]["path"] if side == "left" else actual["default_prim"] + "/Geometry"
_require(actual["articulation_roots"] == [expected_site], "Unexpected source articulation container")
_require(actual["root_link"] == "hand_base_link" and len(actual["joints"]) == 21, "Wrong L20 model")
_require(actual["physics_variant"] == "physx", "PhysX variant required")
validate_mimic(Usd.Stage.Open(str(source)), actual)
@@ -84,7 +89,8 @@ def bind_source(source, manifest):
def validate_mimic(stage, manifest, passive=False):
joints = {j["name"]: j for j in manifest["joints"]}
followers = {eq["joint"] for eq in MIMIC}
equations = model_mimic(manifest.get("hand_side", "left"))
followers = {eq["joint"] for eq in equations}
for name, joint in joints.items():
prim = stage.GetPrimAtPath(joint["path"])
_require(not any("PhysxMimic" in s or "Tendon" in s for s in schemas(prim)), "Duplicate/unsupported coupling")
@@ -93,7 +99,7 @@ def validate_mimic(stage, manifest, passive=False):
_require(joint["axis"] in ("X", "Y", "Z"), "Invalid revolute axis")
if name not in followers:
continue
equation = next(eq for eq in MIMIC if eq["joint"] == name)
equation = next(eq for eq in equations if eq["joint"] == name)
leader, coefficient, offset = newton_mimic(prim)
_require(leader == joints[equation["reference"]]["path"], "Wrong mimic leader")
_require(abs(coefficient - equation["multiplier"]) < 1e-6, "Wrong mimic multiplier/sign")
@@ -107,7 +113,10 @@ def validate_mimic(stage, manifest, passive=False):
def _preserved(original, stage, before):
"""Only leaf world-anchor deactivation, root migration and follower drive suppression are allowed."""
followers = {j["path"] for j in before["joints"] if j["name"] in {eq["joint"] for eq in MIMIC}}
followers = {
j["path"] for j in before["joints"] if j["name"] in {eq["joint"] for eq in model_mimic(before["hand_side"])}
}
site = before["articulation_roots"][0]
anchor = before["world_fixed_joints"][0]["path"]
root = before["root_body_path"]
source_anchor = original.GetPrimAtPath(anchor)
@@ -115,7 +124,11 @@ def _preserved(original, stage, before):
source_anchor.IsA(UsdPhysics.FixedJoint) and not source_anchor.GetAllChildren(), "Expected leaf world anchor"
)
joint = UsdPhysics.Joint(source_anchor)
_require(not joint.GetBody0Rel().GetTargets(), "Anchor must attach directly to world")
targets = joint.GetBody0Rel().GetTargets()
if targets:
_require(len(targets) == 1 and before["hand_side"] == "right", "Unexpected anchor endpoint")
require_static_endpoint(original, targets[0])
require_static_endpoint(stage, targets[0])
_require(joint.GetBody1Rel().GetTargets() == [Sdf.Path(root)], "Anchor must attach only to root")
c0, c1 = UsdGeom.XformCache(), UsdGeom.XformCache()
# TraverseAll retains the deliberately inactive leaf: no other deletion or
@@ -133,17 +146,22 @@ def _preserved(original, stage, before):
if path == anchor:
_require(after.GetAttribute("physics:jointEnabled").Get() is False, "Inactive anchor must remain disabled")
allowed.add("physics:jointEnabled")
expected -= {"PhysicsArticulationRootAPI", "PhysxArticulationAPI"}
if path == site:
expected -= {"PhysicsArticulationRootAPI", "PhysxArticulationAPI", "NewtonArticulationRootAPI"}
if path == root:
expected |= {"PhysicsArticulationRootAPI", "PhysxArticulationAPI"}
expected.add("PhysicsArticulationRootAPI")
if "PhysxArticulationAPI" in schemas(original.GetPrimAtPath(site)):
expected.add("PhysxArticulationAPI")
if "NewtonArticulationRootAPI" in schemas(original.GetPrimAtPath(site)):
expected.add("NewtonArticulationRootAPI")
allowed |= {
a.GetName()
for a in stage.GetPrimAtPath(anchor).GetAttributes()
if a.GetName().startswith("physxArticulation:")
for a in stage.GetPrimAtPath(site).GetAttributes()
if a.GetName().startswith("physxArticulation:") or a.GetName() == "newton:selfCollisionEnabled"
}
for name in allowed:
_require(
after.GetAttribute(name).Get() == stage.GetPrimAtPath(anchor).GetAttribute(name).Get(),
after.GetAttribute(name).Get() == stage.GetPrimAtPath(site).GetAttribute(name).Get(),
"Root solver property changed",
)
if path in followers:
@@ -184,13 +202,17 @@ def inspect_prepared(output, source_manifest):
source = (output.parent / layer.subLayerPaths[0]).resolve(strict=True)
before = bind_source(source, source_manifest)
_require(layer.customLayerData.get("source_bundle_sha256") == before["asset_sha256"], "Overlay provenance mismatch")
_require(layer.customLayerData.get("source_urdf_sha256") == SOURCE_URDF_SHA, "Overlay URDF mismatch")
_require(
layer.customLayerData.get("source_urdf_sha256") == SOURCE_SHA_BY_SIDE[before["hand_side"]],
"Overlay URDF mismatch",
)
_require(
dict(layer.customLayerData.get("schema_context", {})) == schema_context(),
"Authoring schema context mismatch; regenerate from source with --schema-plugin-path for both "
"installed PhysX and Newton plugins, and use the same plugins when inspecting",
)
after = inspect(output)
after["hand_side"] = before["hand_side"]
_require(
not after["world_fixed_joints"] and after["articulation_roots"] == [before["root_body_path"]], "Not floating"
)
@@ -208,7 +230,8 @@ def inspect_prepared(output, source_manifest):
"prepared_asset_sha256": after["asset_sha256"],
"source_asset_sha256": before["asset_sha256"],
"independent_joint_names": masters,
"mimic": MIMIC,
"mimic": model_mimic(before["hand_side"]),
"hand_side": before["hand_side"],
"required_physx_version": BACKEND_VERSION,
"root_body_path": before["root_body_path"],
"runtime_verified": False,
@@ -225,7 +248,7 @@ def prepare(source, output, manifest):
prepare_floating_overlay(source, temporary)
stage = Usd.Stage.Open(str(temporary))
for joint in before["joints"]:
if joint["name"] not in {eq["joint"] for eq in MIMIC}:
if joint["name"] not in {eq["joint"] for eq in model_mimic(before["hand_side"])}:
continue
prim = stage.GetPrimAtPath(joint["path"])
# Preserve resolved drive defaults as inert attributes when removing API.
@@ -246,7 +269,7 @@ def prepare(source, output, manifest):
stage.GetRootLayer().customLayerData = {
"purpose": "L20 experimental floating tracking; runtime UNVERIFIED",
"source_bundle_sha256": before["asset_sha256"],
"source_urdf_sha256": SOURCE_URDF_SHA,
"source_urdf_sha256": SOURCE_SHA_BY_SIDE[before["hand_side"]],
"physx_evidence_version": BACKEND_VERSION,
"schema_context": schema_context(),
}
@@ -0,0 +1,63 @@
"""Presentation-only USD preview; import at runtime only after AppLauncher.
CPU USD tests may use this without Kit. No robot materials or physics are edited.
"""
import numpy as np
from pxr import Gf, Sdf, UsdGeom, UsdLux, UsdPhysics, UsdShade
def create_preview(stage, wrist_position):
"""Add the approved contrast preview and return its fixed camera eye/target."""
dome = UsdLux.DomeLight.Define(stage, "/World/PreviewLight")
dome.CreateIntensityAttr(700.0)
dome.CreateColorAttr(Gf.Vec3f(1.0, 1.0, 1.0))
key = UsdLux.DistantLight.Define(stage, "/World/PreviewKey")
key.CreateIntensityAttr(800.0)
key.CreateAngleAttr(12.0)
UsdGeom.Xformable(key.GetPrim()).AddRotateXYZOp().Set(Gf.Vec3f(-35, 25, 25))
fill = UsdLux.DistantLight.Define(stage, "/World/PreviewFill")
fill.CreateIntensityAttr(300.0)
fill.CreateAngleAttr(20.0)
UsdGeom.Xformable(fill.GetPrim()).AddRotateXYZOp().Set(Gf.Vec3f(50, -40, -120))
curve = UsdGeom.BasisCurves.Define(stage, "/World/WristReferencePath")
curve.CreateTypeAttr("linear")
curve.CreateCurveVertexCountsAttr([len(wrist_position)])
curve.CreatePointsAttr([Gf.Vec3f(*map(float, p)) for p in wrist_position])
curve.CreateWidthsAttr([0.002])
curve.SetWidthsInterpolation("constant")
curve.CreateDisplayColorAttr([Gf.Vec3f(0.1, 0.8, 1.0)])
center = (wrist_position.min(axis=0) + wrist_position.max(axis=0)) / 2
eye = center + np.array([1.15, -1.35, 0.95])
normal = (eye - center) / np.linalg.norm(eye - center)
right = np.cross(normal, np.array([0.0, 0.0, 1.0]))
right /= np.linalg.norm(right)
up = np.cross(normal, right)
backdrop_center = center - 2.0 * normal
backdrop = UsdGeom.Mesh.Define(stage, "/World/PreviewBackdrop")
backdrop.CreatePointsAttr(
[
Gf.Vec3f(*map(float, backdrop_center + 5.0 * (x * right + y * up)))
for x, y in [(-1, -1), (1, -1), (1, 1), (-1, 1)]
]
)
backdrop.CreateFaceVertexCountsAttr([4])
backdrop.CreateFaceVertexIndicesAttr([0, 1, 2, 3])
backdrop.CreateSubdivisionSchemeAttr("none")
backdrop.CreateDoubleSidedAttr(True)
# Display-only backdrop: no collision/rigid-body API, constant emission
# separates background brightness from the hand's environment lighting.
material = UsdShade.Material.Define(stage, "/World/PreviewBackdropMaterial")
shader = UsdShade.Shader.Define(stage, "/World/PreviewBackdropMaterial/Surface")
shader.CreateIdAttr("UsdPreviewSurface")
shader.CreateInput("diffuseColor", Sdf.ValueTypeNames.Color3f).Set(Gf.Vec3f(0.0))
shader.CreateInput("emissiveColor", Sdf.ValueTypeNames.Color3f).Set(Gf.Vec3f(0.055, 0.065, 0.08))
shader.CreateInput("useSpecularWorkflow", Sdf.ValueTypeNames.Int).Set(1)
shader.CreateInput("specularColor", Sdf.ValueTypeNames.Color3f).Set(Gf.Vec3f(0.0))
shader.CreateInput("roughness", Sdf.ValueTypeNames.Float).Set(1.0)
material.CreateSurfaceOutput().ConnectToSource(shader.ConnectableAPI(), "surface")
UsdShade.MaterialBindingAPI.Apply(backdrop.GetPrim()).Bind(material)
assert not backdrop.GetPrim().HasAPI(UsdPhysics.CollisionAPI)
assert not backdrop.GetPrim().HasAPI(UsdPhysics.RigidBodyAPI)
return tuple(eye), tuple(center)
@@ -7,6 +7,8 @@ from pathlib import Path
import h5py
import numpy as np
from .identity import manifest_side
class ContractError(ValueError):
"""An input cannot be safely interpreted as a reference trajectory."""
@@ -70,8 +72,9 @@ def load(path: str | Path, manifest: dict | None = None) -> Demonstrations:
):
require(name in file.attrs, f"missing root attribute {name}")
attributes[name] = text(file.attrs[name], name)
for name, expected in (("schema_version", "l20_tracking_v1"), ("embodiment", "L20"), ("hand_side", "left")):
for name, expected in (("schema_version", "l20_tracking_v1"), ("embodiment", "L20")):
require(attributes[name] == expected, f"{name}: expected {expected}")
require(attributes["hand_side"] in ("left", "right"), "hand_side must be left or right")
require(
re.fullmatch(r"[0-9a-f]{64}", attributes["asset_sha256"]) is not None,
"asset_sha256: lowercase SHA-256 required",
@@ -150,6 +153,7 @@ def load(path: str | Path, manifest: dict | None = None) -> Demonstrations:
def validate_against_manifest(data: Demonstrations, manifest: dict):
"""Check identity/order/limits and known URDF mimic equations, not actuator validity."""
require(manifest.get("manifest_version") == "l20_asset_manifest_v1", "unsupported manifest")
require(data.metadata["hand_side"] == manifest_side(manifest), "hand_side mismatch")
require(data.metadata["asset_sha256"] == manifest["asset_sha256"], "asset bundle hash mismatch")
require(data.metadata["root_link"] == manifest["root_link"], "root link mismatch")
joints = manifest["joints"]
+48
View File
@@ -4,6 +4,54 @@ Changelog
Unreleased
~~~~~~~~~~
0.1.2 (2026-09-14)
~~~~~~~~~~~~~~~~~~
Local bilateral tracking and bounded expert-reference milestone, not full release
clearance. Right USDs and expert data are local-only and NOT bundled: clean-clone
right-asset tests/replay remain blocked pending storage/licensing resolution.
Full pre-commit is unavailable; consolidated GUI E2E has not been run.
Cartpole and its task ID are unchanged; older prepared overlays need regeneration.
Added
^^^^^
* Add opt-in ``--gui`` to the existing replay runner, with the approved bright-hand
lighting, cyan wrist path, fixed camera and dark non-physical backdrop in a small
presentation helper. Keep default headless execution and 240Hz controls unchanged;
render every eight physics steps only in GUI mode. Add CPU display/flag regressions.
The separate GUI clone was visually approved; this consolidated CLI is NOT_RUN in Kit.
* Add explicit workspace-radius override and finite full-episode HDF5 replay
(at most 32000 steps per repetition), retaining the default 0.1m/1200-step bounds
and all controller/safety thresholds. Add strict duration and preservation tests.
* Add identity-validated, no-overwrite ``stretch-time`` preparation that preserves
original samples, geometry, coordinate metadata and expert provenance; record
the factor and input SHA256. No generic asset-hash override or original-speed claim.
* Complete one separately authorized 600s-budget recovery replay: right expert
reference uniformly slowed 80x, full 2x32000 steps at 240Hz, exit0/PASS in 384s;
81 CPU regressions pass. Preserve the interrupted run as unknown and optional
interpolated-grid equality audit failures; no runtime safety thresholds relaxed.
* Convert the user-provided L20 RIGHT URDF with the specified Isaac Lab converter;
record source provenance and right manifest for its self-contained local USD bundle
(generated USD files are excluded from this commit).
Geometry exceeds the ordinary-Git 2000KB gate: publication remains blocked
pending approved LFS/external storage and redistribution licensing confirmation.
* Extend the existing tracking contract with explicit right-side identity and its
inspected thumb mimic coefficient 1.03; preserve legacy left data/manifest support.
Add cross-side, axis, relocation and strict importer-topology regressions.
* Validate right small 2x480 and synthetic real-file HDF5 2x960 steps, plus left
small 2x480 regression, with unchanged controls and assertions. Those earlier
synthetic runs are not expert-data evidence; the separate slowed expert run is above.
Fixed
^^^^^
* Handle the exact right importer static default-Xform anchor and Geometry root
site. Migrate source-present Newton articulation API too, because its registered
schema inherits PhysicsArticulationRootAPI; preserve resolved source properties.
Check exactly one effective runtime root. Schema context now includes Newton
articulation registration; older prepared overlays must be regenerated.
0.1.1 (2026-09-11)
~~~~~~~~~~~~~~~~~~
@@ -81,6 +81,29 @@ class ControlTests(unittest.TestCase):
with self.assertRaises(ContractError):
Limits(force=value)
def test_workspace_extension_does_not_relax_velocity_or_forces(self):
from dataclasses import replace
manifest = {
"manifest_version": "l20_asset_manifest_v1",
"asset_sha256": "a" * 64,
"root_link": "test",
"joints": [{"name": "a", "lower_rad": 0, "upper_rad": 1}],
}
episode = synthetic(manifest).episodes["demo_000000"]
episode.time[:] *= 80
episode.wrist_position[:, 0] *= 300
with self.assertRaisesRegex(ContractError, "outside workspace"):
validate_reference(episode, self.limits)
extended = replace(self.limits, workspace_radius=0.8)
validate_reference(episode, extended)
for key, value in vars(self.limits).items():
if key != "workspace_radius":
self.assertEqual(getattr(extended, key), value)
episode.time[:] /= 80
with self.assertRaisesRegex(ContractError, "too fast"):
validate_reference(episode, extended)
def test_reference_envelope_and_invalid_gap(self):
manifest = {
"manifest_version": "l20_asset_manifest_v1",
@@ -0,0 +1,86 @@
"""CPU-only regression for the approved display scene; never launch Kit."""
import unittest
import numpy as np
from dex_workbench_tracking.preview import create_preview
from pxr import Gf, Sdf, Usd, UsdGeom, UsdLux, UsdPhysics, UsdShade
class TrackingPreviewTests(unittest.TestCase):
def setUp(self):
self.stage = Usd.Stage.CreateInMemory()
UsdGeom.Xform.Define(self.stage, "/World")
hand = UsdGeom.Mesh.Define(self.stage, "/World/Hand")
UsdPhysics.RigidBodyAPI.Apply(hand.GetPrim())
UsdPhysics.CollisionAPI.Apply(hand.GetPrim())
UsdPhysics.ArticulationRootAPI.Apply(hand.GetPrim())
material = UsdShade.Material.Define(self.stage, "/World/Hand/Material")
UsdShade.MaterialBindingAPI.Apply(hand.GetPrim()).Bind(material)
self.path = np.array([[0.1, -0.2, 0.4], [0.7, 0.1, -0.2], [0.2, 0.3, 0.1]])
def hand_spec(self):
layer = Sdf.Layer.CreateAnonymous()
Sdf.CopySpec(self.stage.GetRootLayer(), "/World/Hand", layer, "/Hand")
return layer.ExportToString()
def test_preview_never_changes_robot_materials_or_physics(self):
before = self.hand_spec()
create_preview(self.stage, self.path)
self.assertEqual(self.hand_spec(), before)
for prim in self.stage.Traverse():
if str(prim.GetPath()).startswith("/World/Hand"):
continue
with self.subTest(prim=prim.GetPath()):
for api in (UsdPhysics.RigidBodyAPI, UsdPhysics.CollisionAPI, UsdPhysics.ArticulationRootAPI):
self.assertFalse(prim.HasAPI(api))
self.assertFalse(prim.IsA(UsdPhysics.Joint))
self.assertFalse(any(a.GetTimeSamples() for a in prim.GetAttributes()))
bindings = [p for p in self.stage.Traverse() if p.HasAPI(UsdShade.MaterialBindingAPI)]
self.assertEqual({str(p.GetPath()) for p in bindings}, {"/World/Hand", "/World/PreviewBackdrop"})
def test_lighting_and_linear_cyan_reference_match_approved_preview(self):
original = self.path.copy()
create_preview(self.stage, self.path)
np.testing.assert_array_equal(self.path, original)
dome = UsdLux.DomeLight(self.stage.GetPrimAtPath("/World/PreviewLight"))
self.assertEqual(dome.GetIntensityAttr().Get(), 700)
self.assertEqual(dome.GetColorAttr().Get(), Gf.Vec3f(1))
for name, intensity, angle, rotation in (
("PreviewKey", 800, 12, (-35, 25, 25)),
("PreviewFill", 300, 20, (50, -40, -120)),
):
light = UsdLux.DistantLight(self.stage.GetPrimAtPath("/World/" + name))
self.assertEqual(light.GetIntensityAttr().Get(), intensity)
self.assertEqual(light.GetAngleAttr().Get(), angle)
self.assertEqual(light.GetPrim().GetAttribute("xformOp:rotateXYZ").Get(), Gf.Vec3f(*rotation))
curve = UsdGeom.BasisCurves(self.stage.GetPrimAtPath("/World/WristReferencePath"))
self.assertEqual(curve.GetTypeAttr().Get(), "linear")
self.assertEqual(list(curve.GetCurveVertexCountsAttr().Get()), [3])
np.testing.assert_allclose(curve.GetPointsAttr().Get(), self.path, atol=1e-7)
np.testing.assert_allclose(curve.GetWidthsAttr().Get(), [0.002])
self.assertEqual(curve.GetWidthsInterpolation(), "constant")
self.assertEqual(list(curve.GetDisplayColorAttr().Get()), [Gf.Vec3f(0.1, 0.8, 1)])
def test_fixed_camera_and_dark_camera_facing_backdrop(self):
eye, center = create_preview(self.stage, self.path)
np.testing.assert_array_equal(center, (self.path.min(axis=0) + self.path.max(axis=0)) / 2)
np.testing.assert_allclose(np.array(eye) - center, [1.15, -1.35, 0.95])
normal = (np.array(eye) - center) / np.linalg.norm(np.array(eye) - center)
mesh = UsdGeom.Mesh(self.stage.GetPrimAtPath("/World/PreviewBackdrop"))
points = np.array(mesh.GetPointsAttr().Get())
np.testing.assert_allclose(points.mean(axis=0), np.array(center) - 2 * normal, atol=5e-7)
np.testing.assert_allclose((points - center) @ normal, -2, atol=5e-7)
self.assertEqual(list(mesh.GetFaceVertexCountsAttr().Get()), [4])
self.assertEqual(list(mesh.GetFaceVertexIndicesAttr().Get()), [0, 1, 2, 3])
self.assertEqual(mesh.GetSubdivisionSchemeAttr().Get(), "none")
self.assertTrue(mesh.GetDoubleSidedAttr().Get())
material, _ = UsdShade.MaterialBindingAPI(mesh.GetPrim()).ComputeBoundMaterial()
self.assertEqual(str(material.GetPath()), "/World/PreviewBackdropMaterial")
shader = UsdShade.Shader(self.stage.GetPrimAtPath("/World/PreviewBackdropMaterial/Surface"))
self.assertEqual(shader.GetIdAttr().Get(), "UsdPreviewSurface")
self.assertEqual(shader.GetInput("diffuseColor").Get(), Gf.Vec3f(0))
self.assertEqual(shader.GetInput("emissiveColor").Get(), Gf.Vec3f(0.055, 0.065, 0.08))
self.assertEqual(shader.GetInput("specularColor").Get(), Gf.Vec3f(0))
self.assertEqual(shader.GetInput("roughness").Get(), 1)
@@ -0,0 +1,129 @@
"""Stdlib prelaunch bounds; no Kit import or launch."""
import argparse
import ast
import importlib.util
import unittest
from pathlib import Path
spec = importlib.util.spec_from_file_location(
"track_l20_options", Path(__file__).resolve().parents[3] / "scripts/tracking/track_l20.py"
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
validate_replay_request = module.validate_replay_request
class TrackingReplayTests(unittest.TestCase):
def presentation_args(self, *options):
class ExplicitVisualizer(argparse.Action):
def __call__(self, parser, namespace, values, option_string=None):
setattr(namespace, self.dest, values)
namespace.visualizer_explicit = True
def add_launcher_args(parser):
# Match AppLauncher: explicit 'none' becomes None, not ['none'].
# The separate marker distinguishes disabling from omitting --viz.
parser.add_argument("--headless", action="store_true")
parser.add_argument(
"--visualizer",
"--viz",
type=lambda value: None if value == "none" else value.split(","),
action=ExplicitVisualizer,
)
parser = module.build_parser(add_launcher_args)
args = parser.parse_args(["hand.usda", "--manifest", "manifest.json", *options])
module.configure_presentation(args)
return args
def test_gui_opt_in_and_legacy_headless_defaults(self):
for options in ((), ("--headless",), ("--viz", "none")):
args = self.presentation_args(*options)
self.assertFalse(args.gui)
self.assertTrue(args.headless)
self.assertIn(args.visualizer, ("none", ["none"]))
self.assertEqual(args.steps, 480)
self.assertIsNone(args.workspace_radius)
self.assertFalse(args.full_episode)
for options in (("--gui",), ("--gui", "--viz", "kit"), ("--visualizer=kit", "--gui")):
args = self.presentation_args(*options)
self.assertTrue(args.gui)
self.assertFalse(args.headless)
self.assertEqual(args.visualizer, ["kit"])
self.assertEqual(args.steps, 480)
def test_gui_rejects_explicit_display_conflicts(self):
for options in (
("--gui", "--headless"),
("--headless", "--gui"),
("--gui", "--viz", "none"),
("--visualizer=none", "--gui"),
("--gui", "--viz", "newton"),
):
with self.subTest(options=options), self.assertRaisesRegex(ValueError, "conflicts"):
self.presentation_args(*options)
def test_explicit_none_is_not_an_omitted_visualizer(self):
omitted = self.presentation_args("--gui")
self.assertFalse(getattr(omitted, "visualizer_explicit", False))
disabled = self.presentation_args("--viz", "none")
self.assertTrue(disabled.visualizer_explicit)
self.assertTrue(disabled.headless)
with self.assertRaisesRegex(ValueError, "conflicts"):
module.configure_presentation(
argparse.Namespace(gui=True, headless=None, visualizer=None, visualizer_explicit=True)
)
def test_render_cadence_is_gui_only_and_does_not_change_physics_dt(self):
tree = ast.parse(Path(module.__file__).read_text())
steps = [
n
for n in ast.walk(tree)
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and ast.unparse(n.func) == "sim.step"
]
self.assertEqual(len(steps), 1)
expression = steps[0].keywords[0]
self.assertEqual(expression.arg, "render")
self.assertEqual(ast.unparse(expression.value), "args.gui and step % 8 == 0")
self.assertEqual(
[
ast.unparse(n.value)
for n in ast.walk(tree)
if isinstance(n, ast.Assign) and any(isinstance(t, ast.Name) and t.id == "dt" for t in n.targets)
],
["1 / 240"],
)
def test_full_episode_duration_rejects_partial_off_grid_and_extrapolation(self):
validate = module.validate_replay_duration
validate(480, 1 / 240, 2.0, False)
validate(480, 1 / 240, 4.0, False) # Legacy bounded partial diagnostic.
validate(32000, 1 / 240, (50 / 30) * 80, True)
for duration in (134.0, 32000 / 240 + 0.001):
with self.subTest(duration=duration), self.assertRaisesRegex(ValueError, "exact on-grid"):
validate(32000, 1 / 240, duration, True)
with self.assertRaisesRegex(ValueError, "exact on-grid"):
validate(480, 1 / 240, 4.0, True)
for full in (False, True):
for duration in (1.0, float("nan")):
with self.subTest(full=full, duration=duration), self.assertRaisesRegex(ValueError, "shorter"):
validate(480, 1 / 240, duration, full)
def test_finite_launch_bounds_and_full_episode_opt_in(self):
validate_replay_request(480, False, False, None)
validate_replay_request(1200, False, True, None)
validate_replay_request(32000, True, True, 0.8)
for args in (
(32000, False, True, 0.8),
(1201, False, True, None),
(32001, True, True, 0.8),
(99, False, True, 0.8),
(480, True, False, 0.8),
(float("inf"), True, True, 0.8),
):
with self.subTest(args=args), self.assertRaises(ValueError):
validate_replay_request(*args)
for value in (0, -1, float("nan"), float("inf")):
with self.subTest(radius=value), self.assertRaises(ValueError):
validate_replay_request(480, False, True, value)
@@ -0,0 +1,150 @@
"""Actual right asset, bilateral identity and preservation regressions; no Kit."""
import copy
import json
import shutil
import tempfile
import unittest
from pathlib import Path
import numpy as np
from dex_workbench_tracking.asset import dependencies, inspect
from dex_workbench_tracking.cli import publish_validated, synthetic
from dex_workbench_tracking.control import Limits, validate_reference
from dex_workbench_tracking.identity import manifest_side
from dex_workbench_tracking.prepared import inspect_prepared, prepare
from dex_workbench_tracking.trajectory import load, validate_against_manifest
from pxr import Gf, Usd, UsdGeom, UsdPhysics
ASSETS = Path(__file__).resolve().parents[3] / "assets/robots/dex_hand"
class TrackingRightTests(unittest.TestCase):
def setUp(self):
self.right = json.loads((ASSETS / "linkerhand_g20_right/tracking_manifest.json").read_text())
self.left = json.loads((ASSETS / "linkerhand_g20_left/tracking_manifest.json").read_text())
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.directory = Path(self.temp.name)
self.bundle = self.directory / "bundle"
shutil.copytree(ASSETS / "linkerhand_g20_right/linkerhand_g20_right", self.bundle / "original")
self.source = self.bundle / "original/linkerhand_g20_right.usda"
self.output = self.bundle / "tracking.usda"
def test_right_profiles_roundtrip_limits_and_source_mimic(self):
for profile, count in (("small", 121), ("range_4s", 241)):
data = synthetic(self.right, profile)
self.assertEqual(data.metadata["hand_side"], "right")
self.assertEqual(data.metadata["provenance"], "synthetic")
path = self.directory / (profile + ".hdf5")
publish_validated(path, data, self.right)
episode = load(path, self.right).episodes["demo_000000"]
self.assertEqual(len(episode.time), count)
validate_reference(episode, Limits())
names = list(data.joint_names)
np.testing.assert_allclose(
episode.joint_position[:, names.index("thumb_ip")],
1.03 * episode.joint_position[:, names.index("thumb_mcp")],
atol=1e-7,
)
with self.assertRaises(FileExistsError):
publish_validated(path, data, self.right)
def test_cross_side_rejected_even_with_forged_hash_and_root(self):
for source, target in ((self.left, self.right), (self.right, self.left)):
data = synthetic(source)
data.metadata["asset_sha256"] = target["asset_sha256"]
data.metadata["root_link"] = target["root_link"]
with self.assertRaisesRegex(ValueError, "hand_side"):
validate_against_manifest(data, target)
def test_right_identity_cannot_be_missing_or_relabelled(self):
for side in (None, "left", "both"):
wrong = copy.deepcopy(self.right)
if side is None:
del wrong["hand_side"]
else:
wrong["hand_side"] = side
with self.assertRaises(ValueError):
manifest_side(wrong)
wrong = copy.deepcopy(self.left)
wrong["hand_side"] = "right"
with self.assertRaises(ValueError):
manifest_side(wrong)
self.assertEqual(manifest_side(self.left), "left")
def test_actual_right_axes_limits_and_relocated_dependencies(self):
actual = inspect(self.source)
self.assertEqual(actual["asset_sha256"], self.right["asset_sha256"])
self.assertEqual(actual["joints"], self.right["joints"])
self.assertEqual(len(actual["bodies"]), 22)
self.assertEqual(len(actual["joints"]), 21)
for dep in actual["dependencies"]:
self.assertTrue((self.source.parent / dep["path"]).resolve().is_relative_to(self.bundle))
stage = Usd.Stage.Open(str(self.source))
for joint in actual["joints"]:
name = joint["name"]
expected = (0, 0, -1) if name == "thumb_cmc_yaw" else ((1, 0, 0) if name.endswith("roll") else (0, 1, 0))
rotation = stage.GetPrimAtPath(joint["path"]).GetAttribute("physics:localRot1").Get()
axis = Gf.Vec3d(*(float(joint["axis"] == c) for c in "XYZ"))
np.testing.assert_allclose(Gf.Rotation(Gf.Quatd(rotation)).TransformDir(axis), expected, atol=1e-6)
before = dependencies(self.source)
result = prepare(self.source, self.output, self.right)
self.assertEqual(len(result["independent_joint_names"]), 16)
self.assertEqual(result["mimic"][-1]["multiplier"], 1.03)
moved = self.directory / "moved"
shutil.copytree(self.bundle, moved)
self.assertEqual(result, inspect_prepared(moved / "tracking.usda", self.right))
self.assertEqual(before, dependencies(self.source))
def test_modified_static_endpoint_transform_api_reset_or_target_rejected(self):
for change in ("transform", "sample", "rigid", "reset", "target"):
with self.subTest(change=change):
stage = Usd.Stage.Open(str(self.source))
stage.SetEditTarget(stage.GetRootLayer())
prim = stage.GetDefaultPrim()
if change == "transform":
UsdGeom.Xformable(prim).AddTranslateOp().Set(Gf.Vec3d(1, 0, 0))
elif change == "sample":
UsdGeom.Xformable(prim).AddTranslateOp().Set(Gf.Vec3d(0), Usd.TimeCode(1))
elif change == "rigid":
UsdPhysics.RigidBodyAPI.Apply(prim)
elif change == "reset":
UsdGeom.Xformable(prim).SetResetXformStack(True)
else:
anchor = stage.GetPrimAtPath(self.right["world_fixed_joints"][0]["path"])
UsdPhysics.Joint(anchor).GetBody0Rel().SetTargets([self.right["default_prim"] + "/Geometry"])
stage.GetRootLayer().Save()
with self.assertRaises(ValueError):
inspect(self.source)
# Only restore this test's private copy; reload to clear USD cache.
shutil.copyfile(
ASSETS / "linkerhand_g20_right/linkerhand_g20_right/linkerhand_g20_right.usda", self.source
)
stage.Reload()
def test_overlay_rejects_reintroduced_newton_root_and_changed_container(self):
prepare(self.source, self.output, self.right)
stage = Usd.Stage.Open(str(self.output))
site = stage.GetPrimAtPath(self.right["articulation_roots"][0])
site.AddAppliedSchema("NewtonArticulationRootAPI")
stage.GetRootLayer().Save()
with self.assertRaises(ValueError):
inspect_prepared(self.output, self.right)
def test_source_container_must_be_exact_geometry_parent(self):
stage = Usd.Stage.Open(str(self.source))
site = stage.GetPrimAtPath(self.right["articulation_roots"][0])
site.RemoveAPI(UsdPhysics.ArticulationRootAPI)
site.RemoveAppliedSchema("NewtonArticulationRootAPI")
wrong = stage.DefinePrim(self.right["default_prim"] + "/Wrong", "Xform")
UsdPhysics.ArticulationRootAPI.Apply(wrong)
stage.GetRootLayer().Save()
manifest = dict(inspect(self.source), hand_side="right", source_urdf=self.right["source_urdf"])
with self.assertRaisesRegex(ValueError, "container"):
prepare(self.source, self.output, manifest)
if __name__ == "__main__":
unittest.main()
@@ -16,6 +16,11 @@ ASSET = REPO / "assets/robots/dex_hand/linkerhand_g20_left"
class TrackingSchemaContextTests(unittest.TestCase):
def test_registered_schema_preparation_preserves_defaults_and_rejects_context_drift(self):
for side in ("left", "right"):
with self.subTest(side=side):
self.check_registered(REPO / f"assets/robots/dex_hand/linkerhand_g20_{side}", side)
def check_registered(self, asset, side):
installation = os.environ.get("ISAAC_PATH")
if not installation:
self.skipTest("ISAAC_PATH required for installed PhysX/Newton CPU schema regression")
@@ -24,8 +29,8 @@ class TrackingSchemaContextTests(unittest.TestCase):
self.assertEqual(len(physx), 1, "Select an unambiguous installed PhysX schema for this test")
newton = installation / "exts/omni.usd.schema.newton/usd/schema/newton/newton_usd_schemas/plugInfo.json"
self.assertTrue(newton.is_file())
manifest = json.loads((ASSET / "tracking_manifest.json").read_text())
source = ASSET / "linkerhand_g20_left/linkerhand_g20_left.usda"
manifest = json.loads((asset / "tracking_manifest.json").read_text())
source = asset / f"linkerhand_g20_{side}/linkerhand_g20_{side}.usda"
with tempfile.TemporaryDirectory() as directory:
directory = Path(directory)
unregistered = directory / "unregistered.usda"
@@ -37,7 +42,7 @@ class TrackingSchemaContextTests(unittest.TestCase):
code = """
import json, sys
from pathlib import Path
from pxr import Usd
from pxr import Usd, UsdPhysics
from dex_workbench_tracking.asset import dependencies
from dex_workbench_tracking.prepared import inspect_prepared, register_schema_plugins
source, manifest_path, unregistered, registered, *plugins = sys.argv[1:]
@@ -64,17 +69,24 @@ assert all(result["schema_context"].values())
assert dependencies(source) == before
original = Usd.Stage.Open(source)
stage = Usd.Stage.Open(registered)
anchor = manifest["world_fixed_joints"][0]["path"]
site = manifest["articulation_roots"][0]
root = stage.GetPrimAtPath(manifest["root_body_path"])
for attr in original.GetPrimAtPath(anchor).GetAttributes():
if attr.GetName().startswith("physxArticulation:"):
for prim in (stage.GetPrimAtPath(anchor), root):
roots = [p for p in stage.Traverse() if p.HasAPI(UsdPhysics.ArticulationRootAPI)]
assert roots == [root], roots
if manifest.get("hand_side") == "right":
assert not stage.GetPrimAtPath(site).HasAPI(UsdPhysics.ArticulationRootAPI)
assert "NewtonArticulationRootAPI" not in stage.GetPrimAtPath(site).GetAppliedSchemas()
assert "NewtonArticulationRootAPI" in root.GetAppliedSchemas()
for attr in original.GetPrimAtPath(site).GetAttributes():
if attr.GetName().startswith("physxArticulation:") or attr.GetName() == "newton:selfCollisionEnabled":
for prim in (stage.GetPrimAtPath(site), root):
after = prim.GetAttribute(attr.GetName())
assert after.Get() == attr.Get(), attr.GetPath()
assert after.HasAuthoredValue(), attr.GetPath()
assert not after.GetTimeSamples(), attr.GetPath()
# Registered schema fallbacks do not relax the static/time-sample gate.
root.GetAttribute("physxArticulation:sleepThreshold").Set(99, Usd.TimeCode(1))
field = "newton:selfCollisionEnabled" if manifest.get("hand_side") == "right" else "physxArticulation:sleepThreshold"
root.GetAttribute(field).Set(True if manifest.get("hand_side") == "right" else 99, Usd.TimeCode(1))
stage.GetRootLayer().Save()
try:
inspect_prepared(registered, manifest)
@@ -82,7 +94,7 @@ except ValueError as error:
assert "Time samples prohibited" in str(error), error
else:
raise AssertionError("Sampled solver property accepted")
root.GetAttribute("physxArticulation:sleepThreshold").ClearAtTime(Usd.TimeCode(1))
root.GetAttribute(field).ClearAtTime(Usd.TimeCode(1))
stage.GetRootLayer().Save()
inspect_prepared(registered, manifest)
print("PASS: isolated registered schema author/inspect, exact default preservation, strict time samples")
@@ -93,7 +105,7 @@ print("PASS: isolated registered schema author/inspect, exact default preservati
"-c",
code,
str(source),
str(ASSET / "tracking_manifest.json"),
str(asset / "tracking_manifest.json"),
str(unregistered),
str(registered),
str(physx[0]),
@@ -11,7 +11,7 @@ from pathlib import Path
import h5py
import numpy as np
from dex_workbench_tracking.cli import publish_validated, synthetic, write
from dex_workbench_tracking.cli import publish_validated, stretch_time, synthetic, write
from dex_workbench_tracking.trajectory import ContractError, load, require_dynamic_replay_ready, sample
@@ -38,6 +38,68 @@ class TrackingTrajectoryTests(unittest.TestCase):
with h5py.File(self.path, "r+") as file:
function(file)
def test_stretch_time_preserves_samples_identity_and_input(self):
original = load(self.path, self.manifest)
original.metadata["provenance"] = "expert_retargeted"
original.world_from_source[:3, :3] = [[1, 0, 0], [0, 0, 1], [0, -1, 0]]
source_bytes = self.path.read_bytes()
result = stretch_time(original, 80)
self.assertEqual(result.metadata["provenance"], "expert_retargeted")
self.assertEqual(result.metadata["asset_sha256"], original.metadata["asset_sha256"])
self.assertEqual(result.joint_names, original.joint_names)
np.testing.assert_array_equal(result.world_from_source, original.world_from_source)
for name, episode in original.episodes.items():
stretched = result.episodes[name]
np.testing.assert_array_equal(stretched.time, episode.time * 80)
for field in ("wrist_position", "wrist_quaternion", "joint_position", "valid"):
np.testing.assert_array_equal(getattr(stretched, field), getattr(episode, field))
self.assertEqual(episode.time[-1], 2)
destination = Path(self.directory.name) / "stretched.hdf5"
publish_validated(destination, result, self.manifest)
self.assertEqual(load(destination, self.manifest).episodes["demo_000000"].time[-1], 160)
with self.assertRaises(FileExistsError):
publish_validated(destination, result, self.manifest)
self.assertEqual(self.path.read_bytes(), source_bytes)
for name, episode in original.episodes.items():
query = np.linspace(0, episode.time[-1], 401)
expected = sample(episode, query)
actual = sample(result.episodes[name], query * 80)
for field in ("wrist_position", "wrist_quaternion", "joint_position"):
np.testing.assert_allclose(getattr(actual, field), getattr(expected, field), rtol=0, atol=1e-7)
def test_stretch_time_rejects_nonfinite_acceleration_and_overflow(self):
original = load(self.path, self.manifest)
for factor in (0, -1, 0.5, float("nan"), float("inf"), np.finfo(float).max):
with self.subTest(factor=factor), self.assertRaises(ContractError):
stretch_time(original, factor)
def test_stretch_time_cli_identity_and_provenance(self):
manifest_path = Path(self.directory.name) / "manifest.json"
manifest_path.write_text(json.dumps(self.manifest))
destination = Path(self.directory.name) / "stretched.hdf5"
command = [
sys.executable,
"-m",
"dex_workbench_tracking.cli",
"stretch-time",
str(self.path),
"--manifest",
str(manifest_path),
"--factor",
"80",
"--output",
str(destination),
]
result = subprocess.run(command, capture_output=True, text=True)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("stretch input HDF5 SHA256=", load(destination, self.manifest).metadata["source_description"])
self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0)
bad_destination = Path(self.directory.name) / "bad.hdf5"
command[-1] = str(bad_destination)
self.edit(lambda file: file.attrs.__setitem__("asset_sha256", "b" * 64))
self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0)
self.assertFalse(bad_destination.exists())
def test_load_roundtrip(self):
data = load(self.path, self.manifest)
self.assertEqual(data.joint_names, ("finger_a", "finger_b"))
@@ -68,7 +130,7 @@ class TrackingTrajectoryTests(unittest.TestCase):
for name, bad in (
("schema_version", "v2"),
("embodiment", "L30"),
("hand_side", "right"),
("hand_side", "both"),
("provenance", "unknown"),
("asset_sha256", "xyz"),
("root_link", ""),
@@ -274,7 +336,7 @@ class TrackingTrajectoryTests(unittest.TestCase):
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(result.stdout)["asset_compatibility"], "NOT_CHECKED")
self.edit(lambda file: file.attrs.__setitem__("hand_side", "right"))
result = self.run_cli("validate", self.path)
result = self.run_cli("validate", self.path, "--manifest", manifest)
self.assertNotEqual(result.returncode, 0)
self.assertIn("FAIL", result.stderr)