Recorder#

The Recorder class is the base class for all recording functionality in Genesis. It provides the interface for capturing and processing simulation data.

Overview#

Recorders:

  • Capture data at specified frequencies

  • Process data synchronously or asynchronously

  • Support building/cleanup lifecycle

  • Can be reset between episodes

Creating Custom 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 = []

Lifecycle#

  1. __init__: Configure recorder options

  2. build(): Initialize resources (called when scene builds)

  3. process(data, time): Handle each data sample (called during recording)

  4. cleanup(): Finalize and release resources (called when recording stops)

  5. reset(): Reset state for new episode

API Reference#

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

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()[source]#

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

process(data, cur_time)[source]#

Process each incoming data sample.

Parameters:

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

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

cleanup()[source]#

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)[source]#

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()[source]#

Start the recording thread if run_in_thread is True.

stop()[source]#

Stop the recording thread and cleanup resources.

join_thread()[source]#

Wait for the processor thread to finish.

start_thread()[source]#

Wait for the processor thread to finish.

sync(timeout: float | None = None)[source]#

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)[source]#

Process a simulation step, potentially recording data.

property is_built: bool#

See Also#