Spaces:
Running
Running
| """ | |
| FFMPEG_Writer - write set of frames to video file | |
| original from | |
| https://github.com/Zulko/moviepy/blob/master/moviepy/video/io/ffmpeg_writer.py | |
| removed unnecessary dependencies | |
| The MIT License (MIT) | |
| Copyright (c) 2015 Zulko | |
| Copyright (c) 2023 Janvarev Vladislav | |
| """ | |
| import os | |
| import subprocess as sp | |
| PIPE = -1 | |
| STDOUT = -2 | |
| DEVNULL = -3 | |
| FFMPEG_BINARY = "ffmpeg" | |
| class FFMPEG_VideoWriter: | |
| """ A class for FFMPEG-based video writing. | |
| A class to write videos using ffmpeg. ffmpeg will write in a large | |
| choice of formats. | |
| Parameters | |
| ----------- | |
| filename | |
| Any filename like 'video.mp4' etc. but if you want to avoid | |
| complications it is recommended to use the generic extension | |
| '.avi' for all your videos. | |
| size | |
| Size (width,height) of the output video in pixels. | |
| fps | |
| Frames per second in the output video file. | |
| codec | |
| FFMPEG codec. It seems that in terms of quality the hierarchy is | |
| 'rawvideo' = 'png' > 'mpeg4' > 'libx264' | |
| 'png' manages the same lossless quality as 'rawvideo' but yields | |
| smaller files. Type ``ffmpeg -codecs`` in a terminal to get a list | |
| of accepted codecs. | |
| Note for default 'libx264': by default the pixel format yuv420p | |
| is used. If the video dimensions are not both even (e.g. 720x405) | |
| another pixel format is used, and this can cause problem in some | |
| video readers. | |
| audiofile | |
| Optional: The name of an audio file that will be incorporated | |
| to the video. | |
| preset | |
| Sets the time that FFMPEG will take to compress the video. The slower, | |
| the better the compression rate. Possibilities are: ultrafast,superfast, | |
| veryfast, faster, fast, medium (default), slow, slower, veryslow, | |
| placebo. | |
| bitrate | |
| Only relevant for codecs which accept a bitrate. "5000k" offers | |
| nice results in general. | |
| """ | |
| def __init__(self, filename, size, fps, codec="libx265", crf=14, audiofile=None, | |
| preset="medium", bitrate=None, | |
| logfile=None, threads=None, ffmpeg_params=None): | |
| if logfile is None: | |
| logfile = sp.PIPE | |
| self.filename = filename | |
| self.codec = codec | |
| self.ext = self.filename.split(".")[-1] | |
| w = size[0] - 1 if size[0] % 2 != 0 else size[0] | |
| h = size[1] - 1 if size[1] % 2 != 0 else size[1] | |
| # order is important | |
| cmd = [ | |
| FFMPEG_BINARY, | |
| '-hide_banner', | |
| '-hwaccel', 'auto', | |
| '-y', | |
| '-loglevel', 'error' if logfile == sp.PIPE else 'info', | |
| '-f', 'rawvideo', | |
| '-vcodec', 'rawvideo', | |
| '-s', '%dx%d' % (size[0], size[1]), | |
| #'-pix_fmt', 'rgba' if withmask else 'rgb24', | |
| '-pix_fmt', 'bgr24', | |
| '-r', str(fps), | |
| '-an', '-i', '-' | |
| ] | |
| if audiofile is not None: | |
| cmd.extend([ | |
| '-i', audiofile, | |
| '-acodec', 'copy' | |
| ]) | |
| cmd.extend([ | |
| '-vcodec', codec, | |
| '-crf', str(crf) | |
| #'-preset', preset, | |
| ]) | |
| if ffmpeg_params is not None: | |
| cmd.extend(ffmpeg_params) | |
| if bitrate is not None: | |
| cmd.extend([ | |
| '-b', bitrate | |
| ]) | |
| # scale to a resolution divisible by 2 if not even | |
| cmd.extend(['-vf', f'scale={w}:{h}' if w != size[0] or h != size[1] else 'colorspace=bt709:iall=bt601-6-625:fast=1']) | |
| if threads is not None: | |
| cmd.extend(["-threads", str(threads)]) | |
| cmd.extend([ | |
| '-pix_fmt', 'yuv420p', | |
| ]) | |
| cmd.extend([ | |
| filename | |
| ]) | |
| test = str(cmd) | |
| print(test) | |
| popen_params = {"stdout": DEVNULL, | |
| "stderr": logfile, | |
| "stdin": sp.PIPE} | |
| # This was added so that no extra unwanted window opens on windows | |
| # when the child process is created | |
| if os.name == "nt": | |
| popen_params["creationflags"] = 0x08000000 # CREATE_NO_WINDOW | |
| self.proc = sp.Popen(cmd, **popen_params) | |
| def write_frame(self, img_array): | |
| """ Writes one frame in the file.""" | |
| try: | |
| #if PY3: | |
| self.proc.stdin.write(img_array.tobytes()) | |
| # else: | |
| # self.proc.stdin.write(img_array.tostring()) | |
| except IOError as err: | |
| _, ffmpeg_error = self.proc.communicate() | |
| error = (str(err) + ("\n\nMoviePy error: FFMPEG encountered " | |
| "the following error while writing file %s:" | |
| "\n\n %s" % (self.filename, str(ffmpeg_error)))) | |
| if b"Unknown encoder" in ffmpeg_error: | |
| error = error+("\n\nThe video export " | |
| "failed because FFMPEG didn't find the specified " | |
| "codec for video encoding (%s). Please install " | |
| "this codec or change the codec when calling " | |
| "write_videofile. For instance:\n" | |
| " >>> clip.write_videofile('myvid.webm', codec='libvpx')")%(self.codec) | |
| elif b"incorrect codec parameters ?" in ffmpeg_error: | |
| error = error+("\n\nThe video export " | |
| "failed, possibly because the codec specified for " | |
| "the video (%s) is not compatible with the given " | |
| "extension (%s). Please specify a valid 'codec' " | |
| "argument in write_videofile. This would be 'libx264' " | |
| "or 'mpeg4' for mp4, 'libtheora' for ogv, 'libvpx for webm. " | |
| "Another possible reason is that the audio codec was not " | |
| "compatible with the video codec. For instance the video " | |
| "extensions 'ogv' and 'webm' only allow 'libvorbis' (default) as a" | |
| "video codec." | |
| )%(self.codec, self.ext) | |
| elif b"encoder setup failed" in ffmpeg_error: | |
| error = error+("\n\nThe video export " | |
| "failed, possibly because the bitrate you specified " | |
| "was too high or too low for the video codec.") | |
| elif b"Invalid encoder type" in ffmpeg_error: | |
| error = error + ("\n\nThe video export failed because the codec " | |
| "or file extension you provided is not a video") | |
| raise IOError(error) | |
| def close(self): | |
| if self.proc: | |
| self.proc.stdin.close() | |
| if self.proc.stderr is not None: | |
| self.proc.stderr.close() | |
| self.proc.wait() | |
| self.proc = None | |
| # Support the Context Manager protocol, to ensure that resources are cleaned up. | |
| def __enter__(self): | |
| return self | |
| def __exit__(self, exc_type, exc_value, traceback): | |
| self.close() | |
| def ffmpeg_write_image(filename, image, logfile=False): | |
| """ Writes an image (HxWx3 or HxWx4 numpy array) to a file, using | |
| ffmpeg. """ | |
| if image.dtype != 'uint8': | |
| image = image.astype("uint8") | |
| cmd = [ FFMPEG_BINARY, '-y', | |
| '-s', "%dx%d"%(image.shape[:2][::-1]), | |
| "-f", 'rawvideo', | |
| '-pix_fmt', "rgba" if (image.shape[2] == 4) else "rgb24", | |
| '-i','-', filename] | |
| if logfile: | |
| log_file = open(filename + ".log", 'w+') | |
| else: | |
| log_file = sp.PIPE | |
| popen_params = {"stdout": DEVNULL, | |
| "stderr": log_file, | |
| "stdin": sp.PIPE} | |
| if os.name == "nt": | |
| popen_params["creationflags"] = 0x08000000 | |
| proc = sp.Popen(cmd, **popen_params) | |
| out, err = proc.communicate(image.tostring()) | |
| if proc.returncode: | |
| err = "\n".join(["[MoviePy] Running : %s\n" % cmd, | |
| "WARNING: this command returned an error:", | |
| err.decode('utf8')]) | |
| raise IOError(err) | |
| del proc | |