Repository: Genesis-Embodied-AI/Genesis
File: genesis/recorders/file_writers.py (Python)

Code before:
    codec_options: dict[str, str]
        Additional low-level codec options that will be pass to ffmpeg. Empty by default.
    save_on_reset: bool, optional
        Whether to save the data on reset. If True, a counter will be added to the filename and incremented on each
        reset. Defaults to False.
    """

    fps: int | None = None
    name: str = ""
    codec: str = "libx264"
    bitrate: float = 1.0
    codec_options: dict[str, str] = Field(default_factory=dict)

    def validate(self):
        super().validate()

        if not self.codec in av.codecs_available:
            gs.raise_exception(f"[{type(self).__name__}] Codec '{self._options.codec}' not supported.")

        if not self.filename.endswith(".mp4"):
            gs.raise_exception(f"[{type(self).__name__}] Video filename must have '.mp4' extension.")


@register_recording(VideoFileWriterOptions)
class VideoFileWriter(BaseFileWriter):
    def build(self):
        if not IS_PYAV_AVAILABLE:
            gs.raise_exception("PyAV is not installed. Please install it with `pip install av`.")

        self.video_container: av.container.OutputContainer | None = None
        self.video_stream: av.video.stream.VideoStream | None = None
        self.video_frame: av.video.frame.VideoFrame | None = None
        self.video_buffer: np.ndarray | None = None

        self.fps = int(
            round(
                1.0 / (self._steps_per_sample * self._manager._step_dt)
                if self._options.fps is None
                else self._options.fps
            )
        )

        super().build()

    def _initialize_writer(self):
        video_path = self._get_filename()
        video_name = self._options.name or Path(video_path).stem

        # Create ffmpeg video container
        self.video_container = av.open(video_path, mode="w")
        self.video_container.metadata["title"] = video_name

Review comment (suggestion):
this line will run during import time, which will trigger error if `IS_PYAV_AVAILABLE=False`

Code after:
    codec_options: dict[str, str]
        Additional low-level codec options that will be pass to ffmpeg. Empty by default.
    save_on_reset: bool, optional
        Whether to save the data on reset. If True, a counter will be added to the filename and incremented on each
        reset. Defaults to False.
    """

    fps: int | None = None
    name: str = ""
    codec: str = "libx264"
    bitrate: float = 1.0
    codec_options: dict[str, str] = Field(default_factory=dict)

    def validate(self):
        super().validate()

        if not self.codec in av.codecs_available:
            gs.raise_exception(f"[{type(self).__name__}] Codec '{self._options.codec}' not supported.")

        if not self.filename.endswith(".mp4"):
            gs.raise_exception(f"[{type(self).__name__}] Video filename must have '.mp4' extension.")


@register_recording(VideoFileWriterOptions)
class VideoFileWriter(BaseFileWriter):
    video_container: "av.container.OutputContainer | None"
    video_stream: "av.video.stream.VideoStream | None"
    video_frame: "av.video.frame.VideoFrame | None"
    video_buffer: "np.ndarray | None"

    def build(self):
        if not IS_PYAV_AVAILABLE:
            gs.raise_exception("PyAV is not installed. Please install it with `pip install av`.")

        self.video_container = None
        self.video_stream = None
        self.video_frame = None
        self.video_buffer = None

        self.fps = int(
            round(
                1.0 / (self._steps_per_sample * self._manager._step_dt)
                if self._options.fps is None
                else self._options.fps
            )
        )

        super().build()

    def _initialize_writer(self):
        video_path = self._get_filename()