opentau.utils.io_utils

Utilities for file I/O operations.

This module provides functions for reading and writing JSON files, saving videos, and deserializing JSON data into structured objects with type checking. It also provides silence_output_unless_error() for muting noisy subprocess output (captured and replayed only if the wrapped block raises).

Functions

deserialize_json_into_object(fpath, obj)

Load JSON data and recursively fill an object with matching structure.

silence_output_unless_error([label])

Mute this process's stdout/stderr for the block, replaying it only on error.

write_video(video_path, stacked_frames, fps)

Write a list of frames to a video file.
opentau.utils.io_utils.deserialize_json_into_object(fpath: Path, obj: T) → T[source]

Load JSON data and recursively fill an object with matching structure.

Loads the JSON data from fpath and recursively fills obj with the corresponding values (strictly matching structure and types). Tuples in obj are expected to be lists in the JSON data, which will be converted back into tuples.

Parameters:

  • fpath – Path to the JSON file to load.

  • obj – Template object with the desired structure and types.

Returns:

Object with the same structure as obj, filled with values from the JSON file.

Raises:

  • TypeError – If structure or types don’t match between JSON and obj.

  • ValueError – If dictionary keys or list/tuple lengths don’t match.

opentau.utils.io_utils.silence_output_unless_error(label: str = '') → Iterator[None][source]

Mute this process’s stdout/stderr for the block, replaying it only on error.

Redirection happens at the file-descriptor level (os.dup2 on fds 1 and 2), not by swapping sys.stdout / sys.stderr. That is what lets it capture output a Python-level contextlib.redirect_stdout() would miss: writes from C extensions (mujoco / EGL) and logging handlers that grabbed a reference to the original stream at import time (e.g. robosuite’s logger).

The motivating use is muting the noisy one-per-worker robosuite/robocasa import-and-construction banner emitted inside gym.vector.AsyncVectorEnv spawn workers during sim eval — n_envs * world_size duplicate copies of the private-macro / mink / mimicgen / controller-config lines on every eval. On success the captured output is discarded; if the wrapped block raises, the captured bytes are written to the real stderr first (prefixed with label) so the failing worker stays debuggable, then the exception propagates unchanged.

Parameters:

label – Optional identifier (e.g. "task=CloseFridge idx=3") prepended to the replayed output so a failing worker can be attributed.

Yields:

None. Wrap the code whose output should be muted in the with block.

opentau.utils.io_utils.write_video(video_path: str | Path, stacked_frames: list, fps: float) → None[source]

Write a list of frames to a video file.

Parameters:

  • video_path – Path where the video file will be saved.

  • stacked_frames – List of image frames to write.

  • fps – Frames per second for the output video.