Skip to content

rendering

FrameClock

Decides when the next video frame is due, given the simulation time.

Shared by renderers and trajectory recorders so that both sample the simulation at exactly the same timepoints.

Parameters:

Name Type Description Default
playback_speed float

Video playback speed relative to real time.

required
output_fps int

Output video frame rate.

required
timestep float

Simulation timestep in seconds. Half of it is used as a tolerance when comparing times, to absorb floating-point drift.

required

Attributes:

Name Type Description
interval

Simulation time between consecutive frames, in seconds.

Source code in src/flygym/rendering.py
class FrameClock:
    """Decides when the next video frame is due, given the simulation time.

    Shared by renderers and trajectory recorders so that both sample the simulation at
    exactly the same timepoints.

    Args:
        playback_speed: Video playback speed relative to real time.
        output_fps: Output video frame rate.
        timestep: Simulation timestep in seconds. Half of it is used as a tolerance
            when comparing times, to absorb floating-point drift.

    Attributes:
        interval: Simulation time between consecutive frames, in seconds.
    """

    def __init__(self, playback_speed: float, output_fps: int, timestep: float):
        self.playback_speed = playback_speed
        self.output_fps = output_fps
        self.interval = playback_speed / output_fps
        self._tolerance = timestep * 0.5
        self.reset()

    def reset(self) -> None:
        """Make the next call to `tick` fire regardless of the time."""
        self._last_frame_time = -np.inf

    def tick(self, time: float) -> bool:
        """Return True (and register a frame at `time`) if a frame is due."""
        if time < self._last_frame_time + self.interval - self._tolerance:
            return False
        self._last_frame_time = float(time)
        return True

reset()

Make the next call to tick fire regardless of the time.

Source code in src/flygym/rendering.py
def reset(self) -> None:
    """Make the next call to `tick` fire regardless of the time."""
    self._last_frame_time = -np.inf

tick(time)

Return True (and register a frame at time) if a frame is due.

Source code in src/flygym/rendering.py
def tick(self, time: float) -> bool:
    """Return True (and register a frame at `time`) if a frame is due."""
    if time < self._last_frame_time + self.interval - self._tolerance:
        return False
    self._last_frame_time = float(time)
    return True

Renderer

Renders MuJoCo scenes to video frames.

Frames are rendered from a private copy of the simulation state, in which the geometry is recomputed from the current qpos and mocap poses (see update_render_geometry). Rendered frames are therefore consistent with the current qpos, rather than with the geometry computed at the start of the last physics step. The simulation data is not modified. Quantities that do not depend on positions alone (e.g., contact forces, if enabled in scene_option) are not rendered correctly.

Parameters:

Name Type Description Default
mj_model MjModel

Compiled MuJoCo model.

required
cameras str | MjsCamera | list[str | MjsCamera]

Camera(s) to render. Can be a camera name, MJCF camera element, or a sequence of either.

required
camera_res tuple[int, int]

(height, width) in pixels.

(240, 320)
playback_speed float

Video playback speed relative to real time.

0.2
output_fps int

Output video frame rate.

25
buffer_frames bool

If True, store rendered frames on the renderer.

True
scene_option MjvOption | None

MuJoCo scene options. Uses defaults if None.

None
render_rgb bool

If True, render RGB frames (stored in self.frames).

True
render_depth bool

If True, render depth maps (raw float arrays in self.depth_frames). Not saved/shown as video.

False
render_segmentation bool

If True, render segmentation masks (raw int arrays in self.segmentation_frames). Not saved/shown as video.

False
**kwargs Any

Passed to mujoco.Renderer.

{}

render_rgb, render_depth, and render_segmentation are independent; enable any combination. At least one must be True.

Attributes:

Name Type Description
frames

Dict mapping camera name to list of RGB frames (uint8), or None if render_rgb is False. Only populated when buffer_frames=True.

depth_frames

Like frames but float depth maps; None unless render_depth.

segmentation_frames

Like frames but int32 segmentation object ids (-1 = background); None unless render_segmentation.

Source code in src/flygym/rendering.py
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
class Renderer:
    """Renders MuJoCo scenes to video frames.

    Frames are rendered from a private copy of the simulation state, in which the
    geometry is recomputed from the current `qpos` and mocap poses (see
    `update_render_geometry`). Rendered frames are therefore consistent with the
    current `qpos`, rather than with the geometry computed at the start of the last
    physics step. The simulation data is not modified. Quantities that do not depend
    on positions alone (e.g., contact forces, if enabled in `scene_option`) are not
    rendered correctly.

    Args:
        mj_model: Compiled MuJoCo model.
        cameras: Camera(s) to render. Can be a camera name, MJCF camera element,
            or a sequence of either.
        camera_res: ``(height, width)`` in pixels.
        playback_speed: Video playback speed relative to real time.
        output_fps: Output video frame rate.
        buffer_frames: If True, store rendered frames on the renderer.
        scene_option: MuJoCo scene options. Uses defaults if None.
        render_rgb: If True, render RGB frames (stored in ``self.frames``).
        render_depth: If True, render depth maps (raw float arrays in
            ``self.depth_frames``). Not saved/shown as video.
        render_segmentation: If True, render segmentation masks (raw int arrays in
            ``self.segmentation_frames``). Not saved/shown as video.
        **kwargs: Passed to ``mujoco.Renderer``.

    ``render_rgb``, ``render_depth``, and ``render_segmentation`` are independent;
    enable any combination. At least one must be True.

    Attributes:
        frames: Dict mapping camera name to list of RGB frames (uint8), or None if
            ``render_rgb`` is False. Only populated when ``buffer_frames=True``.
        depth_frames: Like ``frames`` but float depth maps; None unless
            ``render_depth``.
        segmentation_frames: Like ``frames`` but int32 segmentation object ids
            (-1 = background); None unless ``render_segmentation``.
    """

    def __init__(
        self,
        mj_model: mj.MjModel,
        cameras: str | mj.MjsCamera | list[str | mj.MjsCamera],
        *,
        camera_res: tuple[int, int] = (240, 320),
        playback_speed: float = 0.2,
        output_fps: int = 25,
        buffer_frames: bool = True,
        scene_option: mj.MjvOption | None = None,
        render_rgb: bool = True,
        render_depth: bool = False,
        render_segmentation: bool = False,
        **kwargs: Any,
    ):
        self.mj_model = mj_model
        self.camera_res = camera_res
        nrows, ncols = camera_res
        self.buffer_frames = buffer_frames

        self.mj_renderer = mj.Renderer(mj_model, nrows, ncols, **kwargs)
        self._render_data = mj.MjData(mj_model)
        # RGB / depth / segmentation are independent and may be enabled in any
        # combination. Each is produced by its own render() pass (mujoco renders
        # one output type per call), so we don't enable a mode here -- the mode is
        # toggled per pass in render_as_needed and left in the default RGB state.
        self.render_rgb = render_rgb
        self.render_depth = render_depth
        self.render_segmentation = render_segmentation
        if not (render_rgb or render_depth or render_segmentation):
            raise ValueError(
                "At least one of render_rgb, render_depth, or "
                "render_segmentation must be True."
            )

        if scene_option is None:
            self.scene_option = mj.MjvOption()
        else:
            self.scene_option = scene_option
        mj.mjv_defaultOption(self.scene_option)  # this sets default scene options

        self._cameras_names2id = {}
        for spec in cameras if isinstance(cameras, list) else [cameras]:
            cam_id, cam_name = self._resolve_camera_id_and_name(spec)
            if cam_id == -1:
                raise ValueError(f"Camera {spec} not found in the model.")
            if cam_name in self._cameras_names2id:
                raise ValueError(f"Duplicate camera name detected: {cam_name}.")
            self._cameras_names2id[cam_name] = cam_id
        if len(self._cameras_names2id) == 0:
            raise ValueError("At least one valid camera must be specified.")
        self._cameras_id2name = {v: k for k, v in self._cameras_names2id.items()}

        self.playback_speed = playback_speed
        self.output_fps = output_fps
        self._clock = FrameClock(playback_speed, output_fps, mj_model.opt.timestep)

        buffer = self.buffer_frames
        self.frames = self._new_frame_buffer() if buffer and render_rgb else None
        self.depth_frames = (
            self._new_frame_buffer() if buffer and render_depth else None
        )
        self.segmentation_frames = (
            self._new_frame_buffer() if buffer and render_segmentation else None
        )

    def _new_frame_buffer(self) -> dict[str, list]:
        return {cam_name: [] for cam_name in self._cameras_names2id}

    def get_camera_matrix(
        self, camera: str | mj.MjsCamera, mj_data: mj.MjData, mj_model: mj.MjModel
    ) -> np.ndarray:
        """Get the 3x4 camera projection matrix from the current MjData.

        The returned matrix maps homogeneous world coordinates to homogeneous
        image (pixel) coordinates, following the standard MuJoCo
        ``image @ focal @ rotation @ translation`` composition.
        """
        internal_cam_id, _ = self._resolve_camera_id_and_name(camera)
        # update the scene to get the latest camera position and orientation
        self.mj_renderer.update_scene(mj_data, internal_cam_id, self.scene_option)
        pos = mj_data.cam_xpos[internal_cam_id]
        rot = mj_data.cam_xmat[internal_cam_id].reshape(3, 3)
        fov = mj_model.cam_fovy[internal_cam_id]
        height, width = self.camera_res

        # Translation matrix (4x4).
        translation = np.eye(4)
        translation[0:3, 3] = -pos
        # Rotation matrix (4x4).
        rotation = np.eye(4)
        rotation[0:3, 0:3] = rot
        # Focal transformation matrix (3x4).
        focal_scaling = (1.0 / np.tan(np.deg2rad(fov) / 2)) * height / 2.0
        focal = np.diag([-focal_scaling, focal_scaling, 1.0, 0])[0:3, :]
        # Image matrix (3x3).
        image = np.eye(3)
        image[0, 2] = (width - 1) / 2.0
        image[1, 2] = (height - 1) / 2.0

        return image @ focal @ rotation @ translation

    def render_as_needed(self, mj_data: mj.MjData) -> bool:
        """Render frames for all cameras if enough time has elapsed.

        Args:
            mj_data: Current MuJoCo data.

        Returns:
            True if frames were rendered, False otherwise.
        """
        if not self._clock.tick(mj_data.time):
            return False

        self._render_data.qpos[:] = mj_data.qpos
        self._render_data.mocap_pos[:] = mj_data.mocap_pos
        self._render_data.mocap_quat[:] = mj_data.mocap_quat
        update_render_geometry(self.mj_model, self._render_data)
        for cam_name, internal_cam_id in self._cameras_names2id.items():
            self.mj_renderer.update_scene(
                self._render_data, internal_cam_id, self.scene_option
            )
            # One render() pass per enabled output. mujoco renders a single output
            # type per call, so toggle the mode for each and always restore the
            # default (RGB) state afterwards.
            if self.render_rgb:
                rgb = self.mj_renderer.render()
                if self.buffer_frames:
                    self.frames[cam_name].append(rgb)
            if self.render_depth:
                self.mj_renderer.enable_depth_rendering()
                depth = self.mj_renderer.render()
                self.mj_renderer.disable_depth_rendering()
                if self.buffer_frames:
                    self.depth_frames[cam_name].append(depth)
            if self.render_segmentation:
                self.mj_renderer.enable_segmentation_rendering()
                # MuJoCo segmentation returns int32 (H, W, 2): channel 0 is the
                # object id, channel 1 the object type, with -1 marking background.
                # Keep channel 0 only (assumes objects are geoms, so the id is
                # unambiguous), copied to a standalone int32 array so the unused
                # second channel isn't retained.
                seg = self.mj_renderer.render()[:, :, 0].copy()
                self.mj_renderer.disable_segmentation_rendering()
                if self.buffer_frames:
                    self.segmentation_frames[cam_name].append(seg)
        return True

    def reset(self) -> None:
        """Clear buffered frames and reset the render timer."""
        self._clock.reset()
        if not self.buffer_frames:
            return
        if self.render_rgb:
            self.frames = self._new_frame_buffer()
        if self.render_depth:
            self.depth_frames = self._new_frame_buffer()
        if self.render_segmentation:
            self.segmentation_frames = self._new_frame_buffer()

    def close(self) -> None:
        """Release the underlying MuJoCo renderer resources."""
        self.mj_renderer.close()

    def __enter__(self) -> "Renderer":
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: object,
    ) -> bool:
        self.close()
        return False  # don't suppress exceptions

    def show_in_notebook(
        self,
        camera: str | mj.MjsCamera | list[str | mj.MjsCamera] | None = None,
        **kwargs: Any,
    ) -> None:
        """Display recorded frames in a Jupyter notebook.

        Only RGB frames are displayed. Depth/segmentation renders are kept as raw
        arrays in ``self.depth_frames`` / ``self.segmentation_frames``.

        Args:
            camera: Camera(s) to display. If None, displays all cameras.
            **kwargs: Additional arguments passed to mediapy.show_video
        """
        if not self._warn_unless_rgb("displayed"):
            return
        camera_names = self._normalize_camera_spec(camera)

        for cam_name in camera_names:
            frames = self.frames[cam_name]
            if len(frames) == 0:
                raise RuntimeError(f"No frames recorded yet for camera '{cam_name}'.")
            mediapy.show_video(frames, fps=self.output_fps, title=cam_name, **kwargs)

    def save_video(
        self,
        output_path: dict[str | mj.MjsCamera, PathLike] | PathLike,
        **kwargs: Any,
    ) -> None:
        """Save recorded frames as video files.

        Only RGB frames are saved. Depth/segmentation renders are kept as raw
        arrays in ``self.depth_frames`` / ``self.segmentation_frames``.

        Args:
            output_path: Either a dict mapping camera specs to file paths, or:
                - If single camera: a file path to save to
                - If multiple cameras: a directory path to save all videos to
            **kwargs: Additional arguments passed to imageio.imwrite
        """
        if not self._warn_unless_rgb("saved"):
            return
        path_by_camera = self._resolve_output_paths(output_path)

        for cam_name, path in path_by_camera.items():
            frames = self.frames[cam_name]
            if len(frames) == 0:
                raise RuntimeError(f"No frames recorded yet for camera '{cam_name}'.")

            path.parent.mkdir(parents=True, exist_ok=True)

            iio.imwrite(
                path, frames, fps=self.output_fps, codec="libx264", quality=8, **kwargs
            )

    def _warn_unless_rgb(self, verb: str) -> bool:
        """Return True if RGB frames exist to save/show; warn and return False if not.

        Only RGB frames can be encoded as video; depth (float) and segmentation
        (int) are kept as raw arrays rather than lossily cast to uint8. Warn when
        such buffers exist so they are not silently expected in the video.
        """
        if not self.render_rgb:
            warnings.warn(
                f"No RGB frames to be {verb} as video (render_rgb=False). Depth/"
                "segmentation frames are raw arrays in `Renderer.depth_frames` / "
                "`Renderer.segmentation_frames`."
            )
            return False
        if self.render_depth or self.render_segmentation:
            warnings.warn(
                f"Only RGB frames are {verb} as video; depth/segmentation frames "
                "are kept as raw arrays in `Renderer.depth_frames` / "
                "`Renderer.segmentation_frames`."
            )
        return True

    def _normalize_camera_spec(
        self,
        camera: str | mj.MjsCamera | list[str | mj.MjsCamera] | None,
    ) -> list[str]:
        """Convert various camera specifications to a list of camera names.

        Args:
            camera: Camera specification (single, sequence, or None for all)

        Returns:
            List of camera names

        Raises:
            ValueError: If camera spec is invalid or refers to unavailable cameras
        """
        if camera is None:
            return list(self._cameras_names2id.keys())
        elif isinstance(camera, (str, mj.MjsCamera)):
            _, cam_name = self._resolve_camera_id_and_name(camera)
            camera_names = [cam_name]
        elif isinstance(camera, list):
            camera_names = [self._resolve_camera_id_and_name(c)[1] for c in camera]
        else:
            raise ValueError(
                f"Invalid camera spec type: {type(camera)}. Must be str, "
                "mj.MjsCamera, list of these, or None."
            )

        # Validate all cameras are available
        for cam_name in camera_names:
            if cam_name not in self._cameras_names2id:
                raise ValueError(
                    f"Camera '{cam_name}' is not available in this renderer. "
                    f"Available cameras: {list(self._cameras_names2id.keys())}"
                )

        return camera_names

    def _resolve_output_paths(
        self,
        output_path: dict[str | mj.MjsCamera, PathLike] | PathLike,
    ) -> dict[str, Path]:
        """Convert output_path specification to dict mapping camera names to Paths.

        Args:
            output_path: Either a dict mapping cameras to paths, or a single path

        Returns:
            Dict mapping camera name to output Path

        Raises:
            ValueError: If output_path format is invalid
        """
        if isinstance(output_path, dict):
            # Explicit mapping provided
            result = {}
            for cam_spec, path in output_path.items():
                _, cam_name = self._resolve_camera_id_and_name(cam_spec)
                if cam_name not in self._cameras_names2id:
                    raise ValueError(
                        f"Camera '{cam_name}' in output_path is not available. "
                        f"Available cameras: {list(self._cameras_names2id.keys())}"
                    )
                result[cam_name] = Path(path)
            return result

        # Single path provided - interpret based on number of cameras
        path = Path(output_path)
        available_cameras = list(self._cameras_names2id.keys())

        if len(available_cameras) == 1:
            # Single camera: interpret path as file
            return {available_cameras[0]: path}
        else:
            # Multiple cameras: interpret path as directory
            return {
                cam_name: path / f"{cam_name.replace('/', '_')}.mp4"
                for cam_name in available_cameras
            }

    def _resolve_camera_id_and_name(
        self, camera: str | mj.MjsCamera, /
    ) -> tuple[int, str]:
        """Convert a camera specification to (internal_id, camera_name)."""
        if isinstance(camera, str):
            cam_id = mj.mj_name2id(self.mj_model, mj.mjtObj.mjOBJ_CAMERA, camera)
            return cam_id, camera
        elif isinstance(camera, mj.MjsCamera):
            cam_name = camera.name
            cam_id = mj.mj_name2id(self.mj_model, mj.mjtObj.mjOBJ_CAMERA, cam_name)
            return cam_id, cam_name
        else:
            raise ValueError(
                f"Invalid camera spec: {camera}. Must be one of str or mj.MjsCamera."
            )

close()

Release the underlying MuJoCo renderer resources.

Source code in src/flygym/rendering.py
def close(self) -> None:
    """Release the underlying MuJoCo renderer resources."""
    self.mj_renderer.close()

get_camera_matrix(camera, mj_data, mj_model)

Get the 3x4 camera projection matrix from the current MjData.

The returned matrix maps homogeneous world coordinates to homogeneous image (pixel) coordinates, following the standard MuJoCo image @ focal @ rotation @ translation composition.

Source code in src/flygym/rendering.py
def get_camera_matrix(
    self, camera: str | mj.MjsCamera, mj_data: mj.MjData, mj_model: mj.MjModel
) -> np.ndarray:
    """Get the 3x4 camera projection matrix from the current MjData.

    The returned matrix maps homogeneous world coordinates to homogeneous
    image (pixel) coordinates, following the standard MuJoCo
    ``image @ focal @ rotation @ translation`` composition.
    """
    internal_cam_id, _ = self._resolve_camera_id_and_name(camera)
    # update the scene to get the latest camera position and orientation
    self.mj_renderer.update_scene(mj_data, internal_cam_id, self.scene_option)
    pos = mj_data.cam_xpos[internal_cam_id]
    rot = mj_data.cam_xmat[internal_cam_id].reshape(3, 3)
    fov = mj_model.cam_fovy[internal_cam_id]
    height, width = self.camera_res

    # Translation matrix (4x4).
    translation = np.eye(4)
    translation[0:3, 3] = -pos
    # Rotation matrix (4x4).
    rotation = np.eye(4)
    rotation[0:3, 0:3] = rot
    # Focal transformation matrix (3x4).
    focal_scaling = (1.0 / np.tan(np.deg2rad(fov) / 2)) * height / 2.0
    focal = np.diag([-focal_scaling, focal_scaling, 1.0, 0])[0:3, :]
    # Image matrix (3x3).
    image = np.eye(3)
    image[0, 2] = (width - 1) / 2.0
    image[1, 2] = (height - 1) / 2.0

    return image @ focal @ rotation @ translation

render_as_needed(mj_data)

Render frames for all cameras if enough time has elapsed.

Parameters:

Name Type Description Default
mj_data MjData

Current MuJoCo data.

required

Returns:

Type Description
bool

True if frames were rendered, False otherwise.

Source code in src/flygym/rendering.py
def render_as_needed(self, mj_data: mj.MjData) -> bool:
    """Render frames for all cameras if enough time has elapsed.

    Args:
        mj_data: Current MuJoCo data.

    Returns:
        True if frames were rendered, False otherwise.
    """
    if not self._clock.tick(mj_data.time):
        return False

    self._render_data.qpos[:] = mj_data.qpos
    self._render_data.mocap_pos[:] = mj_data.mocap_pos
    self._render_data.mocap_quat[:] = mj_data.mocap_quat
    update_render_geometry(self.mj_model, self._render_data)
    for cam_name, internal_cam_id in self._cameras_names2id.items():
        self.mj_renderer.update_scene(
            self._render_data, internal_cam_id, self.scene_option
        )
        # One render() pass per enabled output. mujoco renders a single output
        # type per call, so toggle the mode for each and always restore the
        # default (RGB) state afterwards.
        if self.render_rgb:
            rgb = self.mj_renderer.render()
            if self.buffer_frames:
                self.frames[cam_name].append(rgb)
        if self.render_depth:
            self.mj_renderer.enable_depth_rendering()
            depth = self.mj_renderer.render()
            self.mj_renderer.disable_depth_rendering()
            if self.buffer_frames:
                self.depth_frames[cam_name].append(depth)
        if self.render_segmentation:
            self.mj_renderer.enable_segmentation_rendering()
            # MuJoCo segmentation returns int32 (H, W, 2): channel 0 is the
            # object id, channel 1 the object type, with -1 marking background.
            # Keep channel 0 only (assumes objects are geoms, so the id is
            # unambiguous), copied to a standalone int32 array so the unused
            # second channel isn't retained.
            seg = self.mj_renderer.render()[:, :, 0].copy()
            self.mj_renderer.disable_segmentation_rendering()
            if self.buffer_frames:
                self.segmentation_frames[cam_name].append(seg)
    return True

reset()

Clear buffered frames and reset the render timer.

Source code in src/flygym/rendering.py
def reset(self) -> None:
    """Clear buffered frames and reset the render timer."""
    self._clock.reset()
    if not self.buffer_frames:
        return
    if self.render_rgb:
        self.frames = self._new_frame_buffer()
    if self.render_depth:
        self.depth_frames = self._new_frame_buffer()
    if self.render_segmentation:
        self.segmentation_frames = self._new_frame_buffer()

save_video(output_path, **kwargs)

Save recorded frames as video files.

Only RGB frames are saved. Depth/segmentation renders are kept as raw arrays in self.depth_frames / self.segmentation_frames.

Parameters:

Name Type Description Default
output_path dict[str | MjsCamera, PathLike] | PathLike

Either a dict mapping camera specs to file paths, or: - If single camera: a file path to save to - If multiple cameras: a directory path to save all videos to

required
**kwargs Any

Additional arguments passed to imageio.imwrite

{}
Source code in src/flygym/rendering.py
def save_video(
    self,
    output_path: dict[str | mj.MjsCamera, PathLike] | PathLike,
    **kwargs: Any,
) -> None:
    """Save recorded frames as video files.

    Only RGB frames are saved. Depth/segmentation renders are kept as raw
    arrays in ``self.depth_frames`` / ``self.segmentation_frames``.

    Args:
        output_path: Either a dict mapping camera specs to file paths, or:
            - If single camera: a file path to save to
            - If multiple cameras: a directory path to save all videos to
        **kwargs: Additional arguments passed to imageio.imwrite
    """
    if not self._warn_unless_rgb("saved"):
        return
    path_by_camera = self._resolve_output_paths(output_path)

    for cam_name, path in path_by_camera.items():
        frames = self.frames[cam_name]
        if len(frames) == 0:
            raise RuntimeError(f"No frames recorded yet for camera '{cam_name}'.")

        path.parent.mkdir(parents=True, exist_ok=True)

        iio.imwrite(
            path, frames, fps=self.output_fps, codec="libx264", quality=8, **kwargs
        )

show_in_notebook(camera=None, **kwargs)

Display recorded frames in a Jupyter notebook.

Only RGB frames are displayed. Depth/segmentation renders are kept as raw arrays in self.depth_frames / self.segmentation_frames.

Parameters:

Name Type Description Default
camera str | MjsCamera | list[str | MjsCamera] | None

Camera(s) to display. If None, displays all cameras.

None
**kwargs Any

Additional arguments passed to mediapy.show_video

{}
Source code in src/flygym/rendering.py
def show_in_notebook(
    self,
    camera: str | mj.MjsCamera | list[str | mj.MjsCamera] | None = None,
    **kwargs: Any,
) -> None:
    """Display recorded frames in a Jupyter notebook.

    Only RGB frames are displayed. Depth/segmentation renders are kept as raw
    arrays in ``self.depth_frames`` / ``self.segmentation_frames``.

    Args:
        camera: Camera(s) to display. If None, displays all cameras.
        **kwargs: Additional arguments passed to mediapy.show_video
    """
    if not self._warn_unless_rgb("displayed"):
        return
    camera_names = self._normalize_camera_spec(camera)

    for cam_name in camera_names:
        frames = self.frames[cam_name]
        if len(frames) == 0:
            raise RuntimeError(f"No frames recorded yet for camera '{cam_name}'.")
        mediapy.show_video(frames, fps=self.output_fps, title=cam_name, **kwargs)

launch_interactive_viewer(mj_model, mj_data, run_async=False, init_keyframe='neutral')

Launch MuJoCo's built-in interactive viewer.

Parameters:

Name Type Description Default
mj_model MjModel

Compiled MuJoCo model.

required
mj_data MjData

MuJoCo data.

required
run_async bool

If True, launch the viewer in a separate process and return immediately. Use this when calling from a Jupyter notebook.

False
init_keyframe str | None

Keyframe name to reset to before launching. Uses the current state if None.

'neutral'
Source code in src/flygym/rendering.py
def launch_interactive_viewer(
    mj_model: mj.MjModel,
    mj_data: mj.MjData,
    run_async: bool = False,
    init_keyframe: str | None = "neutral",
) -> None:
    """Launch MuJoCo's built-in interactive viewer.

    Args:
        mj_model: Compiled MuJoCo model.
        mj_data: MuJoCo data.
        run_async: If True, launch the viewer in a separate process and return
            immediately. Use this when calling from a Jupyter notebook.
        init_keyframe: Keyframe name to reset to before launching. Uses the current
            state if None.
    """

    if init_keyframe is not None:
        key_id = mj.mj_name2id(mj_model, mj.mjtObj.mjOBJ_KEY, init_keyframe)
        mj.mj_resetDataKeyframe(mj_model, mj_data, key_id)

    if run_async:
        p = Process(target=mj.viewer.launch, args=(mj_model, mj_data))
        p.start()
        # Don't join!
    else:
        mjviewer.launch(mj_model, mj_data)

preview_model(mj_model, mj_data, camera, *, init_keyframe='neutral', duration=0.1, camera_res=(240, 320), playback_speed=0.1, output_fps=25, show_in_notebook=False, output_path=None, **kwargs)

Run a short simulation and render a preview.

Parameters:

Name Type Description Default
mj_model MjModel

Compiled MuJoCo model.

required
mj_data MjData

MuJoCo data.

required
camera MjsCamera | str

Camera name or MJCF element to use for rendering.

required
init_keyframe str | None

Keyframe name to reset to before rendering. Uses the current state if None.

'neutral'
duration float

Duration to simulate in seconds.

0.1
camera_res tuple[int, int]

(height, width) in pixels.

(240, 320)
playback_speed float

Video playback speed relative to real time.

0.1
output_fps int

Output video frame rate.

25
show_in_notebook bool

If True, display the video in a Jupyter notebook.

False
output_path PathLike | None

Path to save the video. Not saved if None.

None
**kwargs Any

Passed to Renderer.

{}
Source code in src/flygym/rendering.py
def preview_model(
    mj_model: mj.MjModel,
    mj_data: mj.MjData,
    camera: mj.MjsCamera | str,
    *,
    init_keyframe: str | None = "neutral",
    duration: float = 0.1,
    camera_res: tuple[int, int] = (240, 320),
    playback_speed: float = 0.1,
    output_fps: int = 25,
    show_in_notebook: bool = False,
    output_path: PathLike | None = None,
    **kwargs: Any,
) -> None:
    """Run a short simulation and render a preview.

    Args:
        mj_model: Compiled MuJoCo model.
        mj_data: MuJoCo data.
        camera: Camera name or MJCF element to use for rendering.
        init_keyframe: Keyframe name to reset to before rendering. Uses the current
            state if None.
        duration: Duration to simulate in seconds.
        camera_res: ``(height, width)`` in pixels.
        playback_speed: Video playback speed relative to real time.
        output_fps: Output video frame rate.
        show_in_notebook: If True, display the video in a Jupyter notebook.
        output_path: Path to save the video. Not saved if None.
        **kwargs: Passed to `Renderer`.
    """
    if init_keyframe is not None:
        key_id = mj.mj_name2id(mj_model, mj.mjtObj.mjOBJ_KEY, init_keyframe)
        mj.mj_resetDataKeyframe(mj_model, mj_data, key_id)

    n_steps = int(duration / mj_model.opt.timestep)

    with Renderer(
        mj_model,
        camera,
        camera_res=camera_res,
        playback_speed=playback_speed,
        output_fps=output_fps,
        **kwargs,
    ) as renderer:
        for step in range(n_steps):
            mj.mj_step(mj_model, mj_data)
            renderer.render_as_needed(mj_data)

        if show_in_notebook:
            renderer.show_in_notebook()
        if output_path:
            renderer.save_video(output_path)

update_render_geometry(mj_model, mj_data)

Recompute everything that rendering depends on from qpos and mocap poses.

This runs the stages of mj_forward that determine rendered geometry (kinematics, center-of-mass positions, camera and light poses, and tendon paths). It skips collision detection and all velocity- and force-dependent stages, which do not affect rendering with default scene options.

Why renderers call this before rendering

MuJoCo's scene update does not read qpos. It reads derived positions such as xpos/xmat (bodies), geom_xpos/geom_xmat (geoms), and cam_xpos (cameras). mj_step computes these derived positions first and integrates afterwards, so right after a step, qpos is the new state while the derived positions still reflect the state at the start of the step. Rendering mj_data directly would therefore show the fly one timestep in the past.

FlyGym renderers instead copy qpos and mocap poses into a private MjData and call this function on it, so that frames are consistent with the current qpos and time. Trajectory recorders (flygym.recording) store only qpos and mocap poses, and replay calls this function in the same way. Live and replayed frames are therefore identical.

Recording the derived positions instead would also reproduce live frames, but it would take far more space (about 20 times more for NeuroMechFly), tie the recording format to the renderer's internals, and lose the ability to restore or analyze the simulation state from the recording. Recomputing the geometry is cheap compared to rendering itself.

Source code in src/flygym/rendering.py
def update_render_geometry(mj_model: mj.MjModel, mj_data: mj.MjData) -> None:
    """Recompute everything that rendering depends on from `qpos` and mocap poses.

    This runs the stages of `mj_forward` that determine rendered geometry (kinematics,
    center-of-mass positions, camera and light poses, and tendon paths). It skips
    collision detection and all velocity- and force-dependent stages, which do not
    affect rendering with default scene options.

    !!! note "Why renderers call this before rendering"

        MuJoCo's scene update does not read `qpos`. It reads derived positions such as
        `xpos`/`xmat` (bodies), `geom_xpos`/`geom_xmat` (geoms), and `cam_xpos`
        (cameras). `mj_step` computes these derived positions first and integrates
        afterwards, so right after a step, `qpos` is the new state while the derived
        positions still reflect the state at the start of the step. Rendering
        `mj_data` directly would therefore show the fly one timestep in the past.

        FlyGym renderers instead copy `qpos` and mocap poses into a private `MjData`
        and call this function on it, so that frames are consistent with the current
        `qpos` and time. Trajectory recorders (`flygym.recording`) store only `qpos`
        and mocap poses, and replay calls this function in the same way. Live and
        replayed frames are therefore identical.

        Recording the derived positions instead would also reproduce live frames, but
        it would take far more space (about 20 times more for NeuroMechFly), tie the
        recording format to the renderer's internals, and lose the ability to restore
        or analyze the simulation state from the recording. Recomputing the geometry
        is cheap compared to rendering itself.
    """
    mj.mj_kinematics(mj_model, mj_data)
    mj.mj_comPos(mj_model, mj_data)
    mj.mj_camlight(mj_model, mj_data)
    mj.mj_tendon(mj_model, mj_data)