Method __init__

detrsmpl/utils/ffmpeg_utils.py:23–73

(self,
                 output_path: str,
                 resolution: Iterable[int],
                 fps: float = 30.0,
                 num_frame: int = 1e9,
                 disable_log: bool = False)

class video_writer:

    def __init__(self,
                 output_path: str,
                 resolution: Iterable[int],
                 fps: float = 30.0,
                 num_frame: int = 1e9,
                 disable_log: bool = False) -> None:
        prepare_output_path(
            output_path,
            allowed_suffix=['.mp4'],
            tag='output video',
            path_type='file',
            overwrite=True)
        height, width = resolution
        width += width % 2
        height += height % 2
        command = [
            'ffmpeg',
            '-y',  # (optional) overwrite output file if it exists
            '-f',
            'rawvideo',
            '-pix_fmt',
            'bgr24',
            '-s',
            f'{int(width)}x{int(height)}',
            '-r',
            f'{fps}',  # frames per second
            '-loglevel',
            'error',
            '-threads',
            '1',
            '-i',
            '-',  # The input comes from a pipe
            '-vcodec',
            'libx264',
            '-r',
            f'{fps}',  # frames per second
            '-an',  # Tells FFMPEG not to expect any audio
            output_path,
        ]
        if not disable_log:
            print(f'Running \"{" ".join(command)}\"')
        process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if process.stdin is None or process.stderr is None:
            raise BrokenPipeError('No buffer received.')
        self.process = process
        self.num_frame = num_frame
        self.len = 0

    def write(self, image_array: np.ndarray):
        if self.len <= self.num_frame: