Skip to content

Replay an episode

Play a recorded episode on the robot again: the arms ease from where they stand to where the episode starts, play it at its recorded pace or slower, and hold where it ended. This runs in the sim as it is, recording the episode it replays first.

"""Records an arm homing in the sim, then plays the episode back on it from where it stands."""

import tempfile
import time
from pathlib import Path

import anvil.apps
import anvil.control
import anvil.hw
import anvil.recorder
import anvil.sim
import anvil.workcell

workcell = anvil.workcell.WorkcellConfig.OPENYAM_V1_SINGLE
robot = workcell.robot
data_root = Path(tempfile.mkdtemp())

with anvil.sim.World(robot=robot, scene="lift") as world:
    hw = anvil.hw.SimHw(world=world)
    hw.enable()
    # A replay eases the arms in under the trajectory controller and plays joint angles through the joint controller.
    controllers = [
        anvil.control.JointTrajectoryController(robot=robot),
        anvil.control.JointController(robot=robot),
    ]
    with anvil.control.Runtime(hw=hw, controllers=controllers) as runtime:
        runtime.start()
        handle = runtime.controller(kind=anvil.control.JointTrajectoryController)
        runtime.activate(controller=handle)

        # The episode to replay: the arm homing.
        recorder = anvil.recorder.Recorder(workcell=workcell, runtime=runtime)
        recorder.start(session=anvil.recorder.Session(root=data_root))
        episode = recorder.path
        handle.play(trajectory=robot.home_trajectory(from_q=runtime.latest().q), timeout_s=30.0)
        recorder.stop()
        handle.play(trajectory=robot.dehome_trajectory(from_q=runtime.latest().q), timeout_s=30.0)

        # Poll the replay at a steady pace: make the controller it names active, and send what it says is due.
        replay = anvil.apps.EpisodeReplay(path=episode, robot=robot, mode=anvil.apps.ReplayMode.JOINT_ANGLES)
        print(f"replaying {replay.duration_s:.1f} s of {episode.name}")
        started_ns = time.monotonic_ns()
        active_kind = anvil.control.JointTrajectoryController
        while True:
            polled = replay.poll(elapsed_ns=time.monotonic_ns() - started_ns, q=runtime.latest().q)
            if polled.phase == anvil.apps.ReplayPhase.HOLDING:
                break
            if polled.controller is not active_kind:
                active_kind = polled.controller
                runtime.activate(controller=runtime.controller(kind=active_kind))
            if polled.command is not None:
                runtime.send_command(command=polled.command)
            time.sleep(0.005)
        print("the replay ended where the episode did")
  1. Open the episode with EpisodeReplay, for the robot that is running. It refuses an episode recorded on other arms, variants or grippers. Choose what to play back with ReplayMode: the joint angles the arms measured, the commands each controller was sent, or what the drives were told. speed below 1 plays it slower.
  2. Hold the controllers it names in the runtime: the joint trajectory controller for the approach, and the joint controller for the rest, or, replaying commands, every controller the episode was driven by.
  3. Poll it at a steady pace, with the time since the first poll and the measured joints. poll says which controller should be active and what command is due now: activate that controller when it changes, and send the command when there is one. Stop polling once the phase is HOLDING; the arms hold where the episode ended.

An episode recorded on real drives carries which ones in EpisodeReplay.hardware; compare it with the bench's Report.hardware to know whether this is the same physical robot. Commands replayed run under the settings in effect now, so other gains or IK parameters change what the arms do.