Recorder

Recorder 类是 Genesis 中所有录制功能的基类。它提供了捕获和处理仿真数据的接口。

概述

Recorders:

• 以指定频率捕获数据

• 支持同步或异步处理数据

• 支持构建/清理生命周期

• 可在 episode 之间重置

创建自定义 Recorders

import genesis as gs
from genesis.recorders import Recorder

class MyRecorder(Recorder):
    def __init__(self, manager, options, data_func):
        super().__init__(manager, options, data_func)
        self.data_buffer = []

    def build(self):
        super().build()
        self.data_buffer = []

    def process(self, data, cur_time):
        self.data_buffer.append({
            "time": cur_time,
            "data": data,
        })

    def cleanup(self):
        # Save or finalize data
        print(f"Recorded {len(self.data_buffer)} samples")
        self.data_buffer = []

    def reset(self, envs_idx=None):
        self.data_buffer = []

生命周期

1. __init__: 配置 recorder 选项

2. build(): 初始化资源（在 scene 构建时调用）

3. process(data, time): 处理每个数据样本（在录制期间调用）

4. cleanup(): 完成并释放资源（在录制停止时调用）

5. reset(): 为新 episode 重置状态

API 参考

class genesis.recorders.base_recorder.Recorder(manager: RecorderManager, options: RecorderOptions, data_func: Callable[[], T])

Bases: Generic[T]

Base class for all recorders.

Note that modifying the signature of this class in recorder implementations should be avoided since instantiation is done through the RecorderManager.

build()

Build the recorder, e.g. by initializing variables and creating widgets or file handles.

process(data, cur_time)

Process each incoming data sample.

Parameters:

• data (Any) – The data to be processed.

• cur_time (float) – The current time of the simulation.

cleanup()

Cleanup all resources, e.g. by closing widgets or files.

This method is called when recording is stopped by scene.stop_recording().

reset(envs_idx=None)

Reset the recorder, e.g. by flushing stored data.

This method is called when the scene is reset by scene.reset().

Parameters:

envs_idx (array_like, optional) – The indices of the environments to reset. If None, all environments are reset.

property run_in_thread: bool

Whether to run the recorder in a background thread.

Running in a background thread allows for processing data without blocking the main thread, so this is encouraged for most recorders (simply return True), but implementers should check that the recorder is thread-safe on all devices (threading on macOS tends to be less supported).

start()

Start the recording thread if run_in_thread is True.

stop()

Stop the recording thread and cleanup resources.

join_thread()

Wait for the processor thread to finish.

start_thread()

Wait for the processor thread to finish.

sync(timeout: float | None = None)

Wait until the data queue is empty.

Parameters:

timeout (float | None) – The maximum time to wait for the data queue to be empty. If None, wait indefinitely. If the timeout is reached, an exception is raised.

step(global_step: int)

Process a simulation step, potentially recording data.

property is_built: bool

另请参阅

• RecorderManager - 管理多个 recorders

• File Writers - 内置文件写入器

• Plotters - 内置绘图器