Python API

core.skill_registry

register_skill

Decorator that registers an atomic skill class with its required affordances and execution phase.

>>> from core.skill_registry import register_skill, SKILL_REGISTRY
>>>
>>> @register_skill(
...     name="pure_rotate",
...     phase="continuation",
...     requires=["rotatable"],
... )
... class PureRotate(Skill):
...     def step(self, env, part): ...

SKILL_REGISTRY

Module-level singleton; the lookup tables used by task graphs and the asset onboarding pipeline.

>>> SKILL_REGISTRY.applicable_skills("100221", part="lid")
['grasp_part', 'pure_rotate', 'lift_lid']
>>> SKILL_REGISTRY.get("pure_rotate").requires
['rotatable']

core.predicates

compile_predicate

Parse a predicate-DSL string into a callable that takes an env state and returns bool.

>>> from core.predicates import compile_predicate
>>> pred = compile_predicate('grasped("lid") and joint_value("lid", ">", 0.5)')
>>> pred(env.state)
True

utils.task_graph

TaskGraph

A parsed compositional task: list of nodes (skill + asset + part) plus a goal predicate. Built from YAML.

>>> from utils.task_graph import TaskGraph
>>> graph = TaskGraph.from_yaml("configs/lid_then_rotate.yaml")
>>> graph.nodes
['grasp_part', 'lift_lid', 'pure_rotate', 'release_gripper']

run_task_graph

Step a task graph against a policy. Returns a per-stage success record.

>>> from utils.task_graph import run_task_graph
>>> result = run_task_graph(graph, env, model, predict, max_steps=400)
>>> result.stage_success
{'grasp_part': True, 'lift_lid': True, 'pure_rotate': False}

utils.eval_metrics

EpisodeResult

Per-episode dataclass: stage_success, smoothness arrays, final predicate state.

>>> from utils.eval_metrics import EpisodeResult
>>> ep = EpisodeResult(stage_success={"grasp_part": True}, jerk_rms=12.4, vel_var=0.02)

EvalSummary

Aggregates a list of EpisodeResults into the final results.json with the three top-level keys.

>>> from utils.eval_metrics import EvalSummary
>>> summary = EvalSummary.from_episodes(episodes)
>>> summary.to_dict().keys()
dict_keys(['understanding', 'perception', 'behavior'])

compute_smoothness

Given a joint trajectory, return {jerk_rms, vel_var, path_length}.

>>> from utils.eval_metrics import compute_smoothness
>>> compute_smoothness(joint_traj, dt=1/30.)
{'jerk_rms': 12.4, 'vel_var': 0.018, 'path_length': 2.31}

utils.eval_sweep

dr_sweep

Run a policy across a list of DR cells and return per-axis AUSC scores.

>>> from utils.eval_sweep import dr_sweep, standard_dr_sweeps
>>> ausc = dr_sweep(env, model, predict, sweeps=standard_dr_sweeps(), episodes=20)
>>> ausc
{'lighting': 0.78, 'view': 0.64, 'jitter': 0.71}

standard_dr_sweeps

The canonical lighting / view / jitter axis specification used in the MetaFine report.

utils.eval_setup

make_eval_env

Construct a SAPIEN env from a task-graph YAML, ready to be stepped against a policy.

>>> from utils.eval_setup import make_eval_env
>>> env = make_eval_env("configs/peg_in_hole.yaml", seed=0)
>>> obs, info = env.reset()