# Artefacts ツールキットの Rerun ヘルパー

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

---

Artefacts ツールキットの Rerun ヘルパーは、シミュレーション中に [Rerun](https://rerun.io) の `.rrd` ファイルを記録し、シミュレータ自体ではなくその記録に対してアサーションを行うテストのための便利な関数を提供します。

これによりテストは独立した 2 つの半分に分かれます。シミュレータ（またはその隣のロガーノード）が記録を行い、テスト自体はその記録に対してアサーションを行うため、アサーションを書くのに ROS やシミュレータの知識は必要ありません。

インポート方法：
```python
from artefacts_toolkit.rerun import recorder, reader, video
```

> [!NOTE]
> * これらのヘルパーは、リリースごとに変わる `rerun-sdk` の部分（記録シンク、データフレーム/クエリ API）をラップしているため、ヘルパーに対して書かれたテストコードは Rerun をアップグレードしても動作し続けます。
> * データ列は `"/entity:Archetype:field"` という形式で名付けられます。例：`"/base:Transform3D:translation"` や `"/base/yaw:Scalars:scalars"`。記録内の列を一覧するには [`reader.get_columns`](#readerget_columns) を使用してください。
> * すべての `reader` および `video` 関数は、`.rrd` ファイルへのパス、または [`reader.load`](#readerload) が返すオブジェクトのどちらも受け付けます。

## 関数
* [`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)

## 関数リファレンス

### `recorder.get_output_dir`

記録やその他のテスト出力を書き込むべきディレクトリを返します（存在しなければ作成します）。

```python
recorder.get_output_dir(
    directory=None
)
```

#### パラメータ

| **パラメータ** | **型**              | **説明**                                             | **デフォルト** |
| ------------- | ------------------- | ---------------------------------------------------- | ------------- |
| `directory`   | `str` または `Path` | デフォルトの代わりに使用するディレクトリを明示的に指定 | `None`        |

#### 戻り値

`Path`: 出力ディレクトリの絶対パス。

> [!NOTE]
> * `directory` が指定されていない場合、環境変数 `ARTEFACTS_SCENARIO_UPLOAD_DIR` が使用されます。これは `artefacts run` によって設定され、その中のすべてのファイルはシナリオ終了時に Artefacts ダッシュボードにアップロードされます。
> * どちらも利用できない場合（例えばローカルで `pytest` を実行する場合）、`./test_outputs` が使用されます。

#### 例

シミュレータ側とテスト側で同じディレクトリが解決されるため、記録の場所について両者が一致します：

```python
from artefacts_toolkit.rerun import recorder

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

---

### `recorder.start_recording`

`directory/filename` に保存される Rerun の記録を開始し、記録ストリームとファイルパスを返します。

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

#### パラメータ

| **パラメータ**    | **型**              | **説明**                                                                                  | **デフォルト**     |
| ---------------- | ------------------- | ----------------------------------------------------------------------------------------- | ----------------- |
| `application_id` | `str`               | Rerun ビューアに表示されるアプリケーション名                                               | 必須              |
| `filename`       | `str` または `Path` | `.rrd` ファイルの名前。絶対パスも指定可能で、その場合 `directory` は無視されます            | `"recording.rrd"` |
| `directory`      | `str` または `Path` | 保存先ディレクトリ。[`get_output_dir`](#recorderget_output_dir) と同様に解決されます       | `None`            |
| `handle_sigterm` | `bool`              | プロセスが SIGTERM を受け取ったときに記録を閉じ、クリーンに終了する                          | `True`            |

#### 戻り値

`tuple`: 以下を含むタプルを返します：

* `recording` (`rerun.RecordingStream`): 開始された記録。グローバルな記録としても設定されるため、通常の `rr.log(...)` や `rr.set_time(...)` の呼び出しはこの記録に送られます。
* `path` (`Path`): 書き込まれる `.rrd` ファイルへのパス。

> [!NOTE]
> * 記録を開始する前に、同名の古いファイルは削除されます。
> * テストハーネスは通常、`Popen.terminate()`（SIGTERM）でシミュレータを停止します。`handle_sigterm` が有効な場合、まず記録が閉じられるため、動画エンコーダの終了などの遅いクリーンアップ中にハーネスがプロセスを強制終了してもファイルは完全な状態になります。
> * 一部のネイティブライブラリは初期化時にシグナルハンドラをリセットします（limxsdk の `Robot.init` など）。それらの後に `start_recording` を呼び出してください。
> * `artefacts run` で `--no-upload` フラグを使用しない限り、記録は自動的に Artefacts ダッシュボードにアップロードされます。

#### 例

以下の例では、シミュレータはベースの姿勢とジョイスティックのコマンドを `sim_time` タイムライン上に記録し、ミッション終了後にテストハーネスによって停止されます。`rr.set_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")

    # robot.init() の後に呼び出す：SDK がシグナルハンドラをリセットするため
    recorder.start_recording("tron2_loop", "recording-loop.rrd")

    RerunSimulator(...).run()
```

---

### `reader.load`

記録を一度メモリに読み込み、ファイルを再読み込みすることなく複数のアサーションで共有できるようにします。

```python
reader.load(
    rrd
)
```

#### パラメータ

| **パラメータ** | **型**              | **説明**               | **デフォルト** |
| ------------- | ------------------- | ---------------------- | ------------- |
| `rrd`         | `str` または `Path` | `.rrd` ファイルへのパス | 必須          |

#### 戻り値

`ChunkStore`: 読み込まれた記録。他のすべての `reader` および `video` 関数は、パスの代わりにこれを受け付けます。

> [!NOTE]
> * 古いバージョンの Rerun で書かれたファイルや、完了前に強制終了されたプロセスが書いたファイルも読み込めます。

#### 例

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

記録に記録されたすべてのエンティティパスを返します。

```python
reader.get_entity_paths(
    rrd
)
```

#### パラメータ

| **パラメータ** | **型**                            | **説明**                               | **デフォルト** |
| ------------- | --------------------------------- | -------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録   | 必須          |

#### 戻り値

`list[str]`: エンティティパス。例：`["/base", "/base/yaw", "/cmd/fwd"]`。

#### 例

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

---

### `reader.get_timelines`

記録内のタイムライン名を返します。

```python
reader.get_timelines(
    rrd
)
```

#### パラメータ

| **パラメータ** | **型**                            | **説明**                               | **デフォルト** |
| ------------- | --------------------------------- | -------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録   | 必須          |

#### 戻り値

`list[str]`: タイムライン名。例：`["log_time", "sim_time"]`。

> [!NOTE]
> * `log_time` は常に存在します（Rerun が自動的に追加します）。`rr.set_time` で自分で設定したタイムラインが、他のヘルパーがデフォルトで使用するものです。

---

### `reader.get_columns`

記録のデータ列名を返します。オプションで 1 つのエンティティの列のみに絞り込めます。

```python
reader.get_columns(
    rrd,
    entity=None
)
```

#### パラメータ

| **パラメータ** | **型**                            | **説明**                               | **デフォルト** |
| ------------- | --------------------------------- | -------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録   | 必須          |
| `entity`      | `str`                             | このエンティティの列のみを一覧する       | `None`        |

#### 戻り値

`list[str]`: `"/entity:Archetype:field"` 形式の列名。

#### 例

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

---

### `reader.get_column`

1 つの列の時刻と値を、タイムラインで並べ替えて返します。

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

#### パラメータ

| **パラメータ** | **型**                            | **説明**                                                                                                   | **デフォルト** |
| ------------- | --------------------------------- | ---------------------------------------------------------------------------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録                                                                       | 必須          |
| `column`      | `str`                             | 列名。例：`"/base/yaw:Scalars:scalars"`                                                                    | 必須          |
| `timeline`    | `str`                             | 並べ替えに使うタイムライン。デフォルトは記録固有のタイムライン（`log_time` 以外の最初のもの）、なければ `log_time` | `None`        |

#### 戻り値

`tuple`: 以下を含むタプルを返します：

* `times` (`numpy.ndarray`): duration タイムラインでは秒（float）、timestamp タイムラインでは `datetime64`、sequence タイムラインでは整数。
* `values` (`numpy.ndarray`): スカラーでは形状 `(n,)`、3 次元ベクトルでは `(n, 3)`。行ごとにインスタンス数が異なる場合（点群など）は、行ごとの配列を格納したオブジェクト配列。

> [!NOTE]
> * 列が存在しない場合は `KeyError` を送出します。

#### 例

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

列に最後に記録された値、または静的な列の値を取得します。

```python
reader.get_final_message(
    rrd,
    column
)
```

#### パラメータ

| **パラメータ** | **型**                            | **説明**                                      | **デフォルト** |
| ------------- | --------------------------------- | --------------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録          | 必須          |
| `column`      | `str`                             | 列名。例：`"/base:Transform3D:translation"`   | 必須          |

#### 戻り値

`Any`: アンラップされた値。`Scalars` では float、平行移動では長さ 3 の配列、複数点の `Points3D` では `(n, 3)` の配列。

#### 例

ウェイポイントは `Points3D` として一度だけ（静的に）記録され、テストはそれらが描く軌道からロボットがどれだけ逸れたかを測定します：

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

エンティティに対して行われたログ呼び出しの回数を返します。

```python
reader.get_message_count(
    rrd,
    entity
)
```

#### パラメータ

| **パラメータ** | **型**                            | **説明**                               | **デフォルト** |
| ------------- | --------------------------------- | -------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録   | 必須          |
| `entity`      | `str`                             | エンティティパス。例：`"/cam"`          | 必須          |

#### 戻り値

`int`: エンティティに記録された（静的でない）行数。存在しない場合は `0`。

#### 例

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

---

### `reader.to_dataframe`

記録を pandas の DataFrame として返します。選択したタイムライン上の時刻ごとに 1 行です。

異なるエンティティは通常、異なるレートで記録されます。姿勢はシミュレーションステップごと、ジョイスティックのコマンドは変化したときだけ、といった具合です。`step` を指定すると、選択したすべてのエンティティが 1 つの均一な時間グリッドにリサンプリングされ、`fill_latest_at` により欠損はそれ以前の最新の値で埋められます。そのため疎なコマンドトピックが密な姿勢ストリームと行ごとに揃い、両者を直接比較できます。

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

#### パラメータ

| **パラメータ**    | **型**                            | **説明**                                                                                                                   | **デフォルト** |
| ---------------- | --------------------------------- | -------------------------------------------------------------------------------------------------------------------------- | ------------- |
| `rrd`            | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録                                                                                       | 必須          |
| `contents`       | `str` または `list[str]`          | 含めるエンティティ：1 つのエンティティなら `"/base"`、サブツリーなら `"/base/**"`、またはそれらのリスト。`None` の場合はすべてのエンティティ | `None` |
| `index`          | `str`                             | 行インデックスとして使うタイムライン。デフォルトは [`get_column`](#readerget_column) と同様                                     | `None`        |
| `step`           | `float`                           | 最初から最後の時刻まで、この秒数間隔の均一なグリッドにリサンプリングする                                                        | `None`        |
| `fill_latest_at` | `bool`                            | 各列の欠損を、それ以前の最新の値で埋める                                                                                      | `True`        |

#### 戻り値

`pandas.DataFrame`: タイムライン列が最初に来て（duration タイムラインでは秒の float）、その後に `"/entity:Archetype:field"` ごとの列が続きます。静的データはすべての行に結合されます。`Scalars` 列は通常の `float64`（何も記録されていない箇所は `NaN`）、ベクトルは配列です。

> [!NOTE]
> * `step` を指定しない場合、選択したエンティティのいずれかが記録された各時刻に 1 行となります。
> * ベクトルのセルは numpy 配列です。`np.vstack(df[column])` で平行移動の列を `(n, 3)` の配列に変換できます。

#### 例

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

記録内のカメラエンティティから MP4 動画を作成します。

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

#### パラメータ

| **パラメータ** | **型**                            | **説明**                                                                            | **デフォルト** |
| ------------- | --------------------------------- | ----------------------------------------------------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録                                                | 必須          |
| `entity`      | `str`                             | カメラのエンティティパス。例：`"/cam"`                                                | 必須          |
| `output_path` | `str` または `Path`               | 動画の保存先パス（.mp4）                                                             | 必須          |
| `timeline`    | `str`                             | フレームの並べ替えに使うタイムライン。デフォルトは [`get_column`](#readerget_column) と同様 | `None`        |
| `frame_rate`  | `int`                             | `rr.Image` フレームをエンコードする際のフレームレート                                   | `20`          |

#### 戻り値

`Path`: 保存された動画のパス。

> [!NOTE]
> * `rr.Image` フレームとして記録されたエンティティ（`frame_rate` で H.264 エンコード）と、H.264 サンプルを持つ `rr.VideoStream` エンティティ（記録のタイムスタンプを保ったまま再多重化）の両方に対応しています。
> * `output_path` に `.mp4` 拡張子がない場合は自動的に追加されます。
> * [`recorder.get_output_dir`](#recorderget_output_dir) が返すディレクトリに保存することで、動画は自動的に Artefacts ダッシュボードにアップロードされます。

#### 例

```python
from artefacts_toolkit.rerun import recorder, video

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

---

### `video.extract_camera_image`

カメラエンティティに最後に記録された画像を PNG として保存します。

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

#### パラメータ

| **パラメータ** | **型**                            | **説明**                               | **デフォルト** |
| ------------- | --------------------------------- | -------------------------------------- | ------------- |
| `rrd`         | `str`、`Path` または `ChunkStore` | 記録へのパス、または読み込み済みの記録   | 必須          |
| `entity`      | `str`                             | カメラのエンティティパス。例：`"/cam"`   | 必須          |
| `output_dir`  | `str` または `Path`               | 抽出した画像の保存先ディレクトリ         | `"output"`    |

#### 戻り値

`Path`: 保存された画像のパス。`output_dir/<entity>.last.png` で、`/` は `_` に置き換えられます（例：`output/_cam.last.png`）。

> [!NOTE]
> * 8 ビットの L、RGB、RGBA、BGR、BGRA ピクセルを持つ `rr.Image` エンティティのみに対応しています。

#### 例

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