diff --git a/.gitignore b/.gitignore index 780f20e4..bab70e1d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +.vscode .cache .eggs .tox/ diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 00000000..30e226a3 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "python.pythonPath": "c:\\Users\\Dyam\\Downloads\\ffmpeg-python\\venv\\Scripts\\python.exe" +} \ No newline at end of file diff --git a/examples/facetime.py b/examples/facetime.py index 58d083ec..efb71b00 100644 --- a/examples/facetime.py +++ b/examples/facetime.py @@ -1,8 +1,8 @@ import ffmpeg ( - ffmpeg - .input('FaceTime', format='avfoundation', pix_fmt='uyvy422', framerate=30) - .output('out.mp4', pix_fmt='yuv420p', vframes=100) - .run() + ffmpeg + .input('FaceTime', format='avfoundation', pix_fmt='uyvy422', framerate=30) + .output('out.mp4', pix_fmt='yuv420p', vframes=100) + .run() ) diff --git a/examples/ffmpeg-numpy.ipynb b/examples/ffmpeg-numpy.ipynb index b6d991bf..2bbf9836 100644 --- a/examples/ffmpeg-numpy.ipynb +++ b/examples/ffmpeg-numpy.ipynb @@ -213,4 +213,4 @@ }, "nbformat": 4, "nbformat_minor": 2 -} +} \ No newline at end of file diff --git a/examples/get_video_thumbnail.py b/examples/get_video_thumbnail.py index b905642f..c811f748 100755 --- a/examples/get_video_thumbnail.py +++ b/examples/get_video_thumbnail.py @@ -32,4 +32,5 @@ def generate_thumbnail(in_filename, out_filename, time, width): if __name__ == '__main__': args = parser.parse_args() - generate_thumbnail(args.in_filename, args.out_filename, args.time, args.width) + generate_thumbnail(args.in_filename, args.out_filename, + args.time, args.width) diff --git a/examples/show_progress.py b/examples/show_progress.py index dd0253a1..95b4f5e5 100755 --- a/examples/show_progress.py +++ b/examples/show_progress.py @@ -15,15 +15,15 @@ parser = argparse.ArgumentParser(description=textwrap.dedent('''\ - Process video and report and show progress bar. + Process video and report and show progress bar. - This is an example of using the ffmpeg `-progress` option with a - unix-domain socket to report progress in the form of a progress - bar. + This is an example of using the ffmpeg `-progress` option with a + unix-domain socket to report progress in the form of a progress + bar. - The video processing simply consists of converting the video to - sepia colors, but the same pattern can be applied to other use - cases. + The video processing simply consists of converting the video to + sepia colors, but the same pattern can be applied to other use + cases. ''')) parser.add_argument('in_filename', help='Input filename') @@ -32,99 +32,99 @@ @contextlib.contextmanager def _tmpdir_scope(): - tmpdir = tempfile.mkdtemp() - try: - yield tmpdir - finally: - shutil.rmtree(tmpdir) + tmpdir = tempfile.mkdtemp() + try: + yield tmpdir + finally: + shutil.rmtree(tmpdir) def _do_watch_progress(filename, sock, handler): - """Function to run in a separate gevent greenlet to read progress - events from a unix-domain socket.""" - connection, client_address = sock.accept() - data = b'' - try: - while True: - more_data = connection.recv(16) - if not more_data: - break - data += more_data - lines = data.split(b'\n') - for line in lines[:-1]: - line = line.decode() - parts = line.split('=') - key = parts[0] if len(parts) > 0 else None - value = parts[1] if len(parts) > 1 else None - handler(key, value) - data = lines[-1] - finally: - connection.close() + """Function to run in a separate gevent greenlet to read progress + events from a unix-domain socket.""" + connection, client_address = sock.accept() + data = b'' + try: + while True: + more_data = connection.recv(16) + if not more_data: + break + data += more_data + lines = data.split(b'\n') + for line in lines[:-1]: + line = line.decode() + parts = line.split('=') + key = parts[0] if len(parts) > 0 else None + value = parts[1] if len(parts) > 1 else None + handler(key, value) + data = lines[-1] + finally: + connection.close() @contextlib.contextmanager def _watch_progress(handler): - """Context manager for creating a unix-domain socket and listen for - ffmpeg progress events. - - The socket filename is yielded from the context manager and the - socket is closed when the context manager is exited. - - Args: - handler: a function to be called when progress events are - received; receives a ``key`` argument and ``value`` - argument. (The example ``show_progress`` below uses tqdm) - - Yields: - socket_filename: the name of the socket file. - """ - with _tmpdir_scope() as tmpdir: - socket_filename = os.path.join(tmpdir, 'sock') - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - with contextlib.closing(sock): - sock.bind(socket_filename) - sock.listen(1) - child = gevent.spawn(_do_watch_progress, socket_filename, sock, handler) - try: - yield socket_filename - except: - gevent.kill(child) - raise + """Context manager for creating a unix-domain socket and listen for + ffmpeg progress events. + + The socket filename is yielded from the context manager and the + socket is closed when the context manager is exited. + + Args: + handler: a function to be called when progress events are + received; receives a ``key`` argument and ``value`` + argument. (The example ``show_progress`` below uses tqdm) + + Yields: + socket_filename: the name of the socket file. + """ + with _tmpdir_scope() as tmpdir: + socket_filename = os.path.join(tmpdir, 'sock') + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + with contextlib.closing(sock): + sock.bind(socket_filename) + sock.listen(1) + child = gevent.spawn(_do_watch_progress, socket_filename, sock, handler) + try: + yield socket_filename + except: + gevent.kill(child) + raise @contextlib.contextmanager def show_progress(total_duration): - """Create a unix-domain socket to watch progress and render tqdm - progress bar.""" - with tqdm(total=round(total_duration, 2)) as bar: - def handler(key, value): - if key == 'out_time_ms': - time = round(float(value) / 1000000., 2) - bar.update(time - bar.n) - elif key == 'progress' and value == 'end': - bar.update(bar.total - bar.n) - with _watch_progress(handler) as socket_filename: - yield socket_filename + """Create a unix-domain socket to watch progress and render tqdm + progress bar.""" + with tqdm(total=round(total_duration, 2)) as bar: + def handler(key, value): + if key == 'out_time_ms': + time = round(float(value) / 1000000., 2) + bar.update(time - bar.n) + elif key == 'progress' and value == 'end': + bar.update(bar.total - bar.n) + with _watch_progress(handler) as socket_filename: + yield socket_filename if __name__ == '__main__': - args = parser.parse_args() - total_duration = float(ffmpeg.probe(args.in_filename)['format']['duration']) - - with show_progress(total_duration) as socket_filename: - # See https://ffmpeg.org/ffmpeg-filters.html#Examples-44 - sepia_values = [.393, .769, .189, 0, .349, .686, .168, 0, .272, .534, .131] - try: - (ffmpeg - .input(args.in_filename) - .colorchannelmixer(*sepia_values) - .output(args.out_filename) - .global_args('-progress', 'unix://{}'.format(socket_filename)) - .overwrite_output() - .run(capture_stdout=True, capture_stderr=True) - ) - except ffmpeg.Error as e: - print(e.stderr, file=sys.stderr) - sys.exit(1) + args = parser.parse_args() + total_duration = float(ffmpeg.probe(args.in_filename)['format']['duration']) + + with show_progress(total_duration) as socket_filename: + # See https://ffmpeg.org/ffmpeg-filters.html#Examples-44 + sepia_values = [.393, .769, .189, 0, .349, .686, .168, 0, .272, .534, .131] + try: + (ffmpeg + .input(args.in_filename) + .colorchannelmixer(*sepia_values) + .output(args.out_filename) + .global_args('-progress', 'unix://{}'.format(socket_filename)) + .overwrite_output() + .run(capture_stdout=True, capture_stderr=True) + ) + except ffmpeg.Error as e: + print(e.stderr, file=sys.stderr) + sys.exit(1) diff --git a/examples/split_silence.py b/examples/split_silence.py index a889db10..640fc54b 100755 --- a/examples/split_silence.py +++ b/examples/split_silence.py @@ -18,124 +18,132 @@ DEFAULT_DURATION = 0.3 DEFAULT_THRESHOLD = -60 -parser = argparse.ArgumentParser(description='Split media into separate chunks wherever silence occurs') +parser = argparse.ArgumentParser( + description='Split media into separate chunks wherever silence occurs') parser.add_argument('in_filename', help='Input filename (`-` for stdin)') -parser.add_argument('out_pattern', help='Output filename pattern (e.g. `out/chunk_{:04d}.wav`)') -parser.add_argument('--silence-threshold', default=DEFAULT_THRESHOLD, type=int, help='Silence threshold (in dB)') -parser.add_argument('--silence-duration', default=DEFAULT_DURATION, type=float, help='Silence duration') +parser.add_argument( + 'out_pattern', help='Output filename pattern (e.g. `out/chunk_{:04d}.wav`)') +parser.add_argument('--silence-threshold', default=DEFAULT_THRESHOLD, + type=int, help='Silence threshold (in dB)') +parser.add_argument('--silence-duration', default=DEFAULT_DURATION, + type=float, help='Silence duration') parser.add_argument('--start-time', type=float, help='Start time (seconds)') parser.add_argument('--end-time', type=float, help='End time (seconds)') -parser.add_argument('-v', dest='verbose', action='store_true', help='Verbose mode') +parser.add_argument('-v', dest='verbose', + action='store_true', help='Verbose mode') -silence_start_re = re.compile(' silence_start: (?P[0-9]+(\.?[0-9]*))$') -silence_end_re = re.compile(' silence_end: (?P[0-9]+(\.?[0-9]*)) ') +silence_start_re = re.compile(' silence_start: (?P[0-9]+(\\.?[0-9]*))$') +silence_end_re = re.compile(' silence_end: (?P[0-9]+(\\.?[0-9]*)) ') total_duration_re = re.compile( - 'size=[^ ]+ time=(?P[0-9]{2}):(?P[0-9]{2}):(?P[0-9\.]{5}) bitrate=') + 'size=[^ ]+ time=(?P[0-9]{2}):(?P[0-9]{2}):(?P[0-9\\.]{5}) bitrate=') def _logged_popen(cmd_line, *args, **kwargs): - logger.debug('Running command: {}'.format(subprocess.list2cmdline(cmd_line))) - return subprocess.Popen(cmd_line, *args, **kwargs) + logger.debug('Running command: {}'.format( + subprocess.list2cmdline(cmd_line))) + return subprocess.Popen(cmd_line, *args, **kwargs) def get_chunk_times(in_filename, silence_threshold, silence_duration, start_time=None, end_time=None): - input_kwargs = {} - if start_time is not None: - input_kwargs['ss'] = start_time - else: - start_time = 0. - if end_time is not None: - input_kwargs['t'] = end_time - start_time - - p = _logged_popen( - (ffmpeg - .input(in_filename, **input_kwargs) - .filter('silencedetect', n='{}dB'.format(silence_threshold), d=silence_duration) - .output('-', format='null') - .compile() - ) + ['-nostats'], # FIXME: use .nostats() once it's implemented in ffmpeg-python. - stderr=subprocess.PIPE - ) - output = p.communicate()[1].decode('utf-8') - if p.returncode != 0: - sys.stderr.write(output) - sys.exit(1) - logger.debug(output) - lines = output.splitlines() - - # Chunks start when silence ends, and chunks end when silence starts. - chunk_starts = [] - chunk_ends = [] - for line in lines: - silence_start_match = silence_start_re.search(line) - silence_end_match = silence_end_re.search(line) - total_duration_match = total_duration_re.search(line) - if silence_start_match: - chunk_ends.append(float(silence_start_match.group('start'))) - if len(chunk_starts) == 0: - # Started with non-silence. - chunk_starts.append(start_time or 0.) - elif silence_end_match: - chunk_starts.append(float(silence_end_match.group('end'))) - elif total_duration_match: - hours = int(total_duration_match.group('hours')) - minutes = int(total_duration_match.group('minutes')) - seconds = float(total_duration_match.group('seconds')) - end_time = hours * 3600 + minutes * 60 + seconds - - if len(chunk_starts) == 0: - # No silence found. - chunk_starts.append(start_time) - - if len(chunk_starts) > len(chunk_ends): - # Finished with non-silence. - chunk_ends.append(end_time or 10000000.) - - return list(zip(chunk_starts, chunk_ends)) + input_kwargs = {} + if start_time is not None: + input_kwargs['ss'] = start_time + else: + start_time = 0. + if end_time is not None: + input_kwargs['t'] = end_time - start_time + + p = _logged_popen( + (ffmpeg + .input(in_filename, **input_kwargs) + .filter('silencedetect', n='{}dB'.format(silence_threshold), d=silence_duration) + .output('-', format='null') + .compile() + ) + ['-nostats'], # FIXME: use .nostats() once it's implemented in ffmpeg-python. + stderr=subprocess.PIPE + ) + output = p.communicate()[1].decode('utf-8') + if p.returncode != 0: + sys.stderr.write(output) + sys.exit(1) + logger.debug(output) + lines = output.splitlines() + + # Chunks start when silence ends, and chunks end when silence starts. + chunk_starts = [] + chunk_ends = [] + for line in lines: + silence_start_match = silence_start_re.search(line) + silence_end_match = silence_end_re.search(line) + total_duration_match = total_duration_re.search(line) + if silence_start_match: + chunk_ends.append(float(silence_start_match.group('start'))) + if len(chunk_starts) == 0: + # Started with non-silence. + chunk_starts.append(start_time or 0.) + elif silence_end_match: + chunk_starts.append(float(silence_end_match.group('end'))) + elif total_duration_match: + hours = int(total_duration_match.group('hours')) + minutes = int(total_duration_match.group('minutes')) + seconds = float(total_duration_match.group('seconds')) + end_time = hours * 3600 + minutes * 60 + seconds + + if len(chunk_starts) == 0: + # No silence found. + chunk_starts.append(start_time) + + if len(chunk_starts) > len(chunk_ends): + # Finished with non-silence. + chunk_ends.append(end_time or 10000000.) + + return list(zip(chunk_starts, chunk_ends)) def _makedirs(path): - """Python2-compatible version of ``os.makedirs(path, exist_ok=True)``.""" - try: - os.makedirs(path) - except OSError as exc: - if exc.errno != errno.EEXIST or not os.path.isdir(path): - raise + """Python2-compatible version of ``os.makedirs(path, exist_ok=True)``.""" + try: + os.makedirs(path) + except OSError as exc: + if exc.errno != errno.EEXIST or not os.path.isdir(path): + raise def split_audio( - in_filename, - out_pattern, - silence_threshold=DEFAULT_THRESHOLD, - silence_duration=DEFAULT_DURATION, - start_time=None, - end_time=None, - verbose=False, + in_filename, + out_pattern, + silence_threshold=DEFAULT_THRESHOLD, + silence_duration=DEFAULT_DURATION, + start_time=None, + end_time=None, + verbose=False, ): - chunk_times = get_chunk_times(in_filename, silence_threshold, silence_duration, start_time, end_time) - - for i, (start_time, end_time) in enumerate(chunk_times): - time = end_time - start_time - out_filename = out_pattern.format(i, i=i) - _makedirs(os.path.dirname(out_filename)) - - logger.info('{}: start={:.02f}, end={:.02f}, duration={:.02f}'.format(out_filename, start_time, end_time, - time)) - _logged_popen( - (ffmpeg - .input(in_filename, ss=start_time, t=time) - .output(out_filename) - .overwrite_output() - .compile() - ), - stdout=subprocess.PIPE if not verbose else None, - stderr=subprocess.PIPE if not verbose else None, - ).communicate() + chunk_times = get_chunk_times( + in_filename, silence_threshold, silence_duration, start_time, end_time) + + for i, (start_time, end_time) in enumerate(chunk_times): + time = end_time - start_time + out_filename = out_pattern.format(i, i=i) + _makedirs(os.path.dirname(out_filename)) + + logger.info('{}: start={:.02f}, end={:.02f}, duration={:.02f}'.format(out_filename, start_time, end_time, + time)) + _logged_popen( + (ffmpeg + .input(in_filename, ss=start_time, t=time) + .output(out_filename) + .overwrite_output() + .compile() + ), + stdout=subprocess.PIPE if not verbose else None, + stderr=subprocess.PIPE if not verbose else None, + ).communicate() if __name__ == '__main__': - kwargs = vars(parser.parse_args()) - if kwargs['verbose']: - logging.basicConfig(level=logging.DEBUG, format='%(levels): %(message)s') - logger.setLevel(logging.DEBUG) - split_audio(**kwargs) + kwargs = vars(parser.parse_args()) + if kwargs['verbose']: + logging.basicConfig(level=logging.DEBUG, + format='%(levels): %(message)s') + logger.setLevel(logging.DEBUG) + split_audio(**kwargs) diff --git a/examples/tensorflow_stream.py b/examples/tensorflow_stream.py index 6c9c9c9d..7d594626 100644 --- a/examples/tensorflow_stream.py +++ b/examples/tensorflow_stream.py @@ -34,215 +34,222 @@ import zipfile -parser = argparse.ArgumentParser(description='Example streaming ffmpeg numpy processing') +parser = argparse.ArgumentParser( + description='Example streaming ffmpeg numpy processing') parser.add_argument('in_filename', help='Input filename') parser.add_argument('out_filename', help='Output filename') parser.add_argument( - '--dream', action='store_true', help='Use DeepDream frame processing (requires tensorflow)') + '--dream', action='store_true', help='Use DeepDream frame processing (requires tensorflow)') logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) def get_video_size(filename): - logger.info('Getting video size for {!r}'.format(filename)) - probe = ffmpeg.probe(filename) - video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video') - width = int(video_info['width']) - height = int(video_info['height']) - return width, height + logger.info('Getting video size for {!r}'.format(filename)) + probe = ffmpeg.probe(filename) + video_info = next(s for s in probe['streams'] + if s['codec_type'] == 'video') + width = int(video_info['width']) + height = int(video_info['height']) + return width, height def start_ffmpeg_process1(in_filename): - logger.info('Starting ffmpeg process1') - args = ( - ffmpeg - .input(in_filename) - .output('pipe:', format='rawvideo', pix_fmt='rgb24') - .compile() - ) - return subprocess.Popen(args, stdout=subprocess.PIPE) + logger.info('Starting ffmpeg process1') + args = ( + ffmpeg + .input(in_filename) + .output('pipe:', format='rawvideo', pix_fmt='rgb24') + .compile() + ) + return subprocess.Popen(args, stdout=subprocess.PIPE) def start_ffmpeg_process2(out_filename, width, height): - logger.info('Starting ffmpeg process2') - args = ( - ffmpeg - .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height)) - .output(out_filename, pix_fmt='yuv420p') - .overwrite_output() - .compile() - ) - return subprocess.Popen(args, stdin=subprocess.PIPE) + logger.info('Starting ffmpeg process2') + args = ( + ffmpeg + .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height)) + .output(out_filename, pix_fmt='yuv420p') + .overwrite_output() + .compile() + ) + return subprocess.Popen(args, stdin=subprocess.PIPE) def read_frame(process1, width, height): - logger.debug('Reading frame') - - # Note: RGB24 == 3 bytes per pixel. - frame_size = width * height * 3 - in_bytes = process1.stdout.read(frame_size) - if len(in_bytes) == 0: - frame = None - else: - assert len(in_bytes) == frame_size - frame = ( - np - .frombuffer(in_bytes, np.uint8) - .reshape([height, width, 3]) - ) - return frame + logger.debug('Reading frame') + + # Note: RGB24 == 3 bytes per pixel. + frame_size = width * height * 3 + in_bytes = process1.stdout.read(frame_size) + if len(in_bytes) == 0: + frame = None + else: + assert len(in_bytes) == frame_size + frame = ( + np + .frombuffer(in_bytes, np.uint8) + .reshape([height, width, 3]) + ) + return frame def process_frame_simple(frame): - '''Simple processing example: darken frame.''' - return frame * 0.3 + '''Simple processing example: darken frame.''' + return frame * 0.3 def write_frame(process2, frame): - logger.debug('Writing frame') - process2.stdin.write( - frame - .astype(np.uint8) - .tobytes() - ) + logger.debug('Writing frame') + process2.stdin.write( + frame + .astype(np.uint8) + .tobytes() + ) def run(in_filename, out_filename, process_frame): - width, height = get_video_size(in_filename) - process1 = start_ffmpeg_process1(in_filename) - process2 = start_ffmpeg_process2(out_filename, width, height) - while True: - in_frame = read_frame(process1, width, height) - if in_frame is None: - logger.info('End of input stream') - break + width, height = get_video_size(in_filename) + process1 = start_ffmpeg_process1(in_filename) + process2 = start_ffmpeg_process2(out_filename, width, height) + while True: + in_frame = read_frame(process1, width, height) + if in_frame is None: + logger.info('End of input stream') + break - logger.debug('Processing frame') - out_frame = process_frame(in_frame) - write_frame(process2, out_frame) + logger.debug('Processing frame') + out_frame = process_frame(in_frame) + write_frame(process2, out_frame) - logger.info('Waiting for ffmpeg process1') - process1.wait() + logger.info('Waiting for ffmpeg process1') + process1.wait() - logger.info('Waiting for ffmpeg process2') - process2.stdin.close() - process2.wait() + logger.info('Waiting for ffmpeg process2') + process2.stdin.close() + process2.wait() - logger.info('Done') + logger.info('Done') class DeepDream(object): - '''DeepDream implementation, adapted from official tensorflow deepdream tutorial: - https://github.com/tensorflow/tensorflow/tree/master/tensorflow/examples/tutorials/deepdream - - Credit: Alexander Mordvintsev - ''' - - _DOWNLOAD_URL = 'https://storage.googleapis.com/download.tensorflow.org/models/inception5h.zip' - _ZIP_FILENAME = 'deepdream_model.zip' - _MODEL_FILENAME = 'tensorflow_inception_graph.pb' - - @staticmethod - def _download_model(): - logger.info('Downloading deepdream model...') - try: - from urllib.request import urlretrieve # python 3 - except ImportError: - from urllib import urlretrieve # python 2 - urlretrieve(DeepDream._DOWNLOAD_URL, DeepDream._ZIP_FILENAME) - - logger.info('Extracting deepdream model...') - zipfile.ZipFile(DeepDream._ZIP_FILENAME, 'r').extractall('.') - - @staticmethod - def _tffunc(*argtypes): - '''Helper that transforms TF-graph generating function into a regular one. - See `_resize` function below. - ''' - placeholders = list(map(tf.placeholder, argtypes)) - def wrap(f): - out = f(*placeholders) - def wrapper(*args, **kw): - return out.eval(dict(zip(placeholders, args)), session=kw.get('session')) - return wrapper - return wrap - - @staticmethod - def _base_resize(img, size): - '''Helper function that uses TF to resize an image''' - img = tf.expand_dims(img, 0) - return tf.image.resize_bilinear(img, size)[0,:,:,:] - - def __init__(self): - if not os.path.exists(DeepDream._MODEL_FILENAME): - self._download_model() - - self._graph = tf.Graph() - self._session = tf.InteractiveSession(graph=self._graph) - self._resize = self._tffunc(np.float32, np.int32)(self._base_resize) - with tf.gfile.FastGFile(DeepDream._MODEL_FILENAME, 'rb') as f: - graph_def = tf.GraphDef() - graph_def.ParseFromString(f.read()) - self._t_input = tf.placeholder(np.float32, name='input') # define the input tensor - imagenet_mean = 117.0 - t_preprocessed = tf.expand_dims(self._t_input-imagenet_mean, 0) - tf.import_graph_def(graph_def, {'input':t_preprocessed}) - - self.t_obj = self.T('mixed4d_3x3_bottleneck_pre_relu')[:,:,:,139] - #self.t_obj = tf.square(self.T('mixed4c')) - - def T(self, layer_name): - '''Helper for getting layer output tensor''' - return self._graph.get_tensor_by_name('import/%s:0'%layer_name) - - def _calc_grad_tiled(self, img, t_grad, tile_size=512): - '''Compute the value of tensor t_grad over the image in a tiled way. - Random shifts are applied to the image to blur tile boundaries over - multiple iterations.''' - sz = tile_size - h, w = img.shape[:2] - sx, sy = np.random.randint(sz, size=2) - img_shift = np.roll(np.roll(img, sx, 1), sy, 0) - grad = np.zeros_like(img) - for y in range(0, max(h-sz//2, sz),sz): - for x in range(0, max(w-sz//2, sz),sz): - sub = img_shift[y:y+sz,x:x+sz] - g = self._session.run(t_grad, {self._t_input:sub}) - grad[y:y+sz,x:x+sz] = g - return np.roll(np.roll(grad, -sx, 1), -sy, 0) - - def process_frame(self, frame, iter_n=10, step=1.5, octave_n=4, octave_scale=1.4): - t_score = tf.reduce_mean(self.t_obj) # defining the optimization objective - t_grad = tf.gradients(t_score, self._t_input)[0] # behold the power of automatic differentiation! - - # split the image into a number of octaves - img = frame - octaves = [] - for i in range(octave_n-1): - hw = img.shape[:2] - lo = self._resize(img, np.int32(np.float32(hw)/octave_scale)) - hi = img-self._resize(lo, hw) - img = lo - octaves.append(hi) - - # generate details octave by octave - for octave in range(octave_n): - if octave>0: - hi = octaves[-octave] - img = self._resize(img, hi.shape[:2])+hi - for i in range(iter_n): - g = self._calc_grad_tiled(img, t_grad) - img += g*(step / (np.abs(g).mean()+1e-7)) - #print('.',end = ' ') - return img + '''DeepDream implementation, adapted from official tensorflow deepdream tutorial: + https://github.com/tensorflow/tensorflow/tree/master/tensorflow/examples/tutorials/deepdream + + Credit: Alexander Mordvintsev + ''' + + _DOWNLOAD_URL = 'https://storage.googleapis.com/download.tensorflow.org/models/inception5h.zip' + _ZIP_FILENAME = 'deepdream_model.zip' + _MODEL_FILENAME = 'tensorflow_inception_graph.pb' + + @staticmethod + def _download_model(): + logger.info('Downloading deepdream model...') + try: + from urllib.request import urlretrieve # python 3 + except ImportError: + from urllib import urlretrieve # python 2 + urlretrieve(DeepDream._DOWNLOAD_URL, DeepDream._ZIP_FILENAME) + + logger.info('Extracting deepdream model...') + zipfile.ZipFile(DeepDream._ZIP_FILENAME, 'r').extractall('.') + + @staticmethod + def _tffunc(*argtypes): + '''Helper that transforms TF-graph generating function into a regular one. + See `_resize` function below. + ''' + placeholders = list(map(tf.placeholder, argtypes)) + + def wrap(f): + out = f(*placeholders) + + def wrapper(*args, **kw): + return out.eval(dict(zip(placeholders, args)), session=kw.get('session')) + return wrapper + return wrap + + @staticmethod + def _base_resize(img, size): + '''Helper function that uses TF to resize an image''' + img = tf.expand_dims(img, 0) + return tf.image.resize_bilinear(img, size)[0, :, :, :] + + def __init__(self): + if not os.path.exists(DeepDream._MODEL_FILENAME): + self._download_model() + + self._graph = tf.Graph() + self._session = tf.InteractiveSession(graph=self._graph) + self._resize = self._tffunc(np.float32, np.int32)(self._base_resize) + with tf.gfile.FastGFile(DeepDream._MODEL_FILENAME, 'rb') as f: + graph_def = tf.GraphDef() + graph_def.ParseFromString(f.read()) + self._t_input = tf.placeholder( + np.float32, name='input') # define the input tensor + imagenet_mean = 117.0 + t_preprocessed = tf.expand_dims(self._t_input-imagenet_mean, 0) + tf.import_graph_def(graph_def, {'input': t_preprocessed}) + + self.t_obj = self.T('mixed4d_3x3_bottleneck_pre_relu')[:, :, :, 139] + #self.t_obj = tf.square(self.T('mixed4c')) + + def T(self, layer_name): + '''Helper for getting layer output tensor''' + return self._graph.get_tensor_by_name('import/%s:0' % layer_name) + + def _calc_grad_tiled(self, img, t_grad, tile_size=512): + '''Compute the value of tensor t_grad over the image in a tiled way. + Random shifts are applied to the image to blur tile boundaries over + multiple iterations.''' + sz = tile_size + h, w = img.shape[:2] + sx, sy = np.random.randint(sz, size=2) + img_shift = np.roll(np.roll(img, sx, 1), sy, 0) + grad = np.zeros_like(img) + for y in range(0, max(h-sz//2, sz), sz): + for x in range(0, max(w-sz//2, sz), sz): + sub = img_shift[y:y+sz, x:x+sz] + g = self._session.run(t_grad, {self._t_input: sub}) + grad[y:y+sz, x:x+sz] = g + return np.roll(np.roll(grad, -sx, 1), -sy, 0) + + def process_frame(self, frame, iter_n=10, step=1.5, octave_n=4, octave_scale=1.4): + # defining the optimization objective + t_score = tf.reduce_mean(self.t_obj) + # behold the power of automatic differentiation! + t_grad = tf.gradients(t_score, self._t_input)[0] + + # split the image into a number of octaves + img = frame + octaves = [] + for i in range(octave_n-1): + hw = img.shape[:2] + lo = self._resize(img, np.int32(np.float32(hw)/octave_scale)) + hi = img-self._resize(lo, hw) + img = lo + octaves.append(hi) + + # generate details octave by octave + for octave in range(octave_n): + if octave > 0: + hi = octaves[-octave] + img = self._resize(img, hi.shape[:2])+hi + for i in range(iter_n): + g = self._calc_grad_tiled(img, t_grad) + img += g*(step / (np.abs(g).mean()+1e-7)) + #print('.',end = ' ') + return img if __name__ == '__main__': - args = parser.parse_args() - if args.dream: - import tensorflow as tf - process_frame = DeepDream().process_frame - else: - process_frame = process_frame_simple - run(args.in_filename, args.out_filename, process_frame) + args = parser.parse_args() + if args.dream: + import tensorflow as tf + process_frame = DeepDream().process_frame + else: + process_frame = process_frame_simple + run(args.in_filename, args.out_filename, process_frame) diff --git a/examples/transcribe.py b/examples/transcribe.py index 0b7200c4..91cdae4a 100755 --- a/examples/transcribe.py +++ b/examples/transcribe.py @@ -14,18 +14,19 @@ logger.setLevel(logging.INFO) -parser = argparse.ArgumentParser(description='Convert speech audio to text using Google Speech API') +parser = argparse.ArgumentParser( + description='Convert speech audio to text using Google Speech API') parser.add_argument('in_filename', help='Input filename (`-` for stdin)') def decode_audio(in_filename, **input_kwargs): try: out, err = (ffmpeg - .input(in_filename, **input_kwargs) - .output('-', format='s16le', acodec='pcm_s16le', ac=1, ar='16k') - .overwrite_output() - .run(capture_stdout=True, capture_stderr=True) - ) + .input(in_filename, **input_kwargs) + .output('-', format='s16le', acodec='pcm_s16le', ac=1, ar='16k') + .overwrite_output() + .run(capture_stdout=True, capture_stderr=True) + ) except ffmpeg.Error as e: print(e.stderr, file=sys.stderr) sys.exit(1) diff --git a/examples/video_info.py b/examples/video_info.py index df9c992e..62a766a4 100755 --- a/examples/video_info.py +++ b/examples/video_info.py @@ -10,22 +10,23 @@ if __name__ == '__main__': - args = parser.parse_args() + args = parser.parse_args() - try: - probe = ffmpeg.probe(args.in_filename) - except ffmpeg.Error as e: - print(e.stderr, file=sys.stderr) - sys.exit(1) + try: + probe = ffmpeg.probe(args.in_filename) + except ffmpeg.Error as e: + print(e.stderr, file=sys.stderr) + sys.exit(1) - video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None) - if video_stream is None: - print('No video stream found', file=sys.stderr) - sys.exit(1) + video_stream = next( + (stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None) + if video_stream is None: + print('No video stream found', file=sys.stderr) + sys.exit(1) - width = int(video_stream['width']) - height = int(video_stream['height']) - num_frames = int(video_stream['nb_frames']) - print('width: {}'.format(width)) - print('height: {}'.format(height)) - print('num_frames: {}'.format(num_frames)) + width = int(video_stream['width']) + height = int(video_stream['height']) + num_frames = int(video_stream['nb_frames']) + print('width: {}'.format(width)) + print('height: {}'.format(height)) + print('num_frames: {}'.format(num_frames)) diff --git a/ffmpeg/__init__.py b/ffmpeg/__init__.py index a88d344d..70a12812 100644 --- a/ffmpeg/__init__.py +++ b/ffmpeg/__init__.py @@ -13,10 +13,10 @@ from ._view import * __all__ = ( - nodes.__all__ - + _ffmpeg.__all__ - + _probe.__all__ - + _run.__all__ - + _view.__all__ - + _filters.__all__ + nodes.__all__ + + _ffmpeg.__all__ + + _probe.__all__ + + _run.__all__ + + _view.__all__ + + _filters.__all__ ) diff --git a/ffmpeg/_ffmpeg.py b/ffmpeg/_ffmpeg.py index 31e2b906..a98f1694 100644 --- a/ffmpeg/_ffmpeg.py +++ b/ffmpeg/_ffmpeg.py @@ -4,94 +4,94 @@ from ._utils import basestring from .nodes import ( - filter_operator, - GlobalNode, - InputNode, - MergeOutputsNode, - OutputNode, - output_operator, + filter_operator, + GlobalNode, + InputNode, + MergeOutputsNode, + OutputNode, + output_operator, ) def input(filename, **kwargs): - """Input file URL (ffmpeg ``-i`` option) + """Input file URL (ffmpeg ``-i`` option) - Any supplied kwargs are passed to ffmpeg verbatim (e.g. ``t=20``, - ``f='mp4'``, ``acodec='pcm'``, etc.). + Any supplied kwargs are passed to ffmpeg verbatim (e.g. ``t=20``, + ``f='mp4'``, ``acodec='pcm'``, etc.). - To tell ffmpeg to read from stdin, use ``pipe:`` as the filename. + To tell ffmpeg to read from stdin, use ``pipe:`` as the filename. - Official documentation: `Main options `__ - """ - kwargs['filename'] = filename - fmt = kwargs.pop('f', None) - if fmt: - if 'format' in kwargs: - raise ValueError("Can't specify both `format` and `f` kwargs") - kwargs['format'] = fmt - return InputNode(input.__name__, kwargs=kwargs).stream() + Official documentation: `Main options `__ + """ + kwargs['filename'] = filename + fmt = kwargs.pop('f', None) + if fmt: + if 'format' in kwargs: + raise ValueError("Can't specify both `format` and `f` kwargs") + kwargs['format'] = fmt + return InputNode(input.__name__, kwargs=kwargs).stream() @output_operator() def global_args(stream, *args): - """Add extra global command-line argument(s), e.g. ``-progress``. - """ - return GlobalNode(stream, global_args.__name__, args).stream() + """Add extra global command-line argument(s), e.g. ``-progress``. + """ + return GlobalNode(stream, global_args.__name__, args).stream() @output_operator() def overwrite_output(stream): - """Overwrite output files without asking (ffmpeg ``-y`` option) + """Overwrite output files without asking (ffmpeg ``-y`` option) - Official documentation: `Main options `__ - """ - return GlobalNode(stream, overwrite_output.__name__, ['-y']).stream() + Official documentation: `Main options `__ + """ + return GlobalNode(stream, overwrite_output.__name__, ['-y']).stream() @output_operator() def merge_outputs(*streams): - """Include all given outputs in one ffmpeg command line - """ - return MergeOutputsNode(streams, merge_outputs.__name__).stream() + """Include all given outputs in one ffmpeg command line + """ + return MergeOutputsNode(streams, merge_outputs.__name__).stream() @filter_operator() def output(*streams_and_filename, **kwargs): - """Output file URL - - Syntax: - `ffmpeg.output(stream1[, stream2, stream3...], filename, **ffmpeg_args)` - - Any supplied keyword arguments are passed to ffmpeg verbatim (e.g. - ``t=20``, ``f='mp4'``, ``acodec='pcm'``, ``vcodec='rawvideo'``, - etc.). Some keyword-arguments are handled specially, as shown below. - - Args: - video_bitrate: parameter for ``-b:v``, e.g. ``video_bitrate=1000``. - audio_bitrate: parameter for ``-b:a``, e.g. ``audio_bitrate=200``. - format: alias for ``-f`` parameter, e.g. ``format='mp4'`` - (equivalent to ``f='mp4'``). - - If multiple streams are provided, they are mapped to the same - output. - - To tell ffmpeg to write to stdout, use ``pipe:`` as the filename. - - Official documentation: `Synopsis `__ - """ - streams_and_filename = list(streams_and_filename) - if 'filename' not in kwargs: - if not isinstance(streams_and_filename[-1], basestring): - raise ValueError('A filename must be provided') - kwargs['filename'] = streams_and_filename.pop(-1) - streams = streams_and_filename - - fmt = kwargs.pop('f', None) - if fmt: - if 'format' in kwargs: - raise ValueError("Can't specify both `format` and `f` kwargs") - kwargs['format'] = fmt - return OutputNode(streams, output.__name__, kwargs=kwargs).stream() + """Output file URL + + Syntax: + `ffmpeg.output(stream1[, stream2, stream3...], filename, **ffmpeg_args)` + + Any supplied keyword arguments are passed to ffmpeg verbatim (e.g. + ``t=20``, ``f='mp4'``, ``acodec='pcm'``, ``vcodec='rawvideo'``, + etc.). Some keyword-arguments are handled specially, as shown below. + + Args: + video_bitrate: parameter for ``-b:v``, e.g. ``video_bitrate=1000``. + audio_bitrate: parameter for ``-b:a``, e.g. ``audio_bitrate=200``. + format: alias for ``-f`` parameter, e.g. ``format='mp4'`` + (equivalent to ``f='mp4'``). + + If multiple streams are provided, they are mapped to the same + output. + + To tell ffmpeg to write to stdout, use ``pipe:`` as the filename. + + Official documentation: `Synopsis `__ + """ + streams_and_filename = list(streams_and_filename) + if 'filename' not in kwargs: + if not isinstance(streams_and_filename[-1], basestring): + raise ValueError('A filename must be provided') + kwargs['filename'] = streams_and_filename.pop(-1) + streams = streams_and_filename + + fmt = kwargs.pop('f', None) + if fmt: + if 'format' in kwargs: + raise ValueError("Can't specify both `format` and `f` kwargs") + kwargs['format'] = fmt + return OutputNode(streams, output.__name__, kwargs=kwargs).stream() __all__ = ['input', 'merge_outputs', 'output', 'overwrite_output'] diff --git a/ffmpeg/_filters.py b/ffmpeg/_filters.py index 2691220a..fa0bf4b8 100644 --- a/ffmpeg/_filters.py +++ b/ffmpeg/_filters.py @@ -214,7 +214,8 @@ def drawbox(stream, x, y, width, height, color, thickness=None, **kwargs): if thickness: kwargs['t'] = thickness return FilterNode( - stream, drawbox.__name__, args=[x, y, width, height, color], kwargs=kwargs + stream, drawbox.__name__, args=[ + x, y, width, height, color], kwargs=kwargs ).stream() @@ -392,7 +393,8 @@ def concat(*streams, **kwargs): if len(streams) % stream_count != 0: raise ValueError( 'Expected concat input streams to have length multiple of {} (v={}, a={}); got {}'.format( - stream_count, video_stream_count, audio_stream_count, len(streams) + stream_count, video_stream_count, audio_stream_count, len( + streams) ) ) kwargs['n'] = int(len(streams) / stream_count) diff --git a/ffmpeg/_probe.py b/ffmpeg/_probe.py index 090d7abf..68c235af 100644 --- a/ffmpeg/_probe.py +++ b/ffmpeg/_probe.py @@ -5,26 +5,26 @@ def probe(filename, cmd='ffprobe', timeout=None, **kwargs): - """Run ffprobe on the specified file and return a JSON representation of the output. + """Run ffprobe on the specified file and return a JSON representation of the output. - Raises: - :class:`ffmpeg.Error`: if ffprobe returns a non-zero exit code, - an :class:`Error` is returned with a generic error message. - The stderr output can be retrieved by accessing the - ``stderr`` property of the exception. - """ - args = [cmd, '-show_format', '-show_streams', '-of', 'json'] - args += convert_kwargs_to_cmd_line_args(kwargs) - args += [filename] + Raises: + :class:`ffmpeg.Error`: if ffprobe returns a non-zero exit code, + an :class:`Error` is returned with a generic error message. + The stderr output can be retrieved by accessing the + ``stderr`` property of the exception. + """ + args = [cmd, '-show_format', '-show_streams', '-of', 'json'] + args += convert_kwargs_to_cmd_line_args(kwargs) + args += [filename] - p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - communicate_kwargs = {} - if timeout is not None: - communicate_kwargs['timeout'] = timeout - out, err = p.communicate(**communicate_kwargs) - if p.returncode != 0: - raise Error('ffprobe', out, err) - return json.loads(out.decode('utf-8')) + p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + communicate_kwargs = {} + if timeout is not None: + communicate_kwargs['timeout'] = timeout + out, err = p.communicate(**communicate_kwargs) + if p.returncode != 0: + raise Error('ffprobe', out, err) + return json.loads(out.decode('utf-8')) __all__ = ['probe'] diff --git a/ffmpeg/_run.py b/ffmpeg/_run.py index c9cbb7ce..e2b1ccf4 100644 --- a/ffmpeg/_run.py +++ b/ffmpeg/_run.py @@ -10,320 +10,332 @@ from ._ffmpeg import input, output from .nodes import ( - get_stream_spec_nodes, - FilterNode, - GlobalNode, - InputNode, - OutputNode, - output_operator, + get_stream_spec_nodes, + FilterNode, + GlobalNode, + InputNode, + OutputNode, + output_operator, ) class Error(Exception): - def __init__(self, cmd, stdout, stderr): - super(Error, self).__init__( - '{} error (see stderr output for detail)'.format(cmd) - ) - self.stdout = stdout - self.stderr = stderr + def __init__(self, cmd, stdout, stderr): + super(Error, self).__init__( + '{} error (see stderr output for detail)'.format(cmd) + ) + self.stdout = stdout + self.stderr = stderr def _get_input_args(input_node): - if input_node.name == input.__name__: - kwargs = copy.copy(input_node.kwargs) - filename = kwargs.pop('filename') - fmt = kwargs.pop('format', None) - video_size = kwargs.pop('video_size', None) - args = [] - if fmt: - args += ['-f', fmt] - if video_size: - args += ['-video_size', '{}x{}'.format(video_size[0], video_size[1])] - args += convert_kwargs_to_cmd_line_args(kwargs) - args += ['-i', filename] - else: - raise ValueError('Unsupported input node: {}'.format(input_node)) - return args + if input_node.name == input.__name__: + kwargs = copy.copy(input_node.kwargs) + filename = kwargs.pop('filename') + fmt = kwargs.pop('format', None) + video_size = kwargs.pop('video_size', None) + args = [] + if fmt: + args += ['-f', fmt] + if video_size: + args += ['-video_size', + '{}x{}'.format(video_size[0], video_size[1])] + args += convert_kwargs_to_cmd_line_args(kwargs) + args += ['-i', filename] + else: + raise ValueError('Unsupported input node: {}'.format(input_node)) + return args def _format_input_stream_name(stream_name_map, edge, is_final_arg=False): - prefix = stream_name_map[edge.upstream_node, edge.upstream_label] - if not edge.upstream_selector: - suffix = '' - else: - suffix = ':{}'.format(edge.upstream_selector) - if is_final_arg and isinstance(edge.upstream_node, InputNode): - ## Special case: `-map` args should not have brackets for input - ## nodes. - fmt = '{}{}' - else: - fmt = '[{}{}]' - return fmt.format(prefix, suffix) + prefix = stream_name_map[edge.upstream_node, edge.upstream_label] + if not edge.upstream_selector: + suffix = '' + else: + suffix = ':{}'.format(edge.upstream_selector) + if is_final_arg and isinstance(edge.upstream_node, InputNode): + # Special case: `-map` args should not have brackets for input + # nodes. + fmt = '{}{}' + else: + fmt = '[{}{}]' + return fmt.format(prefix, suffix) def _format_output_stream_name(stream_name_map, edge): - return '[{}]'.format(stream_name_map[edge.upstream_node, edge.upstream_label]) + return '[{}]'.format(stream_name_map[edge.upstream_node, edge.upstream_label]) def _get_filter_spec(node, outgoing_edge_map, stream_name_map): - incoming_edges = node.incoming_edges - outgoing_edges = get_outgoing_edges(node, outgoing_edge_map) - inputs = [ - _format_input_stream_name(stream_name_map, edge) for edge in incoming_edges - ] - outputs = [ - _format_output_stream_name(stream_name_map, edge) for edge in outgoing_edges - ] - filter_spec = '{}{}{}'.format( - ''.join(inputs), node._get_filter(outgoing_edges), ''.join(outputs) - ) - return filter_spec + incoming_edges = node.incoming_edges + outgoing_edges = get_outgoing_edges(node, outgoing_edge_map) + inputs = [ + _format_input_stream_name(stream_name_map, edge) for edge in incoming_edges + ] + outputs = [ + _format_output_stream_name(stream_name_map, edge) for edge in outgoing_edges + ] + filter_spec = '{}{}{}'.format( + ''.join(inputs), node._get_filter(outgoing_edges), ''.join(outputs) + ) + return filter_spec def _allocate_filter_stream_names(filter_nodes, outgoing_edge_maps, stream_name_map): - stream_count = 0 - for upstream_node in filter_nodes: - outgoing_edge_map = outgoing_edge_maps[upstream_node] - for upstream_label, downstreams in sorted(outgoing_edge_map.items()): - if len(downstreams) > 1: - # TODO: automatically insert `splits` ahead of time via graph transformation. - raise ValueError( - 'Encountered {} with multiple outgoing edges with same upstream label {!r}; a ' - '`split` filter is probably required'.format( - upstream_node, upstream_label - ) - ) - stream_name_map[upstream_node, upstream_label] = 's{}'.format(stream_count) - stream_count += 1 + stream_count = 0 + for upstream_node in filter_nodes: + outgoing_edge_map = outgoing_edge_maps[upstream_node] + for upstream_label, downstreams in sorted(outgoing_edge_map.items()): + if len(downstreams) > 1: + # TODO: automatically insert `splits` ahead of time via graph transformation. + raise ValueError( + 'Encountered {} with multiple outgoing edges with same upstream label {!r}; a ' + '`split` filter is probably required'.format( + upstream_node, upstream_label + ) + ) + stream_name_map[upstream_node, + upstream_label] = 's{}'.format(stream_count) + stream_count += 1 def _get_filter_arg(filter_nodes, outgoing_edge_maps, stream_name_map): - _allocate_filter_stream_names(filter_nodes, outgoing_edge_maps, stream_name_map) - filter_specs = [ - _get_filter_spec(node, outgoing_edge_maps[node], stream_name_map) - for node in filter_nodes - ] - return ';'.join(filter_specs) + _allocate_filter_stream_names( + filter_nodes, outgoing_edge_maps, stream_name_map) + filter_specs = [ + _get_filter_spec(node, outgoing_edge_maps[node], stream_name_map) + for node in filter_nodes + ] + return ';'.join(filter_specs) def _get_global_args(node): - return list(node.args) + return list(node.args) def _get_output_args(node, stream_name_map): - if node.name != output.__name__: - raise ValueError('Unsupported output node: {}'.format(node)) - args = [] - - if len(node.incoming_edges) == 0: - raise ValueError('Output node {} has no mapped streams'.format(node)) - - for edge in node.incoming_edges: - # edge = node.incoming_edges[0] - stream_name = _format_input_stream_name( - stream_name_map, edge, is_final_arg=True - ) - if stream_name != '0' or len(node.incoming_edges) > 1: - args += ['-map', stream_name] - - kwargs = copy.copy(node.kwargs) - filename = kwargs.pop('filename') - if 'format' in kwargs: - args += ['-f', kwargs.pop('format')] - if 'video_bitrate' in kwargs: - args += ['-b:v', str(kwargs.pop('video_bitrate'))] - if 'audio_bitrate' in kwargs: - args += ['-b:a', str(kwargs.pop('audio_bitrate'))] - if 'video_size' in kwargs: - video_size = kwargs.pop('video_size') - if not isinstance(video_size, basestring) and isinstance( - video_size, collections.Iterable - ): - video_size = '{}x{}'.format(video_size[0], video_size[1]) - args += ['-video_size', video_size] - args += convert_kwargs_to_cmd_line_args(kwargs) - args += [filename] - return args + if node.name != output.__name__: + raise ValueError('Unsupported output node: {}'.format(node)) + args = [] + + if len(node.incoming_edges) == 0: + raise ValueError('Output node {} has no mapped streams'.format(node)) + + for edge in node.incoming_edges: + # edge = node.incoming_edges[0] + stream_name = _format_input_stream_name( + stream_name_map, edge, is_final_arg=True + ) + if stream_name != '0' or len(node.incoming_edges) > 1: + args += ['-map', stream_name] + + kwargs = copy.copy(node.kwargs) + filename = kwargs.pop('filename') + if 'format' in kwargs: + args += ['-f', kwargs.pop('format')] + if 'video_bitrate' in kwargs: + args += ['-b:v', str(kwargs.pop('video_bitrate'))] + if 'audio_bitrate' in kwargs: + args += ['-b:a', str(kwargs.pop('audio_bitrate'))] + if 'video_size' in kwargs: + video_size = kwargs.pop('video_size') + if not isinstance(video_size, basestring) and isinstance( + video_size, collections.Iterable + ): + video_size = '{}x{}'.format(video_size[0], video_size[1]) + args += ['-video_size', video_size] + args += convert_kwargs_to_cmd_line_args(kwargs) + args += [filename] + return args @output_operator() def get_args(stream_spec, overwrite_output=False): - """Build command-line arguments to be passed to ffmpeg.""" - nodes = get_stream_spec_nodes(stream_spec) - args = [] - # TODO: group nodes together, e.g. `-i somefile -r somerate`. - sorted_nodes, outgoing_edge_maps = topo_sort(nodes) - input_nodes = [node for node in sorted_nodes if isinstance(node, InputNode)] - output_nodes = [node for node in sorted_nodes if isinstance(node, OutputNode)] - global_nodes = [node for node in sorted_nodes if isinstance(node, GlobalNode)] - filter_nodes = [node for node in sorted_nodes if isinstance(node, FilterNode)] - stream_name_map = {(node, None): str(i) for i, node in enumerate(input_nodes)} - filter_arg = _get_filter_arg(filter_nodes, outgoing_edge_maps, stream_name_map) - args += reduce(operator.add, [_get_input_args(node) for node in input_nodes]) - if filter_arg: - args += ['-filter_complex', filter_arg] - args += reduce( - operator.add, [_get_output_args(node, stream_name_map) for node in output_nodes] - ) - args += reduce(operator.add, [_get_global_args(node) for node in global_nodes], []) - if overwrite_output: - args += ['-y'] - return args + """Build command-line arguments to be passed to ffmpeg.""" + nodes = get_stream_spec_nodes(stream_spec) + args = [] + # TODO: group nodes together, e.g. `-i somefile -r somerate`. + sorted_nodes, outgoing_edge_maps = topo_sort(nodes) + input_nodes = [ + node for node in sorted_nodes if isinstance(node, InputNode)] + output_nodes = [ + node for node in sorted_nodes if isinstance(node, OutputNode)] + global_nodes = [ + node for node in sorted_nodes if isinstance(node, GlobalNode)] + filter_nodes = [ + node for node in sorted_nodes if isinstance(node, FilterNode)] + stream_name_map = {(node, None): str(i) + for i, node in enumerate(input_nodes)} + filter_arg = _get_filter_arg( + filter_nodes, outgoing_edge_maps, stream_name_map) + args += reduce(operator.add, [_get_input_args(node) + for node in input_nodes]) + if filter_arg: + args += ['-filter_complex', filter_arg] + args += reduce( + operator.add, [_get_output_args(node, stream_name_map) + for node in output_nodes] + ) + args += reduce(operator.add, [_get_global_args(node) + for node in global_nodes], []) + if overwrite_output: + args += ['-y'] + return args @output_operator() def compile(stream_spec, cmd='ffmpeg', overwrite_output=False): - """Build command-line for invoking ffmpeg. + """Build command-line for invoking ffmpeg. - The :meth:`run` function uses this to build the command line - arguments and should work in most cases, but calling this function - directly is useful for debugging or if you need to invoke ffmpeg - manually for whatever reason. + The :meth:`run` function uses this to build the command line + arguments and should work in most cases, but calling this function + directly is useful for debugging or if you need to invoke ffmpeg + manually for whatever reason. - This is the same as calling :meth:`get_args` except that it also - includes the ``ffmpeg`` command as the first argument. - """ - if isinstance(cmd, basestring): - cmd = [cmd] - elif type(cmd) != list: - cmd = list(cmd) - return cmd + get_args(stream_spec, overwrite_output=overwrite_output) + This is the same as calling :meth:`get_args` except that it also + includes the ``ffmpeg`` command as the first argument. + """ + if isinstance(cmd, basestring): + cmd = [cmd] + elif type(cmd) != list: + cmd = list(cmd) + return cmd + get_args(stream_spec, overwrite_output=overwrite_output) @output_operator() def run_async( - stream_spec, - cmd='ffmpeg', - pipe_stdin=False, - pipe_stdout=False, - pipe_stderr=False, - quiet=False, - overwrite_output=False, + stream_spec, + cmd='ffmpeg', + pipe_stdin=False, + pipe_stdout=False, + pipe_stderr=False, + quiet=False, + overwrite_output=False, ): - """Asynchronously invoke ffmpeg for the supplied node graph. - - Args: - pipe_stdin: if True, connect pipe to subprocess stdin (to be - used with ``pipe:`` ffmpeg inputs). - pipe_stdout: if True, connect pipe to subprocess stdout (to be - used with ``pipe:`` ffmpeg outputs). - pipe_stderr: if True, connect pipe to subprocess stderr. - quiet: shorthand for setting ``capture_stdout`` and - ``capture_stderr``. - **kwargs: keyword-arguments passed to ``get_args()`` (e.g. - ``overwrite_output=True``). - - Returns: - A `subprocess Popen`_ object representing the child process. - - Examples: - Run and stream input:: - - process = ( - ffmpeg - .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height)) - .output(out_filename, pix_fmt='yuv420p') - .overwrite_output() - .run_async(pipe_stdin=True) - ) - process.communicate(input=input_data) - - Run and capture output:: - - process = ( - ffmpeg - .input(in_filename) - .output('pipe':, format='rawvideo', pix_fmt='rgb24') - .run_async(pipe_stdout=True, pipe_stderr=True) - ) - out, err = process.communicate() - - Process video frame-by-frame using numpy:: - - process1 = ( - ffmpeg - .input(in_filename) - .output('pipe:', format='rawvideo', pix_fmt='rgb24') - .run_async(pipe_stdout=True) - ) - - process2 = ( - ffmpeg - .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height)) - .output(out_filename, pix_fmt='yuv420p') - .overwrite_output() - .run_async(pipe_stdin=True) - ) - - while True: - in_bytes = process1.stdout.read(width * height * 3) - if not in_bytes: - break - in_frame = ( - np - .frombuffer(in_bytes, np.uint8) - .reshape([height, width, 3]) - ) - out_frame = in_frame * 0.3 - process2.stdin.write( - frame - .astype(np.uint8) - .tobytes() - ) - - process2.stdin.close() - process1.wait() - process2.wait() - - .. _subprocess Popen: https://docs.python.org/3/library/subprocess.html#popen-objects - """ - args = compile(stream_spec, cmd, overwrite_output=overwrite_output) - stdin_stream = subprocess.PIPE if pipe_stdin else None - stdout_stream = subprocess.PIPE if pipe_stdout or quiet else None - stderr_stream = subprocess.PIPE if pipe_stderr or quiet else None - return subprocess.Popen( - args, stdin=stdin_stream, stdout=stdout_stream, stderr=stderr_stream - ) + """Asynchronously invoke ffmpeg for the supplied node graph. + + Args: + pipe_stdin: if True, connect pipe to subprocess stdin (to be + used with ``pipe:`` ffmpeg inputs). + pipe_stdout: if True, connect pipe to subprocess stdout (to be + used with ``pipe:`` ffmpeg outputs). + pipe_stderr: if True, connect pipe to subprocess stderr. + quiet: shorthand for setting ``capture_stdout`` and + ``capture_stderr``. + **kwargs: keyword-arguments passed to ``get_args()`` (e.g. + ``overwrite_output=True``). + + Returns: + A `subprocess Popen`_ object representing the child process. + + Examples: + Run and stream input:: + + process = ( + ffmpeg + .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height)) + .output(out_filename, pix_fmt='yuv420p') + .overwrite_output() + .run_async(pipe_stdin=True) + ) + process.communicate(input=input_data) + + Run and capture output:: + + process = ( + ffmpeg + .input(in_filename) + .output('pipe':, format='rawvideo', pix_fmt='rgb24') + .run_async(pipe_stdout=True, pipe_stderr=True) + ) + out, err = process.communicate() + + Process video frame-by-frame using numpy:: + + process1 = ( + ffmpeg + .input(in_filename) + .output('pipe:', format='rawvideo', pix_fmt='rgb24') + .run_async(pipe_stdout=True) + ) + + process2 = ( + ffmpeg + .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height)) + .output(out_filename, pix_fmt='yuv420p') + .overwrite_output() + .run_async(pipe_stdin=True) + ) + + while True: + in_bytes = process1.stdout.read(width * height * 3) + if not in_bytes: + break + in_frame = ( + np + .frombuffer(in_bytes, np.uint8) + .reshape([height, width, 3]) + ) + out_frame = in_frame * 0.3 + process2.stdin.write( + frame + .astype(np.uint8) + .tobytes() + ) + + process2.stdin.close() + process1.wait() + process2.wait() + + .. _subprocess Popen: https://docs.python.org/3/library/subprocess.html#popen-objects + """ + args = compile(stream_spec, cmd, overwrite_output=overwrite_output) + stdin_stream = subprocess.PIPE if pipe_stdin else None + stdout_stream = subprocess.PIPE if pipe_stdout or quiet else None + stderr_stream = subprocess.PIPE if pipe_stderr or quiet else None + return subprocess.Popen( + args, stdin=stdin_stream, stdout=stdout_stream, stderr=stderr_stream + ) @output_operator() def run( - stream_spec, - cmd='ffmpeg', - capture_stdout=False, - capture_stderr=False, - input=None, - quiet=False, - overwrite_output=False, + stream_spec, + cmd='ffmpeg', + capture_stdout=False, + capture_stderr=False, + input=None, + quiet=False, + overwrite_output=False, ): - """Invoke ffmpeg for the supplied node graph. - - Args: - capture_stdout: if True, capture stdout (to be used with - ``pipe:`` ffmpeg outputs). - capture_stderr: if True, capture stderr. - quiet: shorthand for setting ``capture_stdout`` and ``capture_stderr``. - input: text to be sent to stdin (to be used with ``pipe:`` - ffmpeg inputs) - **kwargs: keyword-arguments passed to ``get_args()`` (e.g. - ``overwrite_output=True``). - - Returns: (out, err) tuple containing captured stdout and stderr data. - """ - process = run_async( - stream_spec, - cmd, - pipe_stdin=input is not None, - pipe_stdout=capture_stdout, - pipe_stderr=capture_stderr, - quiet=quiet, - overwrite_output=overwrite_output, - ) - out, err = process.communicate(input) - retcode = process.poll() - if retcode: - raise Error('ffmpeg', out, err) - return out, err + """Invoke ffmpeg for the supplied node graph. + + Args: + capture_stdout: if True, capture stdout (to be used with + ``pipe:`` ffmpeg outputs). + capture_stderr: if True, capture stderr. + quiet: shorthand for setting ``capture_stdout`` and ``capture_stderr``. + input: text to be sent to stdin (to be used with ``pipe:`` + ffmpeg inputs) + **kwargs: keyword-arguments passed to ``get_args()`` (e.g. + ``overwrite_output=True``). + + Returns: (out, err) tuple containing captured stdout and stderr data. + """ + process = run_async( + stream_spec, + cmd, + pipe_stdin=input is not None, + pipe_stdout=capture_stdout, + pipe_stderr=capture_stderr, + quiet=quiet, + overwrite_output=overwrite_output, + ) + out, err = process.communicate(input) + retcode = process.poll() + if retcode: + raise Error('ffmpeg', out, err) + return out, err __all__ = ['compile', 'Error', 'get_args', 'run', 'run_async'] diff --git a/ffmpeg/_utils.py b/ffmpeg/_utils.py index 92d76110..3efd3b74 100644 --- a/ffmpeg/_utils.py +++ b/ffmpeg/_utils.py @@ -7,99 +7,99 @@ if sys.version_info.major == 2: - # noinspection PyUnresolvedReferences,PyShadowingBuiltins - str = str + # noinspection PyUnresolvedReferences,PyShadowingBuiltins + str = str # `past.builtins.basestring` module can't be imported on Python3 in some environments (Ubuntu). # This code is copy-pasted from it to avoid crashes. class BaseBaseString(type): - def __instancecheck__(cls, instance): - return isinstance(instance, (bytes, str)) + def __instancecheck__(cls, instance): + return isinstance(instance, (bytes, str)) - def __subclasshook__(cls, thing): - # TODO: What should go here? - raise NotImplemented + def __subclasshook__(cls, thing): + # TODO: What should go here? + raise NotImplemented def with_metaclass(meta, *bases): - class metaclass(meta): - __call__ = type.__call__ - __init__ = type.__init__ + class metaclass(meta): + __call__ = type.__call__ + __init__ = type.__init__ - def __new__(cls, name, this_bases, d): - if this_bases is None: - return type.__new__(cls, name, (), d) - return meta(name, bases, d) + def __new__(cls, name, this_bases, d): + if this_bases is None: + return type.__new__(cls, name, (), d) + return meta(name, bases, d) - return metaclass('temporary_class', None, {}) + return metaclass('temporary_class', None, {}) if sys.version_info.major >= 3: - class basestring(with_metaclass(BaseBaseString)): - pass + class basestring(with_metaclass(BaseBaseString)): + pass else: - # noinspection PyUnresolvedReferences,PyCompatibility - from builtins import basestring + # noinspection PyUnresolvedReferences,PyCompatibility + from builtins import basestring def _recursive_repr(item): - """Hack around python `repr` to deterministically represent dictionaries. - - This is able to represent more things than json.dumps, since it does not require things to be JSON serializable - (e.g. datetimes). - """ - if isinstance(item, basestring): - result = str(item) - elif isinstance(item, list): - result = '[{}]'.format(', '.join([_recursive_repr(x) for x in item])) - elif isinstance(item, dict): - kv_pairs = [ - '{}: {}'.format(_recursive_repr(k), _recursive_repr(item[k])) - for k in sorted(item) - ] - result = '{' + ', '.join(kv_pairs) + '}' - else: - result = repr(item) - return result + """Hack around python `repr` to deterministically represent dictionaries. + + This is able to represent more things than json.dumps, since it does not require things to be JSON serializable + (e.g. datetimes). + """ + if isinstance(item, basestring): + result = str(item) + elif isinstance(item, list): + result = '[{}]'.format(', '.join([_recursive_repr(x) for x in item])) + elif isinstance(item, dict): + kv_pairs = [ + '{}: {}'.format(_recursive_repr(k), _recursive_repr(item[k])) + for k in sorted(item) + ] + result = '{' + ', '.join(kv_pairs) + '}' + else: + result = repr(item) + return result def get_hash(item): - repr_ = _recursive_repr(item).encode('utf-8') - return hashlib.md5(repr_).hexdigest() + repr_ = _recursive_repr(item).encode('utf-8') + return hashlib.md5(repr_).hexdigest() def get_hash_int(item): - return int(get_hash(item), base=16) + return int(get_hash(item), base=16) def escape_chars(text, chars): - """Helper function to escape uncomfortable characters.""" - text = str(text) - chars = list(set(chars)) - if '\\' in chars: - chars.remove('\\') - chars.insert(0, '\\') - for ch in chars: - text = text.replace(ch, '\\' + ch) - return text + """Helper function to escape uncomfortable characters.""" + text = str(text) + chars = list(set(chars)) + if '\\' in chars: + chars.remove('\\') + chars.insert(0, '\\') + for ch in chars: + text = text.replace(ch, '\\' + ch) + return text def convert_kwargs_to_cmd_line_args(kwargs): - """Helper function to build command line arguments out of dict.""" - args = [] - for k in sorted(kwargs.keys()): - v = kwargs[k] - if isinstance(v, collections.Iterable) and not isinstance(v, str): - for value in v: - args.append('-{}'.format(k)) - if value is not None: - args.append('{}'.format(value)) - continue - args.append('-{}'.format(k)) - if v is not None: - args.append('{}'.format(v)) - return args + """Helper function to build command line arguments out of dict.""" + args = [] + for k in sorted(kwargs.keys()): + v = kwargs[k] + if isinstance(v, collections.Iterable) and not isinstance(v, str): + for value in v: + args.append('-{}'.format(k)) + if value is not None: + args.append('{}'.format(value)) + continue + args.append('-{}'.format(k)) + if v is not None: + args.append('{}'.format(v)) + return args diff --git a/ffmpeg/_view.py b/ffmpeg/_view.py index fb129fa8..2e3455de 100644 --- a/ffmpeg/_view.py +++ b/ffmpeg/_view.py @@ -6,11 +6,11 @@ import tempfile from ffmpeg.nodes import ( - FilterNode, - get_stream_spec_nodes, - InputNode, - OutputNode, - stream_operator, + FilterNode, + get_stream_spec_nodes, + InputNode, + OutputNode, + stream_operator, ) @@ -18,91 +18,91 @@ def _get_node_color(node): - if isinstance(node, InputNode): - color = '#99cc00' - elif isinstance(node, OutputNode): - color = '#99ccff' - elif isinstance(node, FilterNode): - color = '#ffcc00' - else: - color = None - return color + if isinstance(node, InputNode): + color = '#99cc00' + elif isinstance(node, OutputNode): + color = '#99ccff' + elif isinstance(node, FilterNode): + color = '#ffcc00' + else: + color = None + return color @stream_operator() def view(stream_spec, detail=False, filename=None, pipe=False, **kwargs): - try: - import graphviz - except ImportError: - raise ImportError( - 'failed to import graphviz; please make sure graphviz is installed (e.g. `pip install ' - 'graphviz`)' - ) - - show_labels = kwargs.pop('show_labels', True) - if pipe and filename is not None: - raise ValueError('Can\'t specify both `filename` and `pipe`') - elif not pipe and filename is None: - filename = tempfile.mktemp() - - nodes = get_stream_spec_nodes(stream_spec) - - sorted_nodes, outgoing_edge_maps = topo_sort(nodes) - graph = graphviz.Digraph(format='png') - graph.attr(rankdir='LR') - if len(list(kwargs.keys())) != 0: - raise ValueError( - 'Invalid kwargs key(s): {}'.format(', '.join(list(kwargs.keys()))) - ) - - for node in sorted_nodes: - color = _get_node_color(node) - - if detail: - lines = [node.short_repr] - lines += ['{!r}'.format(arg) for arg in node.args] - lines += [ - '{}={!r}'.format(key, node.kwargs[key]) for key in sorted(node.kwargs) - ] - node_text = '\n'.join(lines) - else: - node_text = node.short_repr - graph.node( - str(hash(node)), node_text, shape='box', style='filled', fillcolor=color - ) - outgoing_edge_map = outgoing_edge_maps.get(node, {}) - - for edge in get_outgoing_edges(node, outgoing_edge_map): - kwargs = {} - up_label = edge.upstream_label - down_label = edge.downstream_label - up_selector = edge.upstream_selector - - if show_labels and ( - up_label is not None - or down_label is not None - or up_selector is not None - ): - if up_label is None: - up_label = '' - if up_selector is not None: - up_label += ":" + up_selector - if down_label is None: - down_label = '' - if up_label != '' and down_label != '': - middle = ' {} '.format(_RIGHT_ARROW) - else: - middle = '' - kwargs['label'] = '{} {} {}'.format(up_label, middle, down_label) - upstream_node_id = str(hash(edge.upstream_node)) - downstream_node_id = str(hash(edge.downstream_node)) - graph.edge(upstream_node_id, downstream_node_id, **kwargs) - - if pipe: - return graph.pipe() - else: - graph.view(filename, cleanup=True) - return stream_spec + try: + import graphviz + except ImportError: + raise ImportError( + 'failed to import graphviz; please make sure graphviz is installed (e.g. `pip install ' + 'graphviz`)' + ) + + show_labels = kwargs.pop('show_labels', True) + if pipe and filename is not None: + raise ValueError('Can\'t specify both `filename` and `pipe`') + elif not pipe and filename is None: + filename = tempfile.mktemp() + + nodes = get_stream_spec_nodes(stream_spec) + + sorted_nodes, outgoing_edge_maps = topo_sort(nodes) + graph = graphviz.Digraph(format='png') + graph.attr(rankdir='LR') + if len(list(kwargs.keys())) != 0: + raise ValueError( + 'Invalid kwargs key(s): {}'.format(', '.join(list(kwargs.keys()))) + ) + + for node in sorted_nodes: + color = _get_node_color(node) + + if detail: + lines = [node.short_repr] + lines += ['{!r}'.format(arg) for arg in node.args] + lines += [ + '{}={!r}'.format(key, node.kwargs[key]) for key in sorted(node.kwargs) + ] + node_text = '\n'.join(lines) + else: + node_text = node.short_repr + graph.node( + str(hash(node)), node_text, shape='box', style='filled', fillcolor=color + ) + outgoing_edge_map = outgoing_edge_maps.get(node, {}) + + for edge in get_outgoing_edges(node, outgoing_edge_map): + kwargs = {} + up_label = edge.upstream_label + down_label = edge.downstream_label + up_selector = edge.upstream_selector + + if show_labels and ( + up_label is not None + or down_label is not None + or up_selector is not None + ): + if up_label is None: + up_label = '' + if up_selector is not None: + up_label += ":" + up_selector + if down_label is None: + down_label = '' + if up_label != '' and down_label != '': + middle = ' {} '.format(_RIGHT_ARROW) + else: + middle = '' + kwargs['label'] = '{} {} {}'.format(up_label, middle, down_label) + upstream_node_id = str(hash(edge.upstream_node)) + downstream_node_id = str(hash(edge.downstream_node)) + graph.edge(upstream_node_id, downstream_node_id, **kwargs) + + if pipe: + return graph.pipe() + else: + graph.view(filename, cleanup=True) + return stream_spec __all__ = ['view'] diff --git a/ffmpeg/dag.py b/ffmpeg/dag.py index 9564d7f8..2a25e608 100644 --- a/ffmpeg/dag.py +++ b/ffmpeg/dag.py @@ -6,226 +6,226 @@ class DagNode(object): - """Node in a directed-acyclic graph (DAG). + """Node in a directed-acyclic graph (DAG). - Edges: - DagNodes are connected by edges. An edge connects two nodes with a label for each side: - - ``upstream_node``: upstream/parent node - - ``upstream_label``: label on the outgoing side of the upstream node - - ``downstream_node``: downstream/child node - - ``downstream_label``: label on the incoming side of the downstream node + Edges: + DagNodes are connected by edges. An edge connects two nodes with a label for each side: + - ``upstream_node``: upstream/parent node + - ``upstream_label``: label on the outgoing side of the upstream node + - ``downstream_node``: downstream/child node + - ``downstream_label``: label on the incoming side of the downstream node - For example, DagNode A may be connected to DagNode B with an edge labelled "foo" on A's side, and "bar" on B's - side: + For example, DagNode A may be connected to DagNode B with an edge labelled "foo" on A's side, and "bar" on B's + side: - _____ _____ - | | | | - | A >[foo]---[bar]> B | - |_____| |_____| + _____ _____ + | | | | + | A >[foo]---[bar]> B | + |_____| |_____| - Edge labels may be integers or strings, and nodes cannot have more than one incoming edge with the same label. + Edge labels may be integers or strings, and nodes cannot have more than one incoming edge with the same label. - DagNodes may have any number of incoming edges and any number of outgoing edges. DagNodes keep track only of - their incoming edges, but the entire graph structure can be inferred by looking at the furthest downstream - nodes and working backwards. + DagNodes may have any number of incoming edges and any number of outgoing edges. DagNodes keep track only of + their incoming edges, but the entire graph structure can be inferred by looking at the furthest downstream + nodes and working backwards. - Hashing: - DagNodes must be hashable, and two nodes are considered to be equivalent if they have the same hash value. + Hashing: + DagNodes must be hashable, and two nodes are considered to be equivalent if they have the same hash value. - Nodes are immutable, and the hash should remain constant as a result. If a node with new contents is required, - create a new node and throw the old one away. + Nodes are immutable, and the hash should remain constant as a result. If a node with new contents is required, + create a new node and throw the old one away. - String representation: - In order for graph visualization tools to show useful information, nodes must be representable as strings. The - ``repr`` operator should provide a more or less "full" representation of the node, and the ``short_repr`` - property should be a shortened, concise representation. + String representation: + In order for graph visualization tools to show useful information, nodes must be representable as strings. The + ``repr`` operator should provide a more or less "full" representation of the node, and the ``short_repr`` + property should be a shortened, concise representation. - Again, because nodes are immutable, the string representations should remain constant. - """ + Again, because nodes are immutable, the string representations should remain constant. + """ - def __hash__(self): - """Return an integer hash of the node.""" - raise NotImplementedError() + def __hash__(self): + """Return an integer hash of the node.""" + raise NotImplementedError() - def __eq__(self, other): - """Compare two nodes; implementations should return True if (and only if) hashes match.""" - raise NotImplementedError() + def __eq__(self, other): + """Compare two nodes; implementations should return True if (and only if) hashes match.""" + raise NotImplementedError() - def __repr__(self, other): - """Return a full string representation of the node.""" - raise NotImplementedError() + def __repr__(self, other): + """Return a full string representation of the node.""" + raise NotImplementedError() - @property - def short_repr(self): - """Return a partial/concise representation of the node.""" - raise NotImplementedError() + @property + def short_repr(self): + """Return a partial/concise representation of the node.""" + raise NotImplementedError() - @property - def incoming_edge_map(self): - """Provides information about all incoming edges that connect to this node. + @property + def incoming_edge_map(self): + """Provides information about all incoming edges that connect to this node. - The edge map is a dictionary that maps an ``incoming_label`` to ``(outgoing_node, outgoing_label)``. Note that - implicity, ``incoming_node`` is ``self``. See "Edges" section above. - """ - raise NotImplementedError() + The edge map is a dictionary that maps an ``incoming_label`` to ``(outgoing_node, outgoing_label)``. Note that + implicity, ``incoming_node`` is ``self``. See "Edges" section above. + """ + raise NotImplementedError() DagEdge = namedtuple( - 'DagEdge', - [ - 'downstream_node', - 'downstream_label', - 'upstream_node', - 'upstream_label', - 'upstream_selector', - ], + 'DagEdge', + [ + 'downstream_node', + 'downstream_label', + 'upstream_node', + 'upstream_label', + 'upstream_selector', + ], ) def get_incoming_edges(downstream_node, incoming_edge_map): - edges = [] - for downstream_label, upstream_info in list(incoming_edge_map.items()): - upstream_node, upstream_label, upstream_selector = upstream_info - edges += [ - DagEdge( - downstream_node, - downstream_label, - upstream_node, - upstream_label, - upstream_selector, - ) - ] - return edges + edges = [] + for downstream_label, upstream_info in list(incoming_edge_map.items()): + upstream_node, upstream_label, upstream_selector = upstream_info + edges += [ + DagEdge( + downstream_node, + downstream_label, + upstream_node, + upstream_label, + upstream_selector, + ) + ] + return edges def get_outgoing_edges(upstream_node, outgoing_edge_map): - edges = [] - for upstream_label, downstream_infos in sorted(outgoing_edge_map.items()): - for downstream_info in downstream_infos: - downstream_node, downstream_label, downstream_selector = downstream_info - edges += [ - DagEdge( - downstream_node, - downstream_label, - upstream_node, - upstream_label, - downstream_selector, - ) - ] - return edges + edges = [] + for upstream_label, downstream_infos in sorted(outgoing_edge_map.items()): + for downstream_info in downstream_infos: + downstream_node, downstream_label, downstream_selector = downstream_info + edges += [ + DagEdge( + downstream_node, + downstream_label, + upstream_node, + upstream_label, + downstream_selector, + ) + ] + return edges class KwargReprNode(DagNode): - """A DagNode that can be represented as a set of args+kwargs. - """ - - @property - def __upstream_hashes(self): - hashes = [] - for downstream_label, upstream_info in list(self.incoming_edge_map.items()): - upstream_node, upstream_label, upstream_selector = upstream_info - hashes += [ - hash(x) - for x in [ - downstream_label, - upstream_node, - upstream_label, - upstream_selector, - ] - ] - return hashes - - @property - def __inner_hash(self): - props = {'args': self.args, 'kwargs': self.kwargs} - return get_hash(props) - - def __get_hash(self): - hashes = self.__upstream_hashes + [self.__inner_hash] - return get_hash_int(hashes) - - def __init__(self, incoming_edge_map, name, args, kwargs): - self.__incoming_edge_map = incoming_edge_map - self.name = name - self.args = args - self.kwargs = kwargs - self.__hash = self.__get_hash() - - def __hash__(self): - return self.__hash - - def __eq__(self, other): - return hash(self) == hash(other) - - @property - def short_hash(self): - return '{:x}'.format(abs(hash(self)))[:12] - - def long_repr(self, include_hash=True): - formatted_props = ['{!r}'.format(arg) for arg in self.args] - formatted_props += [ - '{}={!r}'.format(key, self.kwargs[key]) for key in sorted(self.kwargs) - ] - out = '{}({})'.format(self.name, ', '.join(formatted_props)) - if include_hash: - out += ' <{}>'.format(self.short_hash) - return out - - def __repr__(self): - return self.long_repr() - - @property - def incoming_edges(self): - return get_incoming_edges(self, self.incoming_edge_map) - - @property - def incoming_edge_map(self): - return self.__incoming_edge_map - - @property - def short_repr(self): - return self.name + """A DagNode that can be represented as a set of args+kwargs. + """ + + @property + def __upstream_hashes(self): + hashes = [] + for downstream_label, upstream_info in list(self.incoming_edge_map.items()): + upstream_node, upstream_label, upstream_selector = upstream_info + hashes += [ + hash(x) + for x in [ + downstream_label, + upstream_node, + upstream_label, + upstream_selector, + ] + ] + return hashes + + @property + def __inner_hash(self): + props = {'args': self.args, 'kwargs': self.kwargs} + return get_hash(props) + + def __get_hash(self): + hashes = self.__upstream_hashes + [self.__inner_hash] + return get_hash_int(hashes) + + def __init__(self, incoming_edge_map, name, args, kwargs): + self.__incoming_edge_map = incoming_edge_map + self.name = name + self.args = args + self.kwargs = kwargs + self.__hash = self.__get_hash() + + def __hash__(self): + return self.__hash + + def __eq__(self, other): + return hash(self) == hash(other) + + @property + def short_hash(self): + return '{:x}'.format(abs(hash(self)))[:12] + + def long_repr(self, include_hash=True): + formatted_props = ['{!r}'.format(arg) for arg in self.args] + formatted_props += [ + '{}={!r}'.format(key, self.kwargs[key]) for key in sorted(self.kwargs) + ] + out = '{}({})'.format(self.name, ', '.join(formatted_props)) + if include_hash: + out += ' <{}>'.format(self.short_hash) + return out + + def __repr__(self): + return self.long_repr() + + @property + def incoming_edges(self): + return get_incoming_edges(self, self.incoming_edge_map) + + @property + def incoming_edge_map(self): + return self.__incoming_edge_map + + @property + def short_repr(self): + return self.name def topo_sort(downstream_nodes): - marked_nodes = [] - sorted_nodes = [] - outgoing_edge_maps = {} - - def visit( - upstream_node, - upstream_label, - downstream_node, - downstream_label, - downstream_selector=None, - ): - if upstream_node in marked_nodes: - raise RuntimeError('Graph is not a DAG') - - if downstream_node is not None: - outgoing_edge_map = outgoing_edge_maps.get(upstream_node, {}) - outgoing_edge_infos = outgoing_edge_map.get(upstream_label, []) - outgoing_edge_infos += [ - (downstream_node, downstream_label, downstream_selector) - ] - outgoing_edge_map[upstream_label] = outgoing_edge_infos - outgoing_edge_maps[upstream_node] = outgoing_edge_map - - if upstream_node not in sorted_nodes: - marked_nodes.append(upstream_node) - for edge in upstream_node.incoming_edges: - visit( - edge.upstream_node, - edge.upstream_label, - edge.downstream_node, - edge.downstream_label, - edge.upstream_selector, - ) - marked_nodes.remove(upstream_node) - sorted_nodes.append(upstream_node) - - unmarked_nodes = [(node, None) for node in downstream_nodes] - while unmarked_nodes: - upstream_node, upstream_label = unmarked_nodes.pop() - visit(upstream_node, upstream_label, None, None) - return sorted_nodes, outgoing_edge_maps + marked_nodes = [] + sorted_nodes = [] + outgoing_edge_maps = {} + + def visit( + upstream_node, + upstream_label, + downstream_node, + downstream_label, + downstream_selector=None, + ): + if upstream_node in marked_nodes: + raise RuntimeError('Graph is not a DAG') + + if downstream_node is not None: + outgoing_edge_map = outgoing_edge_maps.get(upstream_node, {}) + outgoing_edge_infos = outgoing_edge_map.get(upstream_label, []) + outgoing_edge_infos += [ + (downstream_node, downstream_label, downstream_selector) + ] + outgoing_edge_map[upstream_label] = outgoing_edge_infos + outgoing_edge_maps[upstream_node] = outgoing_edge_map + + if upstream_node not in sorted_nodes: + marked_nodes.append(upstream_node) + for edge in upstream_node.incoming_edges: + visit( + edge.upstream_node, + edge.upstream_label, + edge.downstream_node, + edge.downstream_label, + edge.upstream_selector, + ) + marked_nodes.remove(upstream_node) + sorted_nodes.append(upstream_node) + + unmarked_nodes = [(node, None) for node in downstream_nodes] + while unmarked_nodes: + upstream_node, upstream_label = unmarked_nodes.pop() + visit(upstream_node, upstream_label, None, None) + return sorted_nodes, outgoing_edge_maps diff --git a/ffmpeg/nodes.py b/ffmpeg/nodes.py index cacab8ee..9c716a26 100644 --- a/ffmpeg/nodes.py +++ b/ffmpeg/nodes.py @@ -8,370 +8,370 @@ def _is_of_types(obj, types): - valid = False - for stream_type in types: - if isinstance(obj, stream_type): - valid = True - break - return valid + valid = False + for stream_type in types: + if isinstance(obj, stream_type): + valid = True + break + return valid def _get_types_str(types): - return ', '.join(['{}.{}'.format(x.__module__, x.__name__) for x in types]) + return ', '.join(['{}.{}'.format(x.__module__, x.__name__) for x in types]) class Stream(object): - """Represents the outgoing edge of an upstream node; may be used to create more downstream nodes.""" - - def __init__( - self, upstream_node, upstream_label, node_types, upstream_selector=None - ): - if not _is_of_types(upstream_node, node_types): - raise TypeError( - 'Expected upstream node to be of one of the following type(s): {}; got {}'.format( - _get_types_str(node_types), type(upstream_node) - ) - ) - self.node = upstream_node - self.label = upstream_label - self.selector = upstream_selector - - def __hash__(self): - return get_hash_int([hash(self.node), hash(self.label)]) - - def __eq__(self, other): - return hash(self) == hash(other) - - def __repr__(self): - node_repr = self.node.long_repr(include_hash=False) - selector = '' - if self.selector: - selector = ':{}'.format(self.selector) - out = '{}[{!r}{}] <{}>'.format( - node_repr, self.label, selector, self.node.short_hash - ) - return out - - def __getitem__(self, index): - """ - Select a component (audio, video) of the stream. - - Example: - Process the audio and video portions of a stream independently:: - - input = ffmpeg.input('in.mp4') - audio = input['a'].filter("aecho", 0.8, 0.9, 1000, 0.3) - video = input['v'].hflip() - out = ffmpeg.output(audio, video, 'out.mp4') - """ - if self.selector is not None: - raise ValueError('Stream already has a selector: {}'.format(self)) - elif not isinstance(index, basestring): - raise TypeError("Expected string index (e.g. 'a'); got {!r}".format(index)) - return self.node.stream(label=self.label, selector=index) - - @property - def audio(self): - """Select the audio-portion of a stream. - - Some ffmpeg filters drop audio streams, and care must be taken - to preserve the audio in the final output. The ``.audio`` and - ``.video`` operators can be used to reference the audio/video - portions of a stream so that they can be processed separately - and then re-combined later in the pipeline. This dilemma is - intrinsic to ffmpeg, and ffmpeg-python tries to stay out of the - way while users may refer to the official ffmpeg documentation - as to why certain filters drop audio. - - ``stream.audio`` is a shorthand for ``stream['a']``. - - Example: - Process the audio and video portions of a stream independently:: - - input = ffmpeg.input('in.mp4') - audio = input.audio.filter("aecho", 0.8, 0.9, 1000, 0.3) - video = input.video.hflip() - out = ffmpeg.output(audio, video, 'out.mp4') - """ - return self['a'] - - @property - def video(self): - """Select the video-portion of a stream. - - Some ffmpeg filters drop audio streams, and care must be taken - to preserve the audio in the final output. The ``.audio`` and - ``.video`` operators can be used to reference the audio/video - portions of a stream so that they can be processed separately - and then re-combined later in the pipeline. This dilemma is - intrinsic to ffmpeg, and ffmpeg-python tries to stay out of the - way while users may refer to the official ffmpeg documentation - as to why certain filters drop audio. - - ``stream.video`` is a shorthand for ``stream['v']``. - - Example: - Process the audio and video portions of a stream independently:: - - input = ffmpeg.input('in.mp4') - audio = input.audio.filter("aecho", 0.8, 0.9, 1000, 0.3) - video = input.video.hflip() - out = ffmpeg.output(audio, video, 'out.mp4') - """ - return self['v'] + """Represents the outgoing edge of an upstream node; may be used to create more downstream nodes.""" + + def __init__( + self, upstream_node, upstream_label, node_types, upstream_selector=None + ): + if not _is_of_types(upstream_node, node_types): + raise TypeError( + 'Expected upstream node to be of one of the following type(s): {}; got {}'.format( + _get_types_str(node_types), type(upstream_node) + ) + ) + self.node = upstream_node + self.label = upstream_label + self.selector = upstream_selector + + def __hash__(self): + return get_hash_int([hash(self.node), hash(self.label)]) + + def __eq__(self, other): + return hash(self) == hash(other) + + def __repr__(self): + node_repr = self.node.long_repr(include_hash=False) + selector = '' + if self.selector: + selector = ':{}'.format(self.selector) + out = '{}[{!r}{}] <{}>'.format( + node_repr, self.label, selector, self.node.short_hash + ) + return out + + def __getitem__(self, index): + """ + Select a component (audio, video) of the stream. + + Example: + Process the audio and video portions of a stream independently:: + + input = ffmpeg.input('in.mp4') + audio = input['a'].filter("aecho", 0.8, 0.9, 1000, 0.3) + video = input['v'].hflip() + out = ffmpeg.output(audio, video, 'out.mp4') + """ + if self.selector is not None: + raise ValueError('Stream already has a selector: {}'.format(self)) + elif not isinstance(index, basestring): + raise TypeError("Expected string index (e.g. 'a'); got {!r}".format(index)) + return self.node.stream(label=self.label, selector=index) + + @property + def audio(self): + """Select the audio-portion of a stream. + + Some ffmpeg filters drop audio streams, and care must be taken + to preserve the audio in the final output. The ``.audio`` and + ``.video`` operators can be used to reference the audio/video + portions of a stream so that they can be processed separately + and then re-combined later in the pipeline. This dilemma is + intrinsic to ffmpeg, and ffmpeg-python tries to stay out of the + way while users may refer to the official ffmpeg documentation + as to why certain filters drop audio. + + ``stream.audio`` is a shorthand for ``stream['a']``. + + Example: + Process the audio and video portions of a stream independently:: + + input = ffmpeg.input('in.mp4') + audio = input.audio.filter("aecho", 0.8, 0.9, 1000, 0.3) + video = input.video.hflip() + out = ffmpeg.output(audio, video, 'out.mp4') + """ + return self['a'] + + @property + def video(self): + """Select the video-portion of a stream. + + Some ffmpeg filters drop audio streams, and care must be taken + to preserve the audio in the final output. The ``.audio`` and + ``.video`` operators can be used to reference the audio/video + portions of a stream so that they can be processed separately + and then re-combined later in the pipeline. This dilemma is + intrinsic to ffmpeg, and ffmpeg-python tries to stay out of the + way while users may refer to the official ffmpeg documentation + as to why certain filters drop audio. + + ``stream.video`` is a shorthand for ``stream['v']``. + + Example: + Process the audio and video portions of a stream independently:: + + input = ffmpeg.input('in.mp4') + audio = input.audio.filter("aecho", 0.8, 0.9, 1000, 0.3) + video = input.video.hflip() + out = ffmpeg.output(audio, video, 'out.mp4') + """ + return self['v'] def get_stream_map(stream_spec): - if stream_spec is None: - stream_map = {} - elif isinstance(stream_spec, Stream): - stream_map = {None: stream_spec} - elif isinstance(stream_spec, (list, tuple)): - stream_map = dict(enumerate(stream_spec)) - elif isinstance(stream_spec, dict): - stream_map = stream_spec - return stream_map + if stream_spec is None: + stream_map = {} + elif isinstance(stream_spec, Stream): + stream_map = {None: stream_spec} + elif isinstance(stream_spec, (list, tuple)): + stream_map = dict(enumerate(stream_spec)) + elif isinstance(stream_spec, dict): + stream_map = stream_spec + return stream_map def get_stream_map_nodes(stream_map): - nodes = [] - for stream in list(stream_map.values()): - if not isinstance(stream, Stream): - raise TypeError('Expected Stream; got {}'.format(type(stream))) - nodes.append(stream.node) - return nodes + nodes = [] + for stream in list(stream_map.values()): + if not isinstance(stream, Stream): + raise TypeError('Expected Stream; got {}'.format(type(stream))) + nodes.append(stream.node) + return nodes def get_stream_spec_nodes(stream_spec): - stream_map = get_stream_map(stream_spec) - return get_stream_map_nodes(stream_map) + stream_map = get_stream_map(stream_spec) + return get_stream_map_nodes(stream_map) class Node(KwargReprNode): - """Node base""" - - @classmethod - def __check_input_len(cls, stream_map, min_inputs, max_inputs): - if min_inputs is not None and len(stream_map) < min_inputs: - raise ValueError( - 'Expected at least {} input stream(s); got {}'.format( - min_inputs, len(stream_map) - ) - ) - elif max_inputs is not None and len(stream_map) > max_inputs: - raise ValueError( - 'Expected at most {} input stream(s); got {}'.format( - max_inputs, len(stream_map) - ) - ) - - @classmethod - def __check_input_types(cls, stream_map, incoming_stream_types): - for stream in list(stream_map.values()): - if not _is_of_types(stream, incoming_stream_types): - raise TypeError( - 'Expected incoming stream(s) to be of one of the following types: {}; got {}'.format( - _get_types_str(incoming_stream_types), type(stream) - ) - ) - - @classmethod - def __get_incoming_edge_map(cls, stream_map): - incoming_edge_map = {} - for downstream_label, upstream in list(stream_map.items()): - incoming_edge_map[downstream_label] = ( - upstream.node, - upstream.label, - upstream.selector, - ) - return incoming_edge_map - - def __init__( - self, - stream_spec, - name, - incoming_stream_types, - outgoing_stream_type, - min_inputs, - max_inputs, - args=[], - kwargs={}, - ): - stream_map = get_stream_map(stream_spec) - self.__check_input_len(stream_map, min_inputs, max_inputs) - self.__check_input_types(stream_map, incoming_stream_types) - incoming_edge_map = self.__get_incoming_edge_map(stream_map) - - super(Node, self).__init__(incoming_edge_map, name, args, kwargs) - self.__outgoing_stream_type = outgoing_stream_type - self.__incoming_stream_types = incoming_stream_types - - def stream(self, label=None, selector=None): - """Create an outgoing stream originating from this node. - - More nodes may be attached onto the outgoing stream. - """ - return self.__outgoing_stream_type(self, label, upstream_selector=selector) - - def __getitem__(self, item): - """Create an outgoing stream originating from this node; syntactic sugar for ``self.stream(label)``. - It can also be used to apply a selector: e.g. ``node[0:'a']`` returns a stream with label 0 and - selector ``'a'``, which is the same as ``node.stream(label=0, selector='a')``. - - Example: - Process the audio and video portions of a stream independently:: - - input = ffmpeg.input('in.mp4') - audio = input[:'a'].filter("aecho", 0.8, 0.9, 1000, 0.3) - video = input[:'v'].hflip() - out = ffmpeg.output(audio, video, 'out.mp4') - """ - if isinstance(item, slice): - return self.stream(label=item.start, selector=item.stop) - else: - return self.stream(label=item) + """Node base""" + + @classmethod + def __check_input_len(cls, stream_map, min_inputs, max_inputs): + if min_inputs is not None and len(stream_map) < min_inputs: + raise ValueError( + 'Expected at least {} input stream(s); got {}'.format( + min_inputs, len(stream_map) + ) + ) + elif max_inputs is not None and len(stream_map) > max_inputs: + raise ValueError( + 'Expected at most {} input stream(s); got {}'.format( + max_inputs, len(stream_map) + ) + ) + + @classmethod + def __check_input_types(cls, stream_map, incoming_stream_types): + for stream in list(stream_map.values()): + if not _is_of_types(stream, incoming_stream_types): + raise TypeError( + 'Expected incoming stream(s) to be of one of the following types: {}; got {}'.format( + _get_types_str(incoming_stream_types), type(stream) + ) + ) + + @classmethod + def __get_incoming_edge_map(cls, stream_map): + incoming_edge_map = {} + for downstream_label, upstream in list(stream_map.items()): + incoming_edge_map[downstream_label] = ( + upstream.node, + upstream.label, + upstream.selector, + ) + return incoming_edge_map + + def __init__( + self, + stream_spec, + name, + incoming_stream_types, + outgoing_stream_type, + min_inputs, + max_inputs, + args=[], + kwargs={}, + ): + stream_map = get_stream_map(stream_spec) + self.__check_input_len(stream_map, min_inputs, max_inputs) + self.__check_input_types(stream_map, incoming_stream_types) + incoming_edge_map = self.__get_incoming_edge_map(stream_map) + + super(Node, self).__init__(incoming_edge_map, name, args, kwargs) + self.__outgoing_stream_type = outgoing_stream_type + self.__incoming_stream_types = incoming_stream_types + + def stream(self, label=None, selector=None): + """Create an outgoing stream originating from this node. + + More nodes may be attached onto the outgoing stream. + """ + return self.__outgoing_stream_type(self, label, upstream_selector=selector) + + def __getitem__(self, item): + """Create an outgoing stream originating from this node; syntactic sugar for ``self.stream(label)``. + It can also be used to apply a selector: e.g. ``node[0:'a']`` returns a stream with label 0 and + selector ``'a'``, which is the same as ``node.stream(label=0, selector='a')``. + + Example: + Process the audio and video portions of a stream independently:: + + input = ffmpeg.input('in.mp4') + audio = input[:'a'].filter("aecho", 0.8, 0.9, 1000, 0.3) + video = input[:'v'].hflip() + out = ffmpeg.output(audio, video, 'out.mp4') + """ + if isinstance(item, slice): + return self.stream(label=item.start, selector=item.stop) + else: + return self.stream(label=item) class FilterableStream(Stream): - def __init__(self, upstream_node, upstream_label, upstream_selector=None): - super(FilterableStream, self).__init__( - upstream_node, upstream_label, {InputNode, FilterNode}, upstream_selector - ) + def __init__(self, upstream_node, upstream_label, upstream_selector=None): + super(FilterableStream, self).__init__( + upstream_node, upstream_label, {InputNode, FilterNode}, upstream_selector + ) # noinspection PyMethodOverriding class InputNode(Node): - """InputNode type""" + """InputNode type""" - def __init__(self, name, args=[], kwargs={}): - super(InputNode, self).__init__( - stream_spec=None, - name=name, - incoming_stream_types={}, - outgoing_stream_type=FilterableStream, - min_inputs=0, - max_inputs=0, - args=args, - kwargs=kwargs, - ) + def __init__(self, name, args=[], kwargs={}): + super(InputNode, self).__init__( + stream_spec=None, + name=name, + incoming_stream_types={}, + outgoing_stream_type=FilterableStream, + min_inputs=0, + max_inputs=0, + args=args, + kwargs=kwargs, + ) - @property - def short_repr(self): - return os.path.basename(self.kwargs['filename']) + @property + def short_repr(self): + return os.path.basename(self.kwargs['filename']) # noinspection PyMethodOverriding class FilterNode(Node): - def __init__(self, stream_spec, name, max_inputs=1, args=[], kwargs={}): - super(FilterNode, self).__init__( - stream_spec=stream_spec, - name=name, - incoming_stream_types={FilterableStream}, - outgoing_stream_type=FilterableStream, - min_inputs=1, - max_inputs=max_inputs, - args=args, - kwargs=kwargs, - ) - - """FilterNode""" - - def _get_filter(self, outgoing_edges): - args = self.args - kwargs = self.kwargs - if self.name in ('split', 'asplit'): - args = [len(outgoing_edges)] - - out_args = [escape_chars(x, '\\\'=:') for x in args] - out_kwargs = {} - for k, v in list(kwargs.items()): - k = escape_chars(k, '\\\'=:') - v = escape_chars(v, '\\\'=:') - out_kwargs[k] = v - - arg_params = [escape_chars(v, '\\\'=:') for v in out_args] - kwarg_params = ['{}={}'.format(k, out_kwargs[k]) for k in sorted(out_kwargs)] - params = arg_params + kwarg_params - - params_text = escape_chars(self.name, '\\\'=:') - - if params: - params_text += '={}'.format(':'.join(params)) - return escape_chars(params_text, '\\\'[],;') + def __init__(self, stream_spec, name, max_inputs=1, args=[], kwargs={}): + super(FilterNode, self).__init__( + stream_spec=stream_spec, + name=name, + incoming_stream_types={FilterableStream}, + outgoing_stream_type=FilterableStream, + min_inputs=1, + max_inputs=max_inputs, + args=args, + kwargs=kwargs, + ) + + """FilterNode""" + + def _get_filter(self, outgoing_edges): + args = self.args + kwargs = self.kwargs + if self.name in ('split', 'asplit'): + args = [len(outgoing_edges)] + + out_args = [escape_chars(x, '\\\'=:') for x in args] + out_kwargs = {} + for k, v in list(kwargs.items()): + k = escape_chars(k, '\\\'=:') + v = escape_chars(v, '\\\'=:') + out_kwargs[k] = v + + arg_params = [escape_chars(v, '\\\'=:') for v in out_args] + kwarg_params = ['{}={}'.format(k, out_kwargs[k]) for k in sorted(out_kwargs)] + params = arg_params + kwarg_params + + params_text = escape_chars(self.name, '\\\'=:') + + if params: + params_text += '={}'.format(':'.join(params)) + return escape_chars(params_text, '\\\'[],;') # noinspection PyMethodOverriding class OutputNode(Node): - def __init__(self, stream, name, args=[], kwargs={}): - super(OutputNode, self).__init__( - stream_spec=stream, - name=name, - incoming_stream_types={FilterableStream}, - outgoing_stream_type=OutputStream, - min_inputs=1, - max_inputs=None, - args=args, - kwargs=kwargs, - ) - - @property - def short_repr(self): - return os.path.basename(self.kwargs['filename']) + def __init__(self, stream, name, args=[], kwargs={}): + super(OutputNode, self).__init__( + stream_spec=stream, + name=name, + incoming_stream_types={FilterableStream}, + outgoing_stream_type=OutputStream, + min_inputs=1, + max_inputs=None, + args=args, + kwargs=kwargs, + ) + + @property + def short_repr(self): + return os.path.basename(self.kwargs['filename']) class OutputStream(Stream): - def __init__(self, upstream_node, upstream_label, upstream_selector=None): - super(OutputStream, self).__init__( - upstream_node, - upstream_label, - {OutputNode, GlobalNode, MergeOutputsNode}, - upstream_selector=upstream_selector, - ) + def __init__(self, upstream_node, upstream_label, upstream_selector=None): + super(OutputStream, self).__init__( + upstream_node, + upstream_label, + {OutputNode, GlobalNode, MergeOutputsNode}, + upstream_selector=upstream_selector, + ) # noinspection PyMethodOverriding class MergeOutputsNode(Node): - def __init__(self, streams, name): - super(MergeOutputsNode, self).__init__( - stream_spec=streams, - name=name, - incoming_stream_types={OutputStream}, - outgoing_stream_type=OutputStream, - min_inputs=1, - max_inputs=None, - ) + def __init__(self, streams, name): + super(MergeOutputsNode, self).__init__( + stream_spec=streams, + name=name, + incoming_stream_types={OutputStream}, + outgoing_stream_type=OutputStream, + min_inputs=1, + max_inputs=None, + ) # noinspection PyMethodOverriding class GlobalNode(Node): - def __init__(self, stream, name, args=[], kwargs={}): - super(GlobalNode, self).__init__( - stream_spec=stream, - name=name, - incoming_stream_types={OutputStream}, - outgoing_stream_type=OutputStream, - min_inputs=1, - max_inputs=1, - args=args, - kwargs=kwargs, - ) + def __init__(self, stream, name, args=[], kwargs={}): + super(GlobalNode, self).__init__( + stream_spec=stream, + name=name, + incoming_stream_types={OutputStream}, + outgoing_stream_type=OutputStream, + min_inputs=1, + max_inputs=1, + args=args, + kwargs=kwargs, + ) def stream_operator(stream_classes={Stream}, name=None): - def decorator(func): - func_name = name or func.__name__ - [setattr(stream_class, func_name, func) for stream_class in stream_classes] - return func + def decorator(func): + func_name = name or func.__name__ + [setattr(stream_class, func_name, func) for stream_class in stream_classes] + return func - return decorator + return decorator def filter_operator(name=None): - return stream_operator(stream_classes={FilterableStream}, name=name) + return stream_operator(stream_classes={FilterableStream}, name=name) def output_operator(name=None): - return stream_operator(stream_classes={OutputStream}, name=name) + return stream_operator(stream_classes={OutputStream}, name=name) __all__ = ['Stream'] diff --git a/ffmpeg/tests/test_ffmpeg.py b/ffmpeg/tests/test_ffmpeg.py index 51ee2587..d3cb03c3 100644 --- a/ffmpeg/tests/test_ffmpeg.py +++ b/ffmpeg/tests/test_ffmpeg.py @@ -12,9 +12,9 @@ try: - import mock # python 2 + import mock # python 2 except ImportError: - from unittest import mock # python 3 + from unittest import mock # python 3 TEST_DIR = os.path.dirname(__file__) @@ -30,397 +30,397 @@ def test_escape_chars(): - assert ffmpeg._utils.escape_chars('a:b', ':') == 'a\:b' - assert ffmpeg._utils.escape_chars('a\\:b', ':\\') == 'a\\\\\\:b' - assert ( - ffmpeg._utils.escape_chars('a:b,c[d]e%{}f\'g\'h\\i', '\\\':,[]%') - == 'a\\:b\\,c\\[d\\]e\\%{}f\\\'g\\\'h\\\\i' - ) - assert ffmpeg._utils.escape_chars(123, ':\\') == '123' + assert ffmpeg._utils.escape_chars('a:b', ':') == 'a\\:b' + assert ffmpeg._utils.escape_chars('a\\:b', ':\\') == 'a\\\\\\:b' + assert ( + ffmpeg._utils.escape_chars('a:b,c[d]e%{}f\'g\'h\\i', '\\\':,[]%') + == 'a\\:b\\,c\\[d\\]e\\%{}f\\\'g\\\'h\\\\i' + ) + assert ffmpeg._utils.escape_chars(123, ':\\') == '123' def test_fluent_equality(): - base1 = ffmpeg.input('dummy1.mp4') - base2 = ffmpeg.input('dummy1.mp4') - base3 = ffmpeg.input('dummy2.mp4') - t1 = base1.trim(start_frame=10, end_frame=20) - t2 = base1.trim(start_frame=10, end_frame=20) - t3 = base1.trim(start_frame=10, end_frame=30) - t4 = base2.trim(start_frame=10, end_frame=20) - t5 = base3.trim(start_frame=10, end_frame=20) - assert t1 == t2 - assert t1 != t3 - assert t1 == t4 - assert t1 != t5 + base1 = ffmpeg.input('dummy1.mp4') + base2 = ffmpeg.input('dummy1.mp4') + base3 = ffmpeg.input('dummy2.mp4') + t1 = base1.trim(start_frame=10, end_frame=20) + t2 = base1.trim(start_frame=10, end_frame=20) + t3 = base1.trim(start_frame=10, end_frame=30) + t4 = base2.trim(start_frame=10, end_frame=20) + t5 = base3.trim(start_frame=10, end_frame=20) + assert t1 == t2 + assert t1 != t3 + assert t1 == t4 + assert t1 != t5 def test_fluent_concat(): - base = ffmpeg.input('dummy.mp4') - trimmed1 = base.trim(start_frame=10, end_frame=20) - trimmed2 = base.trim(start_frame=30, end_frame=40) - trimmed3 = base.trim(start_frame=50, end_frame=60) - concat1 = ffmpeg.concat(trimmed1, trimmed2, trimmed3) - concat2 = ffmpeg.concat(trimmed1, trimmed2, trimmed3) - concat3 = ffmpeg.concat(trimmed1, trimmed3, trimmed2) - assert concat1 == concat2 - assert concat1 != concat3 + base = ffmpeg.input('dummy.mp4') + trimmed1 = base.trim(start_frame=10, end_frame=20) + trimmed2 = base.trim(start_frame=30, end_frame=40) + trimmed3 = base.trim(start_frame=50, end_frame=60) + concat1 = ffmpeg.concat(trimmed1, trimmed2, trimmed3) + concat2 = ffmpeg.concat(trimmed1, trimmed2, trimmed3) + concat3 = ffmpeg.concat(trimmed1, trimmed3, trimmed2) + assert concat1 == concat2 + assert concat1 != concat3 def test_fluent_output(): - ffmpeg.input('dummy.mp4').trim(start_frame=10, end_frame=20).output('dummy2.mp4') + ffmpeg.input('dummy.mp4').trim(start_frame=10, end_frame=20).output('dummy2.mp4') def test_fluent_complex_filter(): - in_file = ffmpeg.input('dummy.mp4') - return ffmpeg.concat( - in_file.trim(start_frame=10, end_frame=20), - in_file.trim(start_frame=30, end_frame=40), - in_file.trim(start_frame=50, end_frame=60), - ).output('dummy2.mp4') + in_file = ffmpeg.input('dummy.mp4') + return ffmpeg.concat( + in_file.trim(start_frame=10, end_frame=20), + in_file.trim(start_frame=30, end_frame=40), + in_file.trim(start_frame=50, end_frame=60), + ).output('dummy2.mp4') def test_node_repr(): - in_file = ffmpeg.input('dummy.mp4') - trim1 = ffmpeg.trim(in_file, start_frame=10, end_frame=20) - trim2 = ffmpeg.trim(in_file, start_frame=30, end_frame=40) - trim3 = ffmpeg.trim(in_file, start_frame=50, end_frame=60) - concatted = ffmpeg.concat(trim1, trim2, trim3) - output = ffmpeg.output(concatted, 'dummy2.mp4') - assert repr(in_file.node) == 'input(filename={!r}) <{}>'.format( - 'dummy.mp4', in_file.node.short_hash - ) - assert repr(trim1.node) == 'trim(end_frame=20, start_frame=10) <{}>'.format( - trim1.node.short_hash - ) - assert repr(trim2.node) == 'trim(end_frame=40, start_frame=30) <{}>'.format( - trim2.node.short_hash - ) - assert repr(trim3.node) == 'trim(end_frame=60, start_frame=50) <{}>'.format( - trim3.node.short_hash - ) - assert repr(concatted.node) == 'concat(n=3) <{}>'.format(concatted.node.short_hash) - assert repr(output.node) == 'output(filename={!r}) <{}>'.format( - 'dummy2.mp4', output.node.short_hash - ) + in_file = ffmpeg.input('dummy.mp4') + trim1 = ffmpeg.trim(in_file, start_frame=10, end_frame=20) + trim2 = ffmpeg.trim(in_file, start_frame=30, end_frame=40) + trim3 = ffmpeg.trim(in_file, start_frame=50, end_frame=60) + concatted = ffmpeg.concat(trim1, trim2, trim3) + output = ffmpeg.output(concatted, 'dummy2.mp4') + assert repr(in_file.node) == 'input(filename={!r}) <{}>'.format( + 'dummy.mp4', in_file.node.short_hash + ) + assert repr(trim1.node) == 'trim(end_frame=20, start_frame=10) <{}>'.format( + trim1.node.short_hash + ) + assert repr(trim2.node) == 'trim(end_frame=40, start_frame=30) <{}>'.format( + trim2.node.short_hash + ) + assert repr(trim3.node) == 'trim(end_frame=60, start_frame=50) <{}>'.format( + trim3.node.short_hash + ) + assert repr(concatted.node) == 'concat(n=3) <{}>'.format(concatted.node.short_hash) + assert repr(output.node) == 'output(filename={!r}) <{}>'.format( + 'dummy2.mp4', output.node.short_hash + ) def test_stream_repr(): - in_file = ffmpeg.input('dummy.mp4') - assert repr(in_file) == 'input(filename={!r})[None] <{}>'.format( - 'dummy.mp4', in_file.node.short_hash - ) - split0 = in_file.filter_multi_output('split')[0] - assert repr(split0) == 'split()[0] <{}>'.format(split0.node.short_hash) - dummy_out = in_file.filter_multi_output('dummy')['out'] - assert repr(dummy_out) == 'dummy()[{!r}] <{}>'.format( - dummy_out.label, dummy_out.node.short_hash - ) + in_file = ffmpeg.input('dummy.mp4') + assert repr(in_file) == 'input(filename={!r})[None] <{}>'.format( + 'dummy.mp4', in_file.node.short_hash + ) + split0 = in_file.filter_multi_output('split')[0] + assert repr(split0) == 'split()[0] <{}>'.format(split0.node.short_hash) + dummy_out = in_file.filter_multi_output('dummy')['out'] + assert repr(dummy_out) == 'dummy()[{!r}] <{}>'.format( + dummy_out.label, dummy_out.node.short_hash + ) def test_repeated_args(): - out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4', streamid=['0:0x101', '1:0x102']) - assert out_file.get_args() == ['-i', 'dummy.mp4', '-streamid', '0:0x101', '-streamid', '1:0x102', 'dummy2.mp4'] + out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4', streamid=['0:0x101', '1:0x102']) + assert out_file.get_args() == ['-i', 'dummy.mp4', '-streamid', '0:0x101', '-streamid', '1:0x102', 'dummy2.mp4'] def test__get_args__simple(): - out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4') - assert out_file.get_args() == ['-i', 'dummy.mp4', 'dummy2.mp4'] + out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4') + assert out_file.get_args() == ['-i', 'dummy.mp4', 'dummy2.mp4'] def test_global_args(): - out_file = ( - ffmpeg.input('dummy.mp4') - .output('dummy2.mp4') - .global_args('-progress', 'someurl') - ) - assert out_file.get_args() == [ - '-i', - 'dummy.mp4', - 'dummy2.mp4', - '-progress', - 'someurl', - ] + out_file = ( + ffmpeg.input('dummy.mp4') + .output('dummy2.mp4') + .global_args('-progress', 'someurl') + ) + assert out_file.get_args() == [ + '-i', + 'dummy.mp4', + 'dummy2.mp4', + '-progress', + 'someurl', + ] def _get_simple_example(): - return ffmpeg.input(TEST_INPUT_FILE1).output(TEST_OUTPUT_FILE1) + return ffmpeg.input(TEST_INPUT_FILE1).output(TEST_OUTPUT_FILE1) def _get_complex_filter_example(): - split = ffmpeg.input(TEST_INPUT_FILE1).vflip().split() - split0 = split[0] - split1 = split[1] - - overlay_file = ffmpeg.input(TEST_OVERLAY_FILE) - overlay_file = ffmpeg.crop(overlay_file, 10, 10, 158, 112) - return ( - ffmpeg.concat( - split0.trim(start_frame=10, end_frame=20), - split1.trim(start_frame=30, end_frame=40), - ) - .overlay(overlay_file.hflip()) - .drawbox(50, 50, 120, 120, color='red', thickness=5) - .output(TEST_OUTPUT_FILE1) - .overwrite_output() - ) + split = ffmpeg.input(TEST_INPUT_FILE1).vflip().split() + split0 = split[0] + split1 = split[1] + + overlay_file = ffmpeg.input(TEST_OVERLAY_FILE) + overlay_file = ffmpeg.crop(overlay_file, 10, 10, 158, 112) + return ( + ffmpeg.concat( + split0.trim(start_frame=10, end_frame=20), + split1.trim(start_frame=30, end_frame=40), + ) + .overlay(overlay_file.hflip()) + .drawbox(50, 50, 120, 120, color='red', thickness=5) + .output(TEST_OUTPUT_FILE1) + .overwrite_output() + ) def test__get_args__complex_filter(): - out = _get_complex_filter_example() - args = ffmpeg.get_args(out) - assert args == [ - '-i', - TEST_INPUT_FILE1, - '-i', - TEST_OVERLAY_FILE, - '-filter_complex', - '[0]vflip[s0];' - '[s0]split=2[s1][s2];' - '[s1]trim=end_frame=20:start_frame=10[s3];' - '[s2]trim=end_frame=40:start_frame=30[s4];' - '[s3][s4]concat=n=2[s5];' - '[1]crop=158:112:10:10[s6];' - '[s6]hflip[s7];' - '[s5][s7]overlay=eof_action=repeat[s8];' - '[s8]drawbox=50:50:120:120:red:t=5[s9]', - '-map', - '[s9]', - TEST_OUTPUT_FILE1, - '-y', - ] + out = _get_complex_filter_example() + args = ffmpeg.get_args(out) + assert args == [ + '-i', + TEST_INPUT_FILE1, + '-i', + TEST_OVERLAY_FILE, + '-filter_complex', + '[0]vflip[s0];' + '[s0]split=2[s1][s2];' + '[s1]trim=end_frame=20:start_frame=10[s3];' + '[s2]trim=end_frame=40:start_frame=30[s4];' + '[s3][s4]concat=n=2[s5];' + '[1]crop=158:112:10:10[s6];' + '[s6]hflip[s7];' + '[s5][s7]overlay=eof_action=repeat[s8];' + '[s8]drawbox=50:50:120:120:red:t=5[s9]', + '-map', + '[s9]', + TEST_OUTPUT_FILE1, + '-y', + ] def test_combined_output(): - i1 = ffmpeg.input(TEST_INPUT_FILE1) - i2 = ffmpeg.input(TEST_OVERLAY_FILE) - out = ffmpeg.output(i1, i2, TEST_OUTPUT_FILE1) - assert out.get_args() == [ - '-i', - TEST_INPUT_FILE1, - '-i', - TEST_OVERLAY_FILE, - '-map', - '0', - '-map', - '1', - TEST_OUTPUT_FILE1, - ] + i1 = ffmpeg.input(TEST_INPUT_FILE1) + i2 = ffmpeg.input(TEST_OVERLAY_FILE) + out = ffmpeg.output(i1, i2, TEST_OUTPUT_FILE1) + assert out.get_args() == [ + '-i', + TEST_INPUT_FILE1, + '-i', + TEST_OVERLAY_FILE, + '-map', + '0', + '-map', + '1', + TEST_OUTPUT_FILE1, + ] @pytest.mark.parametrize('use_shorthand', [True, False]) def test_filter_with_selector(use_shorthand): - i = ffmpeg.input(TEST_INPUT_FILE1) - if use_shorthand: - v1 = i.video.hflip() - a1 = i.audio.filter('aecho', 0.8, 0.9, 1000, 0.3) - else: - v1 = i['v'].hflip() - a1 = i['a'].filter('aecho', 0.8, 0.9, 1000, 0.3) - out = ffmpeg.output(a1, v1, TEST_OUTPUT_FILE1) - assert out.get_args() == [ - '-i', - TEST_INPUT_FILE1, - '-filter_complex', - '[0:a]aecho=0.8:0.9:1000:0.3[s0];' '[0:v]hflip[s1]', - '-map', - '[s0]', - '-map', - '[s1]', - TEST_OUTPUT_FILE1, - ] + i = ffmpeg.input(TEST_INPUT_FILE1) + if use_shorthand: + v1 = i.video.hflip() + a1 = i.audio.filter('aecho', 0.8, 0.9, 1000, 0.3) + else: + v1 = i['v'].hflip() + a1 = i['a'].filter('aecho', 0.8, 0.9, 1000, 0.3) + out = ffmpeg.output(a1, v1, TEST_OUTPUT_FILE1) + assert out.get_args() == [ + '-i', + TEST_INPUT_FILE1, + '-filter_complex', + '[0:a]aecho=0.8:0.9:1000:0.3[s0];' '[0:v]hflip[s1]', + '-map', + '[s0]', + '-map', + '[s1]', + TEST_OUTPUT_FILE1, + ] def test_get_item_with_bad_selectors(): - input = ffmpeg.input(TEST_INPUT_FILE1) + input = ffmpeg.input(TEST_INPUT_FILE1) - with pytest.raises(ValueError) as excinfo: - input['a']['a'] - assert str(excinfo.value).startswith('Stream already has a selector:') + with pytest.raises(ValueError) as excinfo: + input['a']['a'] + assert str(excinfo.value).startswith('Stream already has a selector:') - with pytest.raises(TypeError) as excinfo: - input[:'a'] - assert str(excinfo.value).startswith("Expected string index (e.g. 'a')") + with pytest.raises(TypeError) as excinfo: + input[:'a'] + assert str(excinfo.value).startswith("Expected string index (e.g. 'a')") - with pytest.raises(TypeError) as excinfo: - input[5] - assert str(excinfo.value).startswith("Expected string index (e.g. 'a')") + with pytest.raises(TypeError) as excinfo: + input[5] + assert str(excinfo.value).startswith("Expected string index (e.g. 'a')") def _get_complex_filter_asplit_example(): - split = ffmpeg.input(TEST_INPUT_FILE1).vflip().asplit() - split0 = split[0] - split1 = split[1] + split = ffmpeg.input(TEST_INPUT_FILE1).vflip().asplit() + split0 = split[0] + split1 = split[1] - return ( - ffmpeg.concat( - split0.filter('atrim', start=10, end=20), - split1.filter('atrim', start=30, end=40), - ) - .output(TEST_OUTPUT_FILE1) - .overwrite_output() - ) + return ( + ffmpeg.concat( + split0.filter('atrim', start=10, end=20), + split1.filter('atrim', start=30, end=40), + ) + .output(TEST_OUTPUT_FILE1) + .overwrite_output() + ) def test_filter_concat__video_only(): - in1 = ffmpeg.input('in1.mp4') - in2 = ffmpeg.input('in2.mp4') - args = ffmpeg.concat(in1, in2).output('out.mp4').get_args() - assert args == [ - '-i', - 'in1.mp4', - '-i', - 'in2.mp4', - '-filter_complex', - '[0][1]concat=n=2[s0]', - '-map', - '[s0]', - 'out.mp4', - ] + in1 = ffmpeg.input('in1.mp4') + in2 = ffmpeg.input('in2.mp4') + args = ffmpeg.concat(in1, in2).output('out.mp4').get_args() + assert args == [ + '-i', + 'in1.mp4', + '-i', + 'in2.mp4', + '-filter_complex', + '[0][1]concat=n=2[s0]', + '-map', + '[s0]', + 'out.mp4', + ] def test_filter_concat__audio_only(): - in1 = ffmpeg.input('in1.mp4') - in2 = ffmpeg.input('in2.mp4') - args = ffmpeg.concat(in1, in2, v=0, a=1).output('out.mp4').get_args() - assert args == [ - '-i', - 'in1.mp4', - '-i', - 'in2.mp4', - '-filter_complex', - '[0][1]concat=a=1:n=2:v=0[s0]', - '-map', - '[s0]', - 'out.mp4', - ] + in1 = ffmpeg.input('in1.mp4') + in2 = ffmpeg.input('in2.mp4') + args = ffmpeg.concat(in1, in2, v=0, a=1).output('out.mp4').get_args() + assert args == [ + '-i', + 'in1.mp4', + '-i', + 'in2.mp4', + '-filter_complex', + '[0][1]concat=a=1:n=2:v=0[s0]', + '-map', + '[s0]', + 'out.mp4', + ] def test_filter_concat__audio_video(): - in1 = ffmpeg.input('in1.mp4') - in2 = ffmpeg.input('in2.mp4') - joined = ffmpeg.concat(in1.video, in1.audio, in2.hflip(), in2['a'], v=1, a=1).node - args = ffmpeg.output(joined[0], joined[1], 'out.mp4').get_args() - assert args == [ - '-i', - 'in1.mp4', - '-i', - 'in2.mp4', - '-filter_complex', - '[1]hflip[s0];[0:v][0:a][s0][1:a]concat=a=1:n=2:v=1[s1][s2]', - '-map', - '[s1]', - '-map', - '[s2]', - 'out.mp4', - ] + in1 = ffmpeg.input('in1.mp4') + in2 = ffmpeg.input('in2.mp4') + joined = ffmpeg.concat(in1.video, in1.audio, in2.hflip(), in2['a'], v=1, a=1).node + args = ffmpeg.output(joined[0], joined[1], 'out.mp4').get_args() + assert args == [ + '-i', + 'in1.mp4', + '-i', + 'in2.mp4', + '-filter_complex', + '[1]hflip[s0];[0:v][0:a][s0][1:a]concat=a=1:n=2:v=1[s1][s2]', + '-map', + '[s1]', + '-map', + '[s2]', + 'out.mp4', + ] def test_filter_concat__wrong_stream_count(): - in1 = ffmpeg.input('in1.mp4') - in2 = ffmpeg.input('in2.mp4') - with pytest.raises(ValueError) as excinfo: - ffmpeg.concat(in1.video, in1.audio, in2.hflip(), v=1, a=1).node - assert ( - str(excinfo.value) - == 'Expected concat input streams to have length multiple of 2 (v=1, a=1); got 3' - ) + in1 = ffmpeg.input('in1.mp4') + in2 = ffmpeg.input('in2.mp4') + with pytest.raises(ValueError) as excinfo: + ffmpeg.concat(in1.video, in1.audio, in2.hflip(), v=1, a=1).node + assert ( + str(excinfo.value) + == 'Expected concat input streams to have length multiple of 2 (v=1, a=1); got 3' + ) def test_filter_asplit(): - out = _get_complex_filter_asplit_example() - args = out.get_args() - assert args == [ - '-i', - TEST_INPUT_FILE1, - '-filter_complex', - '[0]vflip[s0];[s0]asplit=2[s1][s2];[s1]atrim=end=20:start=10[s3];[s2]atrim=end=40:start=30[s4];[s3]' - '[s4]concat=n=2[s5]', - '-map', - '[s5]', - TEST_OUTPUT_FILE1, - '-y', - ] + out = _get_complex_filter_asplit_example() + args = out.get_args() + assert args == [ + '-i', + TEST_INPUT_FILE1, + '-filter_complex', + '[0]vflip[s0];[s0]asplit=2[s1][s2];[s1]atrim=end=20:start=10[s3];[s2]atrim=end=40:start=30[s4];[s3]' + '[s4]concat=n=2[s5]', + '-map', + '[s5]', + TEST_OUTPUT_FILE1, + '-y', + ] def test__output__bitrate(): - args = ( - ffmpeg.input('in') - .output('out', video_bitrate=1000, audio_bitrate=200) - .get_args() - ) - assert args == ['-i', 'in', '-b:v', '1000', '-b:a', '200', 'out'] + args = ( + ffmpeg.input('in') + .output('out', video_bitrate=1000, audio_bitrate=200) + .get_args() + ) + assert args == ['-i', 'in', '-b:v', '1000', '-b:a', '200', 'out'] @pytest.mark.parametrize('video_size', [(320, 240), '320x240']) def test__output__video_size(video_size): - args = ffmpeg.input('in').output('out', video_size=video_size).get_args() - assert args == ['-i', 'in', '-video_size', '320x240', 'out'] + args = ffmpeg.input('in').output('out', video_size=video_size).get_args() + assert args == ['-i', 'in', '-video_size', '320x240', 'out'] def test_filter_normal_arg_escape(): - """Test string escaping of normal filter args (e.g. ``font`` param of ``drawtext`` filter).""" - - def _get_drawtext_font_repr(font): - """Build a command-line arg using drawtext ``font`` param and extract the ``-filter_complex`` arg.""" - args = ( - ffmpeg.input('in') - .drawtext('test', font='a{}b'.format(font)) - .output('out') - .get_args() - ) - assert args[:3] == ['-i', 'in', '-filter_complex'] - assert args[4:] == ['-map', '[s0]', 'out'] - match = re.match( - r'\[0\]drawtext=font=a((.|\n)*)b:text=test\[s0\]', args[3], re.MULTILINE - ) - assert match is not None, 'Invalid -filter_complex arg: {!r}'.format(args[3]) - return match.group(1) - - expected_backslash_counts = { - 'x': 0, - '\'': 3, - '\\': 3, - '%': 0, - ':': 2, - ',': 1, - '[': 1, - ']': 1, - '=': 2, - '\n': 0, - } - for ch, expected_backslash_count in list(expected_backslash_counts.items()): - expected = '{}{}'.format('\\' * expected_backslash_count, ch) - actual = _get_drawtext_font_repr(ch) - assert expected == actual + """Test string escaping of normal filter args (e.g. ``font`` param of ``drawtext`` filter).""" + + def _get_drawtext_font_repr(font): + """Build a command-line arg using drawtext ``font`` param and extract the ``-filter_complex`` arg.""" + args = ( + ffmpeg.input('in') + .drawtext('test', font='a{}b'.format(font)) + .output('out') + .get_args() + ) + assert args[:3] == ['-i', 'in', '-filter_complex'] + assert args[4:] == ['-map', '[s0]', 'out'] + match = re.match( + r'\[0\]drawtext=font=a((.|\n)*)b:text=test\[s0\]', args[3], re.MULTILINE + ) + assert match is not None, 'Invalid -filter_complex arg: {!r}'.format(args[3]) + return match.group(1) + + expected_backslash_counts = { + 'x': 0, + '\'': 3, + '\\': 3, + '%': 0, + ':': 2, + ',': 1, + '[': 1, + ']': 1, + '=': 2, + '\n': 0, + } + for ch, expected_backslash_count in list(expected_backslash_counts.items()): + expected = '{}{}'.format('\\' * expected_backslash_count, ch) + actual = _get_drawtext_font_repr(ch) + assert expected == actual def test_filter_text_arg_str_escape(): - """Test string escaping of normal filter args (e.g. ``text`` param of ``drawtext`` filter).""" - - def _get_drawtext_text_repr(text): - """Build a command-line arg using drawtext ``text`` param and extract the ``-filter_complex`` arg.""" - args = ffmpeg.input('in').drawtext('a{}b'.format(text)).output('out').get_args() - assert args[:3] == ['-i', 'in', '-filter_complex'] - assert args[4:] == ['-map', '[s0]', 'out'] - match = re.match(r'\[0\]drawtext=text=a((.|\n)*)b\[s0\]', args[3], re.MULTILINE) - assert match is not None, 'Invalid -filter_complex arg: {!r}'.format(args[3]) - return match.group(1) - - expected_backslash_counts = { - 'x': 0, - '\'': 7, - '\\': 7, - '%': 4, - ':': 2, - ',': 1, - '[': 1, - ']': 1, - '=': 2, - '\n': 0, - } - for ch, expected_backslash_count in list(expected_backslash_counts.items()): - expected = '{}{}'.format('\\' * expected_backslash_count, ch) - actual = _get_drawtext_text_repr(ch) - assert expected == actual + """Test string escaping of normal filter args (e.g. ``text`` param of ``drawtext`` filter).""" + + def _get_drawtext_text_repr(text): + """Build a command-line arg using drawtext ``text`` param and extract the ``-filter_complex`` arg.""" + args = ffmpeg.input('in').drawtext('a{}b'.format(text)).output('out').get_args() + assert args[:3] == ['-i', 'in', '-filter_complex'] + assert args[4:] == ['-map', '[s0]', 'out'] + match = re.match(r'\[0\]drawtext=text=a((.|\n)*)b\[s0\]', args[3], re.MULTILINE) + assert match is not None, 'Invalid -filter_complex arg: {!r}'.format(args[3]) + return match.group(1) + + expected_backslash_counts = { + 'x': 0, + '\'': 7, + '\\': 7, + '%': 4, + ':': 2, + ',': 1, + '[': 1, + ']': 1, + '=': 2, + '\n': 0, + } + for ch, expected_backslash_count in list(expected_backslash_counts.items()): + expected = '{}{}'.format('\\' * expected_backslash_count, ch) + actual = _get_drawtext_text_repr(ch) + assert expected == actual # def test_version(): @@ -428,357 +428,357 @@ def _get_drawtext_text_repr(text): def test__compile(): - out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4') - assert out_file.compile() == ['ffmpeg', '-i', 'dummy.mp4', 'dummy2.mp4'] - assert out_file.compile(cmd='ffmpeg.old') == [ - 'ffmpeg.old', - '-i', - 'dummy.mp4', - 'dummy2.mp4', - ] + out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4') + assert out_file.compile() == ['ffmpeg', '-i', 'dummy.mp4', 'dummy2.mp4'] + assert out_file.compile(cmd='ffmpeg.old') == [ + 'ffmpeg.old', + '-i', + 'dummy.mp4', + 'dummy2.mp4', + ] @pytest.mark.parametrize('pipe_stdin', [True, False]) @pytest.mark.parametrize('pipe_stdout', [True, False]) @pytest.mark.parametrize('pipe_stderr', [True, False]) def test__run_async(mocker, pipe_stdin, pipe_stdout, pipe_stderr): - process__mock = mock.Mock() - popen__mock = mocker.patch.object(subprocess, 'Popen', return_value=process__mock) - stream = _get_simple_example() - process = ffmpeg.run_async( - stream, pipe_stdin=pipe_stdin, pipe_stdout=pipe_stdout, pipe_stderr=pipe_stderr - ) - assert process is process__mock - - expected_stdin = subprocess.PIPE if pipe_stdin else None - expected_stdout = subprocess.PIPE if pipe_stdout else None - expected_stderr = subprocess.PIPE if pipe_stderr else None - (args,), kwargs = popen__mock.call_args - assert args == ffmpeg.compile(stream) - assert kwargs == dict( - stdin=expected_stdin, stdout=expected_stdout, stderr=expected_stderr - ) + process__mock = mock.Mock() + popen__mock = mocker.patch.object(subprocess, 'Popen', return_value=process__mock) + stream = _get_simple_example() + process = ffmpeg.run_async( + stream, pipe_stdin=pipe_stdin, pipe_stdout=pipe_stdout, pipe_stderr=pipe_stderr + ) + assert process is process__mock + + expected_stdin = subprocess.PIPE if pipe_stdin else None + expected_stdout = subprocess.PIPE if pipe_stdout else None + expected_stderr = subprocess.PIPE if pipe_stderr else None + (args,), kwargs = popen__mock.call_args + assert args == ffmpeg.compile(stream) + assert kwargs == dict( + stdin=expected_stdin, stdout=expected_stdout, stderr=expected_stderr + ) def test__run(): - stream = _get_complex_filter_example() - out, err = ffmpeg.run(stream) - assert out is None - assert err is None + stream = _get_complex_filter_example() + out, err = ffmpeg.run(stream) + assert out is None + assert err is None @pytest.mark.parametrize('capture_stdout', [True, False]) @pytest.mark.parametrize('capture_stderr', [True, False]) def test__run__capture_out(mocker, capture_stdout, capture_stderr): - mocker.patch.object(ffmpeg._run, 'compile', return_value=['echo', 'test']) - stream = _get_simple_example() - out, err = ffmpeg.run( - stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr - ) - if capture_stdout: - assert out == 'test\n'.encode() - else: - assert out is None - if capture_stderr: - assert err == ''.encode() - else: - assert err is None + mocker.patch.object(ffmpeg._run, 'compile', return_value=['echo', 'test']) + stream = _get_simple_example() + out, err = ffmpeg.run( + stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr + ) + if capture_stdout: + assert out == 'test\n'.encode() + else: + assert out is None + if capture_stderr: + assert err == ''.encode() + else: + assert err is None def test__run__input_output(mocker): - mocker.patch.object(ffmpeg._run, 'compile', return_value=['cat']) - stream = _get_simple_example() - out, err = ffmpeg.run(stream, input='test'.encode(), capture_stdout=True) - assert out == 'test'.encode() - assert err is None + mocker.patch.object(ffmpeg._run, 'compile', return_value=['cat']) + stream = _get_simple_example() + out, err = ffmpeg.run(stream, input='test'.encode(), capture_stdout=True) + assert out == 'test'.encode() + assert err is None @pytest.mark.parametrize('capture_stdout', [True, False]) @pytest.mark.parametrize('capture_stderr', [True, False]) def test__run__error(mocker, capture_stdout, capture_stderr): - mocker.patch.object(ffmpeg._run, 'compile', return_value=['ffmpeg']) - stream = _get_complex_filter_example() - with pytest.raises(ffmpeg.Error) as excinfo: - out, err = ffmpeg.run( - stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr - ) - assert str(excinfo.value) == 'ffmpeg error (see stderr output for detail)' - out = excinfo.value.stdout - err = excinfo.value.stderr - if capture_stdout: - assert out == ''.encode() - else: - assert out is None - if capture_stderr: - assert err.decode().startswith('ffmpeg version') - else: - assert err is None + mocker.patch.object(ffmpeg._run, 'compile', return_value=['ffmpeg']) + stream = _get_complex_filter_example() + with pytest.raises(ffmpeg.Error) as excinfo: + out, err = ffmpeg.run( + stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr + ) + assert str(excinfo.value) == 'ffmpeg error (see stderr output for detail)' + out = excinfo.value.stdout + err = excinfo.value.stderr + if capture_stdout: + assert out == ''.encode() + else: + assert out is None + if capture_stderr: + assert err.decode().startswith('ffmpeg version') + else: + assert err is None def test__run__multi_output(): - in_ = ffmpeg.input(TEST_INPUT_FILE1) - out1 = in_.output(TEST_OUTPUT_FILE1) - out2 = in_.output(TEST_OUTPUT_FILE2) - ffmpeg.run([out1, out2], overwrite_output=True) + in_ = ffmpeg.input(TEST_INPUT_FILE1) + out1 = in_.output(TEST_OUTPUT_FILE1) + out2 = in_.output(TEST_OUTPUT_FILE2) + ffmpeg.run([out1, out2], overwrite_output=True) def test__run__dummy_cmd(): - stream = _get_complex_filter_example() - ffmpeg.run(stream, cmd='true') + stream = _get_complex_filter_example() + ffmpeg.run(stream, cmd='true') def test__run__dummy_cmd_list(): - stream = _get_complex_filter_example() - ffmpeg.run(stream, cmd=['true', 'ignored']) + stream = _get_complex_filter_example() + ffmpeg.run(stream, cmd=['true', 'ignored']) def test__filter__custom(): - stream = ffmpeg.input('dummy.mp4') - stream = ffmpeg.filter(stream, 'custom_filter', 'a', 'b', kwarg1='c') - stream = ffmpeg.output(stream, 'dummy2.mp4') - assert stream.get_args() == [ - '-i', - 'dummy.mp4', - '-filter_complex', - '[0]custom_filter=a:b:kwarg1=c[s0]', - '-map', - '[s0]', - 'dummy2.mp4', - ] + stream = ffmpeg.input('dummy.mp4') + stream = ffmpeg.filter(stream, 'custom_filter', 'a', 'b', kwarg1='c') + stream = ffmpeg.output(stream, 'dummy2.mp4') + assert stream.get_args() == [ + '-i', + 'dummy.mp4', + '-filter_complex', + '[0]custom_filter=a:b:kwarg1=c[s0]', + '-map', + '[s0]', + 'dummy2.mp4', + ] def test__filter__custom_fluent(): - stream = ( - ffmpeg.input('dummy.mp4') - .filter('custom_filter', 'a', 'b', kwarg1='c') - .output('dummy2.mp4') - ) - assert stream.get_args() == [ - '-i', - 'dummy.mp4', - '-filter_complex', - '[0]custom_filter=a:b:kwarg1=c[s0]', - '-map', - '[s0]', - 'dummy2.mp4', - ] + stream = ( + ffmpeg.input('dummy.mp4') + .filter('custom_filter', 'a', 'b', kwarg1='c') + .output('dummy2.mp4') + ) + assert stream.get_args() == [ + '-i', + 'dummy.mp4', + '-filter_complex', + '[0]custom_filter=a:b:kwarg1=c[s0]', + '-map', + '[s0]', + 'dummy2.mp4', + ] def test__merge_outputs(): - in_ = ffmpeg.input('in.mp4') - out1 = in_.output('out1.mp4') - out2 = in_.output('out2.mp4') - assert ffmpeg.merge_outputs(out1, out2).get_args() == [ - '-i', - 'in.mp4', - 'out1.mp4', - 'out2.mp4', - ] - assert ffmpeg.get_args([out1, out2]) == ['-i', 'in.mp4', 'out2.mp4', 'out1.mp4'] + in_ = ffmpeg.input('in.mp4') + out1 = in_.output('out1.mp4') + out2 = in_.output('out2.mp4') + assert ffmpeg.merge_outputs(out1, out2).get_args() == [ + '-i', + 'in.mp4', + 'out1.mp4', + 'out2.mp4', + ] + assert ffmpeg.get_args([out1, out2]) == ['-i', 'in.mp4', 'out2.mp4', 'out1.mp4'] def test__input__start_time(): - assert ffmpeg.input('in', ss=10.5).output('out').get_args() == [ - '-ss', - '10.5', - '-i', - 'in', - 'out', - ] - assert ffmpeg.input('in', ss=0.0).output('out').get_args() == [ - '-ss', - '0.0', - '-i', - 'in', - 'out', - ] + assert ffmpeg.input('in', ss=10.5).output('out').get_args() == [ + '-ss', + '10.5', + '-i', + 'in', + 'out', + ] + assert ffmpeg.input('in', ss=0.0).output('out').get_args() == [ + '-ss', + '0.0', + '-i', + 'in', + 'out', + ] def test_multi_passthrough(): - out1 = ffmpeg.input('in1.mp4').output('out1.mp4') - out2 = ffmpeg.input('in2.mp4').output('out2.mp4') - out = ffmpeg.merge_outputs(out1, out2) - assert ffmpeg.get_args(out) == [ - '-i', - 'in1.mp4', - '-i', - 'in2.mp4', - 'out1.mp4', - '-map', - '1', - 'out2.mp4', - ] - assert ffmpeg.get_args([out1, out2]) == [ - '-i', - 'in2.mp4', - '-i', - 'in1.mp4', - 'out2.mp4', - '-map', - '1', - 'out1.mp4', - ] + out1 = ffmpeg.input('in1.mp4').output('out1.mp4') + out2 = ffmpeg.input('in2.mp4').output('out2.mp4') + out = ffmpeg.merge_outputs(out1, out2) + assert ffmpeg.get_args(out) == [ + '-i', + 'in1.mp4', + '-i', + 'in2.mp4', + 'out1.mp4', + '-map', + '1', + 'out2.mp4', + ] + assert ffmpeg.get_args([out1, out2]) == [ + '-i', + 'in2.mp4', + '-i', + 'in1.mp4', + 'out2.mp4', + '-map', + '1', + 'out1.mp4', + ] def test_passthrough_selectors(): - i1 = ffmpeg.input(TEST_INPUT_FILE1) - args = ffmpeg.output(i1['1'], i1['2'], TEST_OUTPUT_FILE1).get_args() - assert args == [ - '-i', - TEST_INPUT_FILE1, - '-map', - '0:1', - '-map', - '0:2', - TEST_OUTPUT_FILE1, - ] + i1 = ffmpeg.input(TEST_INPUT_FILE1) + args = ffmpeg.output(i1['1'], i1['2'], TEST_OUTPUT_FILE1).get_args() + assert args == [ + '-i', + TEST_INPUT_FILE1, + '-map', + '0:1', + '-map', + '0:2', + TEST_OUTPUT_FILE1, + ] def test_mixed_passthrough_selectors(): - i1 = ffmpeg.input(TEST_INPUT_FILE1) - args = ffmpeg.output(i1['1'].hflip(), i1['2'], TEST_OUTPUT_FILE1).get_args() - assert args == [ - '-i', - TEST_INPUT_FILE1, - '-filter_complex', - '[0:1]hflip[s0]', - '-map', - '[s0]', - '-map', - '0:2', - TEST_OUTPUT_FILE1, - ] + i1 = ffmpeg.input(TEST_INPUT_FILE1) + args = ffmpeg.output(i1['1'].hflip(), i1['2'], TEST_OUTPUT_FILE1).get_args() + assert args == [ + '-i', + TEST_INPUT_FILE1, + '-filter_complex', + '[0:1]hflip[s0]', + '-map', + '[s0]', + '-map', + '0:2', + TEST_OUTPUT_FILE1, + ] def test_pipe(): - width = 32 - height = 32 - frame_size = width * height * 3 # 3 bytes for rgb24 - frame_count = 10 - start_frame = 2 - - out = ( - ffmpeg.input( - 'pipe:0', - format='rawvideo', - pixel_format='rgb24', - video_size=(width, height), - framerate=10, - ) - .trim(start_frame=start_frame) - .output('pipe:1', format='rawvideo') - ) - - args = out.get_args() - assert args == [ - '-f', - 'rawvideo', - '-video_size', - '{}x{}'.format(width, height), - '-framerate', - '10', - '-pixel_format', - 'rgb24', - '-i', - 'pipe:0', - '-filter_complex', - '[0]trim=start_frame=2[s0]', - '-map', - '[s0]', - '-f', - 'rawvideo', - 'pipe:1', - ] - - cmd = ['ffmpeg'] + args - p = subprocess.Popen( - cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE - ) - - in_data = bytes( - bytearray([random.randint(0, 255) for _ in range(frame_size * frame_count)]) - ) - p.stdin.write(in_data) # note: this could block, in which case need to use threads - p.stdin.close() - - out_data = p.stdout.read() - assert len(out_data) == frame_size * (frame_count - start_frame) - assert out_data == in_data[start_frame * frame_size :] + width = 32 + height = 32 + frame_size = width * height * 3 # 3 bytes for rgb24 + frame_count = 10 + start_frame = 2 + + out = ( + ffmpeg.input( + 'pipe:0', + format='rawvideo', + pixel_format='rgb24', + video_size=(width, height), + framerate=10, + ) + .trim(start_frame=start_frame) + .output('pipe:1', format='rawvideo') + ) + + args = out.get_args() + assert args == [ + '-f', + 'rawvideo', + '-video_size', + '{}x{}'.format(width, height), + '-framerate', + '10', + '-pixel_format', + 'rgb24', + '-i', + 'pipe:0', + '-filter_complex', + '[0]trim=start_frame=2[s0]', + '-map', + '[s0]', + '-f', + 'rawvideo', + 'pipe:1', + ] + + cmd = ['ffmpeg'] + args + p = subprocess.Popen( + cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE + ) + + in_data = bytes( + bytearray([random.randint(0, 255) for _ in range(frame_size * frame_count)]) + ) + p.stdin.write(in_data) # note: this could block, in which case need to use threads + p.stdin.close() + + out_data = p.stdout.read() + assert len(out_data) == frame_size * (frame_count - start_frame) + assert out_data == in_data[start_frame * frame_size :] def test__probe(): - data = ffmpeg.probe(TEST_INPUT_FILE1) - assert set(data.keys()) == {'format', 'streams'} - assert data['format']['duration'] == '7.036000' + data = ffmpeg.probe(TEST_INPUT_FILE1) + assert set(data.keys()) == {'format', 'streams'} + assert data['format']['duration'] == '7.036000' @pytest.mark.skipif(sys.version_info < (3, 3), reason="requires python3.3 or higher") def test__probe_timeout(): - with pytest.raises(subprocess.TimeoutExpired) as excinfo: - data = ffmpeg.probe(TEST_INPUT_FILE1, timeout=0) - assert 'timed out after 0 seconds' in str(excinfo.value) + with pytest.raises(subprocess.TimeoutExpired) as excinfo: + data = ffmpeg.probe(TEST_INPUT_FILE1, timeout=0) + assert 'timed out after 0 seconds' in str(excinfo.value) def test__probe__exception(): - with pytest.raises(ffmpeg.Error) as excinfo: - ffmpeg.probe(BOGUS_INPUT_FILE) - assert str(excinfo.value) == 'ffprobe error (see stderr output for detail)' - assert 'No such file or directory'.encode() in excinfo.value.stderr + with pytest.raises(ffmpeg.Error) as excinfo: + ffmpeg.probe(BOGUS_INPUT_FILE) + assert str(excinfo.value) == 'ffprobe error (see stderr output for detail)' + assert 'No such file or directory'.encode() in excinfo.value.stderr def test__probe__extra_args(): - data = ffmpeg.probe(TEST_INPUT_FILE1, show_frames=None) - assert set(data.keys()) == {'format', 'streams', 'frames'} + data = ffmpeg.probe(TEST_INPUT_FILE1, show_frames=None) + assert set(data.keys()) == {'format', 'streams', 'frames'} def get_filter_complex_input(flt, name): - m = re.search(r'\[([^]]+)\]{}(?=[[;]|$)'.format(name), flt) - if m: - return m.group(1) - else: - return None + m = re.search(r'\[([^]]+)\]{}(?=[[;]|$)'.format(name), flt) + if m: + return m.group(1) + else: + return None def get_filter_complex_outputs(flt, name): - m = re.search(r'(^|[];]){}((\[[^]]+\])+)(?=;|$)'.format(name), flt) - if m: - return m.group(2)[1:-1].split('][') - else: - return None + m = re.search(r'(^|[];]){}((\[[^]]+\])+)(?=;|$)'.format(name), flt) + if m: + return m.group(2)[1:-1].split('][') + else: + return None def test__get_filter_complex_input(): - assert get_filter_complex_input("", "scale") is None - assert get_filter_complex_input("scale", "scale") is None - assert get_filter_complex_input("scale[s3][s4];etc", "scale") is None - assert get_filter_complex_input("[s2]scale", "scale") == "s2" - assert get_filter_complex_input("[s2]scale;etc", "scale") == "s2" - assert get_filter_complex_input("[s2]scale[s3][s4];etc", "scale") == "s2" + assert get_filter_complex_input("", "scale") is None + assert get_filter_complex_input("scale", "scale") is None + assert get_filter_complex_input("scale[s3][s4];etc", "scale") is None + assert get_filter_complex_input("[s2]scale", "scale") == "s2" + assert get_filter_complex_input("[s2]scale;etc", "scale") == "s2" + assert get_filter_complex_input("[s2]scale[s3][s4];etc", "scale") == "s2" def test__get_filter_complex_outputs(): - assert get_filter_complex_outputs("", "scale") is None - assert get_filter_complex_outputs("scale", "scale") is None - assert get_filter_complex_outputs("scalex[s0][s1]", "scale") is None - assert get_filter_complex_outputs("scale[s0][s1]", "scale") == ['s0', 's1'] - assert get_filter_complex_outputs("[s5]scale[s0][s1]", "scale") == ['s0', 's1'] - assert get_filter_complex_outputs("[s5]scale[s1][s0]", "scale") == ['s1', 's0'] - assert get_filter_complex_outputs("[s5]scale[s1]", "scale") == ['s1'] - assert get_filter_complex_outputs("[s5]scale[s1];x", "scale") == ['s1'] - assert get_filter_complex_outputs("y;[s5]scale[s1];x", "scale") == ['s1'] + assert get_filter_complex_outputs("", "scale") is None + assert get_filter_complex_outputs("scale", "scale") is None + assert get_filter_complex_outputs("scalex[s0][s1]", "scale") is None + assert get_filter_complex_outputs("scale[s0][s1]", "scale") == ['s0', 's1'] + assert get_filter_complex_outputs("[s5]scale[s0][s1]", "scale") == ['s0', 's1'] + assert get_filter_complex_outputs("[s5]scale[s1][s0]", "scale") == ['s1', 's0'] + assert get_filter_complex_outputs("[s5]scale[s1]", "scale") == ['s1'] + assert get_filter_complex_outputs("[s5]scale[s1];x", "scale") == ['s1'] + assert get_filter_complex_outputs("y;[s5]scale[s1];x", "scale") == ['s1'] def test__multi_output_edge_label_order(): - scale2ref = ffmpeg.filter_multi_output( - [ffmpeg.input('x'), ffmpeg.input('y')], 'scale2ref' - ) - out = ffmpeg.merge_outputs( - scale2ref[1].filter('scale').output('a'), - scale2ref[10000].filter('hflip').output('b'), - ) - - args = out.get_args() - flt_cmpl = args[args.index('-filter_complex') + 1] - out1, out2 = get_filter_complex_outputs(flt_cmpl, 'scale2ref') - assert out1 == get_filter_complex_input(flt_cmpl, 'scale') - assert out2 == get_filter_complex_input(flt_cmpl, 'hflip') + scale2ref = ffmpeg.filter_multi_output( + [ffmpeg.input('x'), ffmpeg.input('y')], 'scale2ref' + ) + out = ffmpeg.merge_outputs( + scale2ref[1].filter('scale').output('a'), + scale2ref[10000].filter('hflip').output('b'), + ) + + args = out.get_args() + flt_cmpl = args[args.index('-filter_complex') + 1] + out1, out2 = get_filter_complex_outputs(flt_cmpl, 'scale2ref') + assert out1 == get_filter_complex_input(flt_cmpl, 'scale') + assert out2 == get_filter_complex_input(flt_cmpl, 'hflip')