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

from artefacts_toolkit.rerun import recorder, reader, video

recorder.get_output_dir(
    directory=None
)

from artefacts_toolkit.rerun import recorder

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

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

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(
    rrd
)

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(
    rrd
)

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

reader.get_timelines(
    rrd
)

reader.get_columns(
    rrd,
    entity=None
)

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

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

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(
    rrd,
    column
)

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(
    rrd,
    entity
)

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

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

@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(
    rrd,
    entity,
    output_path,
    timeline=None,
    frame_rate=20
)

from artefacts_toolkit.rerun import recorder, video

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

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

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