# Rerun Helpers in the Artefacts Toolkit

Site [llms.txt](/llms.txt)

---

The Artefacts Toolkit Rerun Helpers provide convenient functions for tests that record a [Rerun](https://rerun.io) `.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:
```python
from artefacts_toolkit.rerun import recorder, reader, video
```

> [!NOTE]
> * The helpers wrap the parts of `rerun-sdk` that change between releases (recording sinks, the dataframe/query API), so test code written against them keeps working across Rerun upgrades.
> * Data columns are named `"/entity:Archetype:field"`, for example `"/base:Transform3D:translation"` or `"/base/yaw:Scalars:scalars"`. Use [`reader.get_columns`](#readerget_columns) to list the columns in a recording.
> * Every `reader` and `video` function accepts either the path to a `.rrd` file or the object returned by [`reader.load`](#readerload).

## Functions
* [`recorder.get_output_dir`](#recorderget_output_dir)
* [`recorder.start_recording`](#recorderstart_recording)
* [`reader.load`](#readerload)
* [`reader.get_entity_paths`](#readerget_entity_paths)
* [`reader.get_timelines`](#readerget_timelines)
* [`reader.get_columns`](#readerget_columns)
* [`reader.get_column`](#readerget_column)
* [`reader.get_final_message`](#readerget_final_message)
* [`reader.get_message_count`](#readerget_message_count)
* [`reader.to_dataframe`](#readerto_dataframe)
* [`video.extract_video`](#videoextract_video)
* [`video.extract_camera_image`](#videoextract_camera_image)

## Function Reference

### `recorder.get_output_dir`

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

```python
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.

> [!NOTE]
> * When `directory` is not given, the `ARTEFACTS_SCENARIO_UPLOAD_DIR` environment variable is used. It is set by `artefacts run`, and everything in it is uploaded to the Artefacts Dashboard when the scenario finishes.
> * When neither is available (for example running `pytest` locally), `./test_outputs` is used.

#### Example

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

```python
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.

```python
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`](#recorderget_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.

> [!NOTE]
> * A stale file of the same name is removed before recording starts.
> * Test harnesses usually stop the simulator with `Popen.terminate()` (SIGTERM). With `handle_sigterm`, the recording is closed first, so the file is complete even if the harness then kills the process during slower cleanup such as closing video encoders.
> * Some native libraries reset signal handlers when they initialise (limxsdk's `Robot.init` does). Call `start_recording` after them.
> * Recordings are automatically uploaded to the Artefacts Dashboard unless you use the `--no-upload` flag with `artefacts run`.

#### 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.

```python
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.

```python
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.

> [!NOTE]
> * Files written by older Rerun versions, or by a process that was killed before finishing, are read as well.

#### Example

```python
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.

```python
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

```python
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.

```python
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"]`.

> [!NOTE]
> * `log_time` is always present: Rerun adds it automatically. The timeline you set yourself with `rr.set_time` is the one the other helpers use by default.

---

### `reader.get_columns`

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

```python
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

```python
>>> 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.

```python
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.

> [!NOTE]
> * Raises `KeyError` when the column does not exist.

#### Example

```python
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.

```python
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:

```python
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.

```python
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

```python
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.

```python
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`](#readerget_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.

> [!NOTE]
> * Without `step`, there is one row per time at which any of the selected entities was logged.
> * Vector cells are numpy arrays. `np.vstack(df[column])` turns a translation column into an `(n, 3)` array.

#### Example

```python
@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.

```python
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`](#readerget_column) | `None`      |
| `frame_rate`  | `int`                         | Frame rate used when encoding `rr.Image` frames                             | `20`        |

#### Returns

`Path`: The path of the saved video.

> [!NOTE]
> * Works for entities logged as `rr.Image` frames (encoded with H.264 at `frame_rate`) and for `rr.VideoStream` entities with H.264 samples (remuxed as-is, keeping the recording's timestamps).
> * The `.mp4` extension is added if not present in `output_path`.
> * By saving to the directory returned by [`recorder.get_output_dir`](#recorderget_output_dir), the video is automatically uploaded to the Artefacts Dashboard.

#### Example

```python
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.

```python
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`).

> [!NOTE]
> * Only `rr.Image` entities with 8-bit L, RGB, RGBA, BGR or BGRA pixels are supported.

#### Example

```python
video.extract_camera_image(recording, "/cam", recorder.get_output_dir())
```