Rerun Helpers in the Artefacts Toolkit

Tags:

The Artefacts Toolkit Rerun Helpers provide convenient functions for tests that record a Rerun .rrd file during a simulation and then make their assertions against that recording, rather than against the simulator itself.

This splits a test into two independent halves. The simulator (or a logger node next to it) records while the test itself then asserts against the recording, so no ROS or simulator knowledge is needed to write the assertions.

Import with:

from artefacts_toolkit.rerun import recorder, reader, video

Functions

Function Reference

recorder.get_output_dir

Returns (and creates) the directory that recordings and other test outputs should be written to.

recorder.get_output_dir(
    directory=None
)

Parameters

Parameter Type Description Default
directory str or Path Explicit directory to use instead of the defaults None

Returns

Path: The absolute path of the output directory.

Example

The same directory is resolved on the simulator side and on the test side, so both agree on where the recording is:

from artefacts_toolkit.rerun import recorder

OUTPUT_FOLDER = recorder.get_output_dir()
rrd_path = OUTPUT_FOLDER / "my-rerun-recording.rrd"

recorder.start_recording

Starts a Rerun recording saved to directory/filename and returns the recording stream and the file path.

recorder.start_recording(
    application_id,
    filename="recording.rrd",
    directory=None,
    handle_sigterm=True
)

Parameters

Parameter Type Description Default
application_id str Name of the application shown in the Rerun viewer Required
filename str or Path Name of the .rrd file. May also be an absolute path, in which case directory is ignored "recording.rrd"
directory str or Path Directory to save to, resolved like get_output_dir None
handle_sigterm bool Close the recording and exit cleanly when the process receives SIGTERM True

Returns

tuple: Returns a tuple containing:

  • recording (rerun.RecordingStream): The started recording. It is also made the global recording, so plain rr.log(...) and rr.set_time(...) calls go to it.
  • path (Path): Path to the .rrd file being written.

Example

In the following example the simulator records the base pose and the joystick commands on a sim_time timeline, and is stopped by the test harness once the mission ends. Setting your own timeline with rr.set_time is what lets the test later line data up by simulation time rather than wall-clock time.

import rerun as rr
import simulator as sim
from artefacts_toolkit.rerun import recorder


class RerunSimulator(sim.SimulatorMujoco):
    def _set_sim_time(self):
        rr.set_time("sim_time", duration=float(self.mujoco_data.time))

    def _log_joy(self, joy):
        self._set_sim_time()
        rr.log("cmd/fwd", rr.Scalars(joy.axes[1]))
        rr.log("cmd/yaw", rr.Scalars(joy.axes[2]))

    def _read_state(self):
        super()._read_state()
        q = self.mujoco_data.qpos  # [0:3] base xyz, [3:7] base quat wxyz
        w, x, y, z = q[3:7]
        self._set_sim_time()
        rr.log("base", rr.Transform3D(translation=q[0:3],
                                      quaternion=rr.Quaternion(xyzw=[x, y, z, w])))
        rr.log("base/yaw", rr.Scalars(yaw_from_quat(w, x, y, z)))


def main():
    robot = sim.Robot(sim.RobotType.Tron2, True)
    robot.init("127.0.0.1")

    # After robot.init(): the SDK resets signal handlers
    recorder.start_recording("tron2_loop", "recording-loop.rrd")

    RerunSimulator(...).run()

reader.load

Loads a recording into memory once, so several assertions can share it without re-reading the file.

reader.load(
    rrd
)

Parameters

Parameter Type Description Default
rrd str or Path Path to the .rrd file Required

Returns

ChunkStore: The loaded recording. Every other reader and video function accepts it in place of the path.

Example

from artefacts_toolkit.rerun import reader


@pytest.fixture(scope="module")
def recording(recording_path):
    assert recording_path.exists(), f"Recording not found at {recording_path}"
    return reader.load(recording_path)

reader.get_entity_paths

Returns all entity paths logged in the recording.

reader.get_entity_paths(
    rrd
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required

Returns

list[str]: The entity paths, for example ["/base", "/base/yaw", "/cmd/fwd"].

Example

assert "/bodies/uwb_tag" in reader.get_entity_paths(recording), "target was never logged"

reader.get_timelines

Returns the names of the timelines in the recording.

reader.get_timelines(
    rrd
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required

Returns

list[str]: The timeline names, for example ["log_time", "sim_time"].

reader.get_columns

Returns the data column names of the recording, optionally only those of one entity.

reader.get_columns(
    rrd,
    entity=None
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
entity str Only list the columns of this entity None

Returns

list[str]: Column names in the "/entity:Archetype:field" form.

Example

>>> reader.get_columns(recording, "/base")
['/base:Transform3D:quaternion', '/base:Transform3D:translation']

reader.get_column

Returns the times and values of one column, ordered by a timeline.

reader.get_column(
    rrd,
    column,
    timeline=None
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
column str Column name, e.g. "/base/yaw:Scalars:scalars" Required
timeline str Timeline to order by. Defaults to the recording’s own timeline (the first one that is not log_time), else log_time None

Returns

tuple: Returns a tuple containing:

  • times (numpy.ndarray): Float seconds for duration timelines, datetime64 for timestamp timelines, integers for sequence timelines.
  • values (numpy.ndarray): Shape (n,) for scalars and (n, 3) for 3-vectors. When rows hold a varying number of instances (e.g. point clouds), an object array of per-row arrays.

Example

times, yaw = reader.get_column(recording, "/base/yaw:Scalars:scalars")
assert times[-1] > 60.0, "mission ended early"
assert np.abs(yaw).max() <= np.pi

reader.get_final_message

Retrieves the last value logged to a column, or the value of a static column.

reader.get_final_message(
    rrd,
    column
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
column str Column name, e.g. "/base:Transform3D:translation" Required

Returns

Any: The value, unwrapped: a float for Scalars, a length-3 array for a translation, an (n, 3) array for a Points3D with several points.

Example

Waypoints are logged once (static) as Points3D, and the test measures how far the robot strays from the track they describe:

wp_col = "/network/mission_waypoints:Points3D:positions"
assert wp_col in reader.get_columns(recording), f"Could not find {wp_col} in recording"

track = np.vstack(reader.get_final_message(recording, wp_col))[:, :2]  # N x 2

reader.get_message_count

Returns the number of log calls made to an entity.

reader.get_message_count(
    rrd,
    entity
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
entity str Entity path, e.g. "/cam" Required

Returns

int: The number of (non-static) rows logged to the entity, 0 if it does not exist.

Example

assert reader.get_message_count(recording, "/cam") >= 100, "camera stopped publishing"

reader.to_dataframe

Returns the recording as a pandas DataFrame, one row per time on the chosen timeline.

Different entities are usually logged at different rates: a pose every simulation step, a joystick command only when it changes. With step, every selected entity is resampled onto one uniform time grid and, with fill_latest_at, gaps are filled with the latest earlier value, so a sparse command topic lines up row by row with a dense pose stream and the two can be compared directly.

reader.to_dataframe(
    rrd,
    contents=None,
    index=None,
    step=None,
    fill_latest_at=True
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
contents str or list[str] Entities to include: "/base" for one entity, "/base/**" for a subtree, or a list of those. All entities when None None
index str Timeline to use as the row index. Defaults like get_column None
step float Resample onto a uniform grid of this many seconds, from the first to the last time None
fill_latest_at bool Fill gaps with the latest earlier value of each column True

Returns

pandas.DataFrame: The timeline column comes first (float seconds for duration timelines), followed by one column per "/entity:Archetype:field". Static data is joined onto every row. Scalars columns are plain float64 (NaN where nothing was logged); vectors are arrays.

Example

@pytest.fixture(scope="module")
def df(recording_path):
    """Whole recording resampled onto a uniform 20 ms sim-time grid."""
    frame = reader.to_dataframe(recording_path, ["/base/**", "/cmd/**"], index="sim_time", step=0.02)

    out = frame.dropna(subset=["/base:Transform3D:translation"]).reset_index(drop=True)
    out["x"] = [t[0] for t in out["/base:Transform3D:translation"]]
    out["y"] = [t[1] for t in out["/base:Transform3D:translation"]]
    out["yaw"] = out["/base/yaw:Scalars:scalars"]
    out["cmd_fwd"] = out["/cmd/fwd:Scalars:scalars"].fillna(0.0)
    return out


def test_straights_cover_side_length(df, straights):
    for i, (a, b) in enumerate(straights):
        dist = math.hypot(df["x"][b - 1] - df["x"][a], df["y"][b - 1] - df["y"][a])
        assert abs(dist - SIDE_LENGTH) < DIST_TOLERANCE

video.extract_video

Creates an MP4 video from a camera entity in the recording.

video.extract_video(
    rrd,
    entity,
    output_path,
    timeline=None,
    frame_rate=20
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
entity str Entity path of the camera, e.g. "/cam" Required
output_path str or Path Path where the video will be saved (.mp4) Required
timeline str Timeline to order frames by. Defaults like get_column None
frame_rate int Frame rate used when encoding rr.Image frames 20

Returns

Path: The path of the saved video.

Example

from artefacts_toolkit.rerun import recorder, video

video.extract_video(recording, "/cam", recorder.get_output_dir() / "head_camera.mp4")

video.extract_camera_image

Saves the last image logged to a camera entity as a PNG.

video.extract_camera_image(
    rrd,
    entity,
    output_dir="output"
)

Parameters

Parameter Type Description Default
rrd str, Path or ChunkStore Path to the recording, or a loaded one Required
entity str Entity path of the camera, e.g. "/cam" Required
output_dir str or Path Directory where to save the extracted image "output"

Returns

Path: The path of the saved image, output_dir/<entity>.last.png with / replaced by _ (for example output/_cam.last.png).

Example

video.extract_camera_image(recording, "/cam", recorder.get_output_dir())
Last modified September 18, 2026: Update cli reference (#153) (c83f1b9)