LogRerun

class LogRerun(env, logged_envs=0, recording_stream=None, save_path=None, episode_trigger=None, step_trigger=None, recording_length=0)[source]

Bases: Wrapper[ObsType, ActType, ObsType, ActType], Generic[ObsType, ActType, RenderFrame], RecordConstructorArgs

Logs Isaac Lab based environment episodes with Rerun.io.

The API follows Gymnasium’s gymnasium.wrappers.RecordVideo wrapper, but instead of writing video, it logs Isaac Lab scenes to a Rerun stream.

Typically you only want to log intermittently (e.g. every 100th episode or every N environment steps). Use either episode_trigger or step_trigger to decide when a new recording should start.

The recording stream can either be provided directly via recording_stream, or a new recording stream can be created that saves to save_path. If neither is provided, it will try to find it using rr.get_data_recording(). save_path takes precedence over recording_stream.

Parameters:

  • env (gym.Env[ObsType, ActType]) – The environment to wrap. Must be Isaac Lab based and expose env.unwrapped.scene.

  • logged_envs (int | list[int]) – Indices of environments to log (for multi-env Isaac Lab scenes). If an int, it is treated as a single environment index. If a list, those indices are logged. Defaults to 0 (the first environment).

  • recording_stream (rr.RecordingStream | None) – The Rerun recording stream to use. Ignored if save_path is provided.

  • save_path (Path | str | None) – Path where the Rerun recording will be saved. If provided, a new recording stream is created that saves to this path.

  • episode_trigger (Callable[[int], bool] | None) – Callable episode_trigger(episode_id) -> bool returning True iff a recording should start on the given episode. If both episode_trigger and step_trigger are None, defaults to gymnasium.utils.save_video.capped_cubic_video_schedule(). (Same signature as in gymnasium.wrappers.RecordVideo.)

  • step_trigger (Callable[[int], bool] | None) – Callable step_trigger(step_id) -> bool returning True iff a recording should start on the current global environment step (summed over episodes). (Same signature as in gymnasium.wrappers.RecordVideo.)

  • recording_length (int) – Number of frames to record per snippet. If 0, record full episodes (until reset). If strictly positive, record fixed-length snippets. (Same as video_length in gymnasium.wrappers.RecordVideo.)

Raises:

ValueError – If the environment does not expose a scene attribute on env.unwrapped, i.e. it does not appear to be an Isaac Lab based environment.

Notes

  • No vectorized wrapper variant is provided.

  • When a recording is active, each environment step logs the Isaac Lab scene at a timestamp derived from the scene physics_dt.

Examples

Record every 10th episode and write .rrd files to disk:

trigger = lambda ep: ep % 10 == 0

rr.init("gymnasium_example", spawn=True)
env = LogRerun(env, save_path="./save_rerun1.rrd", episode_trigger=trigger)

Start a recording every 200th step and record 100 frames each time:

trigger = lambda t: t % 200 == 0

# LogRerun will find the global Rerun recording with rr.get_data_recording()
rr.init("gymnasium_example", spawn=True)
env = LogRerun(env, step_trigger=trigger, recording_length=100)

Record everything, but split into chunks of 1000 frames:

# Set the recording stream directly
recording = rr.RecordingStream("gymnasium_example")
env = LogRerun(env, recording_length=1000, recording_stream=recording)
__init__(env, logged_envs=0, recording_stream=None, save_path=None, episode_trigger=None, step_trigger=None, recording_length=0)[source]

Create the wrapper.

Parameters:

  • env (Env[ObsType, ActType])

  • logged_envs (int | list[int])

  • recording_stream (RecordingStream | None)

  • save_path (Path | str | None)

  • episode_trigger (Callable[[int], bool] | None)

  • step_trigger (Callable[[int], bool] | None)

  • recording_length (int)

property action_space: Space[ActType] | Space[WrapperActType]

Return the Env action_space unless overwritten then the wrapper action_space is used.

classmethod class_name()

Returns the class name of the wrapper.

Return type:

str

close()[source]

Closes the wrapper and flushes the recording stream.

get_wrapper_attr(name)

Gets an attribute from the wrapper and lower environments if name doesn’t exist in this object.

Parameters:

name (str) – The variable name to get

Returns:

The variable with name in wrapper or lower environments

Return type:

Any

has_wrapper_attr(name)

Checks if the given attribute is within the wrapper or its environment.

Parameters:

name (str)

Return type:

bool

property metadata: dict[str, Any]

Returns the Env metadata.

property np_random: Generator

Returns the Env np_random attribute.

property np_random_seed: int | None

Returns the base environment’s np_random_seed.

property observation_space: Space[ObsType] | Space[WrapperObsType]

Return the Env observation_space unless overwritten then the wrapper observation_space is used.

render()

Uses the render() of the env that can be overwritten to change the returned data.

Return type:

RenderFrame | list[RenderFrame] | None

property render_mode: str | None

Returns the Env render_mode.

reset(*, seed=None, options=None)[source]

Reset the environment and eventually starts a new recording.

Parameters:

  • seed (int | None)

  • options (dict[str, Any] | None)

Return type:

tuple[ObsType, dict[str, Any]]

property sequence_timeline_name: str | None

Get the name of the step based timeline, if any.

set_wrapper_attr(name, value, *, force=True)

Sets an attribute on this wrapper or lower environment if name is already defined.

Parameters:

  • name (str) – The variable name

  • value (Any) – The new variable value

  • force (bool) – Whether to create the attribute on this wrapper if it does not exists on the lower environment instead of raising an exception

Returns:

If the variable has been set in this or a lower wrapper.

Return type:

bool

property spec: EnvSpec | None

Returns the Env spec attribute with the WrapperSpec if the wrapper inherits from EzPickle.

start_recording(timeline_name)[source]

Start a new recording. If it is already recording, stops the current recording before starting the new one.

Parameters:

timeline_name (str)

step(action)[source]

Steps through the environment using action and logs the environment if recording is active.

Parameters:

action (ActType)

Return type:

tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]

stop_recording()[source]

Stop current recording and flush the recording stream.

property timestamp_timeline_name: str | None

Get the name of the timestamp based timeline, if any.

property unwrapped: Env[ObsType, ActType]

Returns the base environment of the wrapper.

This will be the bare gymnasium.Env environment, underneath all layers of wrappers.

classmethod wrapper_spec(**kwargs)

Generates a WrapperSpec for the wrappers.

Parameters:

kwargs (Any)

Return type:

WrapperSpec