blackteay/ffmpeg-python
1
0
Fork 0
mirror of https://github.com/kkroening/ffmpeg-python.git synced 2025-04-05 04:22:51 +08:00
Code Issues Projects Releases Wiki Activity

Merge pull request #213 from kkroening/black

Browse Source 
Use Black formatter
This commit is contained in:
Karl Kroening 2019-06-10 17:52:48 -05:00 committed by GitHub
commit ab42ab4dfc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 435 additions and 249 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
ffmpeg/_ffmpeg.py
View File

@ -94,9 +94,4 @@ def output(*streams_and_filename, **kwargs):
return OutputNode(streams, output.__name__, kwargs=kwargs).stream() return OutputNode(streams, output.__name__, kwargs=kwargs).stream()
__all__ = [ __all__ = ['input', 'merge_outputs', 'output', 'overwrite_output']
'input',
'merge_outputs',
'output',
'overwrite_output',
]

26
ffmpeg/_filters.py
View File

@ -21,7 +21,9 @@ def filter_multi_output(stream_spec, filter_name, *args, **kwargs):
ffmpeg.concat(split0, split1).output('out.mp4').run() ffmpeg.concat(split0, split1).output('out.mp4').run()
``` ```
""" """
return FilterNode(stream_spec, filter_name, args=args, kwargs=kwargs, max_inputs=None) return FilterNode(
stream_spec, filter_name, args=args, kwargs=kwargs, max_inputs=None
)
@filter_operator() @filter_operator()
@ -144,7 +146,12 @@ def overlay(main_parent_node, overlay_parent_node, eof_action='repeat', **kwargs
Official documentation: `overlay <https://ffmpeg.org/ffmpeg-filters.html#overlay-1>`__ Official documentation: `overlay <https://ffmpeg.org/ffmpeg-filters.html#overlay-1>`__
""" """
kwargs['eof_action'] = eof_action kwargs['eof_action'] = eof_action
return FilterNode([main_parent_node, overlay_parent_node], overlay.__name__, kwargs=kwargs, max_inputs=2).stream() return FilterNode(
[main_parent_node, overlay_parent_node],
overlay.__name__,
kwargs=kwargs,
max_inputs=2,
).stream()
@filter_operator() @filter_operator()
@ -180,10 +187,7 @@ def crop(stream, x, y, width, height, **kwargs):
Official documentation: `crop <https://ffmpeg.org/ffmpeg-filters.html#crop>`__ Official documentation: `crop <https://ffmpeg.org/ffmpeg-filters.html#crop>`__
""" """
return FilterNode( return FilterNode(
stream, stream, crop.__name__, args=[width, height, x, y], kwargs=kwargs
crop.__name__,
args=[width, height, x, y],
kwargs=kwargs
).stream() ).stream()
@ -209,7 +213,9 @@ def drawbox(stream, x, y, width, height, color, thickness=None, **kwargs):
""" """
if thickness: if thickness:
kwargs['t'] = thickness kwargs['t'] = thickness
return FilterNode(stream, drawbox.__name__, args=[x, y, width, height, color], kwargs=kwargs).stream() return FilterNode(
stream, drawbox.__name__, args=[x, y, width, height, color], kwargs=kwargs
).stream()
@filter_operator() @filter_operator()
@ -385,8 +391,10 @@ def concat(*streams, **kwargs):
stream_count = video_stream_count + audio_stream_count stream_count = video_stream_count + audio_stream_count
if len(streams) % stream_count != 0: if len(streams) % stream_count != 0:
raise ValueError( raise ValueError(
'Expected concat input streams to have length multiple of {} (v={}, a={}); got {}' 'Expected concat input streams to have length multiple of {} (v={}, a={}); got {}'.format(
.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) kwargs['n'] = int(len(streams) / stream_count)
return FilterNode(streams, concat.__name__, kwargs=kwargs, max_inputs=None).stream() return FilterNode(streams, concat.__name__, kwargs=kwargs, max_inputs=None).stream()

4
ffmpeg/_probe.py
View File

@ -24,6 +24,4 @@ def probe(filename, cmd='ffprobe', **kwargs):
return json.loads(out.decode('utf-8')) return json.loads(out.decode('utf-8'))
__all__ = [ __all__ = ['probe']
'probe',
]

74
ffmpeg/_run.py
View File

@ -8,10 +8,7 @@ import copy
import operator import operator
import subprocess import subprocess
from ._ffmpeg import ( from ._ffmpeg import input, output
input,
output,
)
from .nodes import ( from .nodes import (
get_stream_spec_nodes, get_stream_spec_nodes,
FilterNode, FilterNode,
@ -24,7 +21,9 @@ from .nodes import (
class Error(Exception): class Error(Exception):
def __init__(self, cmd, stdout, stderr): def __init__(self, cmd, stdout, stderr):
super(Error, self).__init__('{} error (see stderr output for detail)'.format(cmd)) super(Error, self).__init__(
'{} error (see stderr output for detail)'.format(cmd)
)
self.stdout = stdout self.stdout = stdout
self.stderr = stderr self.stderr = stderr
@ -69,9 +68,15 @@ def _format_output_stream_name(stream_name_map, edge):
def _get_filter_spec(node, outgoing_edge_map, stream_name_map): def _get_filter_spec(node, outgoing_edge_map, stream_name_map):
incoming_edges = node.incoming_edges incoming_edges = node.incoming_edges
outgoing_edges = get_outgoing_edges(node, outgoing_edge_map) outgoing_edges = get_outgoing_edges(node, outgoing_edge_map)
inputs = [_format_input_stream_name(stream_name_map, edge) for edge in incoming_edges] inputs = [
outputs = [_format_output_stream_name(stream_name_map, edge) for edge in outgoing_edges] _format_input_stream_name(stream_name_map, edge) for edge in incoming_edges
filter_spec = '{}{}{}'.format(''.join(inputs), node._get_filter(outgoing_edges), ''.join(outputs)) ]
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 return filter_spec
@ -84,14 +89,20 @@ def _allocate_filter_stream_names(filter_nodes, outgoing_edge_maps, stream_name_
# TODO: automatically insert `splits` ahead of time via graph transformation. # TODO: automatically insert `splits` ahead of time via graph transformation.
raise ValueError( raise ValueError(
'Encountered {} with multiple outgoing edges with same upstream label {!r}; a ' 'Encountered {} with multiple outgoing edges with same upstream label {!r}; a '
'`split` filter is probably required'.format(upstream_node, upstream_label)) '`split` filter is probably required'.format(
upstream_node, upstream_label
)
)
stream_name_map[upstream_node, upstream_label] = 's{}'.format(stream_count) stream_name_map[upstream_node, upstream_label] = 's{}'.format(stream_count)
stream_count += 1 stream_count += 1
def _get_filter_arg(filter_nodes, outgoing_edge_maps, stream_name_map): def _get_filter_arg(filter_nodes, outgoing_edge_maps, stream_name_map):
_allocate_filter_stream_names(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] filter_specs = [
_get_filter_spec(node, outgoing_edge_maps[node], stream_name_map)
for node in filter_nodes
]
return ';'.join(filter_specs) return ';'.join(filter_specs)
@ -109,7 +120,9 @@ def _get_output_args(node, stream_name_map):
for edge in node.incoming_edges: for edge in node.incoming_edges:
# edge = node.incoming_edges[0] # edge = node.incoming_edges[0]
stream_name = _format_input_stream_name(stream_name_map, edge, is_final_arg=True) stream_name = _format_input_stream_name(
stream_name_map, edge, is_final_arg=True
)
if stream_name != '0' or len(node.incoming_edges) > 1: if stream_name != '0' or len(node.incoming_edges) > 1:
args += ['-map', stream_name] args += ['-map', stream_name]
@ -123,7 +136,9 @@ def _get_output_args(node, stream_name_map):
args += ['-b:a', str(kwargs.pop('audio_bitrate'))] args += ['-b:a', str(kwargs.pop('audio_bitrate'))]
if 'video_size' in kwargs: if 'video_size' in kwargs:
video_size = kwargs.pop('video_size') video_size = kwargs.pop('video_size')
if not isinstance(video_size, basestring) and isinstance(video_size, collections.Iterable): if not isinstance(video_size, basestring) and isinstance(
video_size, collections.Iterable
):
video_size = '{}x{}'.format(video_size[0], video_size[1]) video_size = '{}x{}'.format(video_size[0], video_size[1])
args += ['-video_size', video_size] args += ['-video_size', video_size]
args += convert_kwargs_to_cmd_line_args(kwargs) args += convert_kwargs_to_cmd_line_args(kwargs)
@ -147,7 +162,9 @@ def get_args(stream_spec, overwrite_output=False):
args += reduce(operator.add, [_get_input_args(node) for node in input_nodes]) args += reduce(operator.add, [_get_input_args(node) for node in input_nodes])
if filter_arg: if filter_arg:
args += ['-filter_complex', 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_output_args(node, stream_name_map) for node in output_nodes]
)
args += reduce(operator.add, [_get_global_args(node) for node in global_nodes], []) args += reduce(operator.add, [_get_global_args(node) for node in global_nodes], [])
if overwrite_output: if overwrite_output:
args += ['-y'] args += ['-y']
@ -175,8 +192,14 @@ def compile(stream_spec, cmd='ffmpeg', overwrite_output=False):
@output_operator() @output_operator()
def run_async( def run_async(
stream_spec, cmd='ffmpeg', pipe_stdin=False, pipe_stdout=False, pipe_stderr=False, stream_spec,
quiet=False, overwrite_output=False): cmd='ffmpeg',
pipe_stdin=False,
pipe_stdout=False,
pipe_stderr=False,
quiet=False,
overwrite_output=False,
):
"""Asynchronously invoke ffmpeg for the supplied node graph. """Asynchronously invoke ffmpeg for the supplied node graph.
Args: Args:
@ -259,13 +282,20 @@ def run_async(
stdout_stream = subprocess.PIPE if pipe_stdout or quiet else None stdout_stream = subprocess.PIPE if pipe_stdout or quiet else None
stderr_stream = subprocess.PIPE if pipe_stderr or quiet else None stderr_stream = subprocess.PIPE if pipe_stderr or quiet else None
return subprocess.Popen( return subprocess.Popen(
args, stdin=stdin_stream, stdout=stdout_stream, stderr=stderr_stream) args, stdin=stdin_stream, stdout=stdout_stream, stderr=stderr_stream
)
@output_operator() @output_operator()
def run( def run(
stream_spec, cmd='ffmpeg', capture_stdout=False, capture_stderr=False, input=None, stream_spec,
quiet=False, overwrite_output=False): cmd='ffmpeg',
capture_stdout=False,
capture_stderr=False,
input=None,
quiet=False,
overwrite_output=False,
):
"""Invoke ffmpeg for the supplied node graph. """Invoke ffmpeg for the supplied node graph.
Args: Args:
@ -296,10 +326,4 @@ def run(
return out, err return out, err
__all__ = [ __all__ = ['compile', 'Error', 'get_args', 'run', 'run_async']
'compile',
'Error',
'get_args',
'run',
'run_async',
]

8
ffmpeg/_utils.py
View File

@ -34,8 +34,11 @@ def with_metaclass(meta, *bases):
if sys.version_info.major >= 3: if sys.version_info.major >= 3:
class basestring(with_metaclass(BaseBaseString)): class basestring(with_metaclass(BaseBaseString)):
pass pass
else: else:
# noinspection PyUnresolvedReferences,PyCompatibility # noinspection PyUnresolvedReferences,PyCompatibility
from builtins import basestring from builtins import basestring
@ -52,7 +55,10 @@ def _recursive_repr(item):
elif isinstance(item, list): elif isinstance(item, list):
result = '[{}]'.format(', '.join([_recursive_repr(x) for x in item])) result = '[{}]'.format(', '.join([_recursive_repr(x) for x in item]))
elif isinstance(item, dict): elif isinstance(item, dict):
kv_pairs = ['{}: {}'.format(_recursive_repr(k), _recursive_repr(item[k])) for k in sorted(item)] kv_pairs = [
'{}: {}'.format(_recursive_repr(k), _recursive_repr(item[k]))
for k in sorted(item)
]
result = '{' + ', '.join(kv_pairs) + '}' result = '{' + ', '.join(kv_pairs) + '}'
else: else:
result = repr(item) result = repr(item)

28
ffmpeg/_view.py
View File

@ -34,8 +34,10 @@ def view(stream_spec, detail=False, filename=None, pipe=False, **kwargs):
try: try:
import graphviz import graphviz
except ImportError: except ImportError:
raise ImportError('failed to import graphviz; please make sure graphviz is installed (e.g. `pip install ' raise ImportError(
'graphviz`)') 'failed to import graphviz; please make sure graphviz is installed (e.g. `pip install '
'graphviz`)'
)
show_labels = kwargs.pop('show_labels', True) show_labels = kwargs.pop('show_labels', True)
if pipe and filename is not None: if pipe and filename is not None:
@ -49,7 +51,9 @@ def view(stream_spec, detail=False, filename=None, pipe=False, **kwargs):
graph = graphviz.Digraph(format='png') graph = graphviz.Digraph(format='png')
graph.attr(rankdir='LR') graph.attr(rankdir='LR')
if len(list(kwargs.keys())) != 0: if len(list(kwargs.keys())) != 0:
raise ValueError('Invalid kwargs key(s): {}'.format(', '.join(list(kwargs.keys())))) raise ValueError(
'Invalid kwargs key(s): {}'.format(', '.join(list(kwargs.keys())))
)
for node in sorted_nodes: for node in sorted_nodes:
color = _get_node_color(node) color = _get_node_color(node)
@ -57,11 +61,15 @@ def view(stream_spec, detail=False, filename=None, pipe=False, **kwargs):
if detail: if detail:
lines = [node.short_repr] lines = [node.short_repr]
lines += ['{!r}'.format(arg) for arg in node.args] lines += ['{!r}'.format(arg) for arg in node.args]
lines += ['{}={!r}'.format(key, node.kwargs[key]) for key in sorted(node.kwargs)] lines += [
'{}={!r}'.format(key, node.kwargs[key]) for key in sorted(node.kwargs)
]
node_text = '\n'.join(lines) node_text = '\n'.join(lines)
else: else:
node_text = node.short_repr node_text = node.short_repr
graph.node(str(hash(node)), node_text, shape='box', style='filled', fillcolor=color) graph.node(
str(hash(node)), node_text, shape='box', style='filled', fillcolor=color
)
outgoing_edge_map = outgoing_edge_maps.get(node, {}) outgoing_edge_map = outgoing_edge_maps.get(node, {})
for edge in get_outgoing_edges(node, outgoing_edge_map): for edge in get_outgoing_edges(node, outgoing_edge_map):
@ -70,7 +78,11 @@ def view(stream_spec, detail=False, filename=None, pipe=False, **kwargs):
down_label = edge.downstream_label down_label = edge.downstream_label
up_selector = edge.upstream_selector 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 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: if up_label is None:
up_label = '' up_label = ''
if up_selector is not None: if up_selector is not None:
@ -93,6 +105,4 @@ def view(stream_spec, detail=False, filename=None, pipe=False, **kwargs):
return stream_spec return stream_spec
__all__ = [ __all__ = ['view']
'view',
]

65
ffmpeg/dag.py
View File

@ -70,14 +70,31 @@ class DagNode(object):
raise NotImplementedError() raise NotImplementedError()
DagEdge = namedtuple('DagEdge', ['downstream_node', 'downstream_label', 'upstream_node', 'upstream_label', 'upstream_selector']) DagEdge = namedtuple(
'DagEdge',
[
'downstream_node',
'downstream_label',
'upstream_node',
'upstream_label',
'upstream_selector',
],
)
def get_incoming_edges(downstream_node, incoming_edge_map): def get_incoming_edges(downstream_node, incoming_edge_map):
edges = [] edges = []
for downstream_label, upstream_info in list(incoming_edge_map.items()): for downstream_label, upstream_info in list(incoming_edge_map.items()):
upstream_node, upstream_label, upstream_selector = upstream_info upstream_node, upstream_label, upstream_selector = upstream_info
edges += [DagEdge(downstream_node, downstream_label, upstream_node, upstream_label, upstream_selector)] edges += [
DagEdge(
downstream_node,
downstream_label,
upstream_node,
upstream_label,
upstream_selector,
)
]
return edges return edges
@ -86,7 +103,15 @@ def get_outgoing_edges(upstream_node, outgoing_edge_map):
for upstream_label, downstream_infos in list(outgoing_edge_map.items()): for upstream_label, downstream_infos in list(outgoing_edge_map.items()):
for downstream_info in downstream_infos: for downstream_info in downstream_infos:
downstream_node, downstream_label, downstream_selector = downstream_info downstream_node, downstream_label, downstream_selector = downstream_info
edges += [DagEdge(downstream_node, downstream_label, upstream_node, upstream_label, downstream_selector)] edges += [
DagEdge(
downstream_node,
downstream_label,
upstream_node,
upstream_label,
downstream_selector,
)
]
return edges return edges
@ -99,7 +124,15 @@ class KwargReprNode(DagNode):
hashes = [] hashes = []
for downstream_label, upstream_info in list(self.incoming_edge_map.items()): for downstream_label, upstream_info in list(self.incoming_edge_map.items()):
upstream_node, upstream_label, upstream_selector = upstream_info upstream_node, upstream_label, upstream_selector = upstream_info
hashes += [hash(x) for x in [downstream_label, upstream_node, upstream_label, upstream_selector]] hashes += [
hash(x)
for x in [
downstream_label,
upstream_node,
upstream_label,
upstream_selector,
]
]
return hashes return hashes
@property @property
@ -130,7 +163,9 @@ class KwargReprNode(DagNode):
def long_repr(self, include_hash=True): def long_repr(self, include_hash=True):
formatted_props = ['{!r}'.format(arg) for arg in self.args] formatted_props = ['{!r}'.format(arg) for arg in self.args]
formatted_props += ['{}={!r}'.format(key, self.kwargs[key]) for key in sorted(self.kwargs)] formatted_props += [
'{}={!r}'.format(key, self.kwargs[key]) for key in sorted(self.kwargs)
]
out = '{}({})'.format(self.name, ', '.join(formatted_props)) out = '{}({})'.format(self.name, ', '.join(formatted_props))
if include_hash: if include_hash:
out += ' <{}>'.format(self.short_hash) out += ' <{}>'.format(self.short_hash)
@ -157,21 +192,35 @@ def topo_sort(downstream_nodes):
sorted_nodes = [] sorted_nodes = []
outgoing_edge_maps = {} outgoing_edge_maps = {}
def visit(upstream_node, upstream_label, downstream_node, downstream_label, downstream_selector=None): def visit(
upstream_node,
upstream_label,
downstream_node,
downstream_label,
downstream_selector=None,
):
if upstream_node in marked_nodes: if upstream_node in marked_nodes:
raise RuntimeError('Graph is not a DAG') raise RuntimeError('Graph is not a DAG')
if downstream_node is not None: if downstream_node is not None:
outgoing_edge_map = outgoing_edge_maps.get(upstream_node, {}) outgoing_edge_map = outgoing_edge_maps.get(upstream_node, {})
outgoing_edge_infos = outgoing_edge_map.get(upstream_label, []) outgoing_edge_infos = outgoing_edge_map.get(upstream_label, [])
outgoing_edge_infos += [(downstream_node, downstream_label, downstream_selector)] outgoing_edge_infos += [
(downstream_node, downstream_label, downstream_selector)
]
outgoing_edge_map[upstream_label] = outgoing_edge_infos outgoing_edge_map[upstream_label] = outgoing_edge_infos
outgoing_edge_maps[upstream_node] = outgoing_edge_map outgoing_edge_maps[upstream_node] = outgoing_edge_map
if upstream_node not in sorted_nodes: if upstream_node not in sorted_nodes:
marked_nodes.append(upstream_node) marked_nodes.append(upstream_node)
for edge in upstream_node.incoming_edges: for edge in upstream_node.incoming_edges:
visit(edge.upstream_node, edge.upstream_label, edge.downstream_node, edge.downstream_label, edge.upstream_selector) visit(
edge.upstream_node,
edge.upstream_label,
edge.downstream_node,
edge.downstream_label,
edge.upstream_selector,
)
marked_nodes.remove(upstream_node) marked_nodes.remove(upstream_node)
sorted_nodes.append(upstream_node) sorted_nodes.append(upstream_node)

80
ffmpeg/nodes.py
View File

@ -23,10 +23,15 @@ def _get_types_str(types):
class Stream(object): class Stream(object):
"""Represents the outgoing edge of an upstream node; may be used to create more downstream nodes.""" """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): def __init__(
self, upstream_node, upstream_label, node_types, upstream_selector=None
):
if not _is_of_types(upstream_node, node_types): 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( raise TypeError(
_get_types_str(node_types), type(upstream_node))) '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.node = upstream_node
self.label = upstream_label self.label = upstream_label
self.selector = upstream_selector self.selector = upstream_selector
@ -42,7 +47,9 @@ class Stream(object):
selector = '' selector = ''
if self.selector: if self.selector:
selector = ':{}'.format(self.selector) selector = ':{}'.format(self.selector)
out = '{}[{!r}{}] <{}>'.format(node_repr, self.label, selector, self.node.short_hash) out = '{}[{!r}{}] <{}>'.format(
node_repr, self.label, selector, self.node.short_hash
)
return out return out
def __getitem__(self, index): def __getitem__(self, index):
@ -146,26 +153,50 @@ class Node(KwargReprNode):
@classmethod @classmethod
def __check_input_len(cls, stream_map, min_inputs, max_inputs): def __check_input_len(cls, stream_map, min_inputs, max_inputs):
if min_inputs is not None and len(stream_map) < min_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))) 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: 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))) raise ValueError(
'Expected at most {} input stream(s); got {}'.format(
max_inputs, len(stream_map)
)
)
@classmethod @classmethod
def __check_input_types(cls, stream_map, incoming_stream_types): def __check_input_types(cls, stream_map, incoming_stream_types):
for stream in list(stream_map.values()): for stream in list(stream_map.values()):
if not _is_of_types(stream, incoming_stream_types): if not _is_of_types(stream, incoming_stream_types):
raise TypeError('Expected incoming stream(s) to be of one of the following types: {}; got {}' raise TypeError(
.format(_get_types_str(incoming_stream_types), type(stream))) 'Expected incoming stream(s) to be of one of the following types: {}; got {}'.format(
_get_types_str(incoming_stream_types), type(stream)
)
)
@classmethod @classmethod
def __get_incoming_edge_map(cls, stream_map): def __get_incoming_edge_map(cls, stream_map):
incoming_edge_map = {} incoming_edge_map = {}
for downstream_label, upstream in list(stream_map.items()): for downstream_label, upstream in list(stream_map.items()):
incoming_edge_map[downstream_label] = (upstream.node, upstream.label, upstream.selector) incoming_edge_map[downstream_label] = (
upstream.node,
upstream.label,
upstream.selector,
)
return incoming_edge_map return incoming_edge_map
def __init__(self, stream_spec, name, incoming_stream_types, outgoing_stream_type, min_inputs, def __init__(
max_inputs, args=[], kwargs={}): self,
stream_spec,
name,
incoming_stream_types,
outgoing_stream_type,
min_inputs,
max_inputs,
args=[],
kwargs={},
):
stream_map = get_stream_map(stream_spec) stream_map = get_stream_map(stream_spec)
self.__check_input_len(stream_map, min_inputs, max_inputs) self.__check_input_len(stream_map, min_inputs, max_inputs)
self.__check_input_types(stream_map, incoming_stream_types) self.__check_input_types(stream_map, incoming_stream_types)
@ -203,8 +234,9 @@ class Node(KwargReprNode):
class FilterableStream(Stream): class FilterableStream(Stream):
def __init__(self, upstream_node, upstream_label, upstream_selector=None): def __init__(self, upstream_node, upstream_label, upstream_selector=None):
super(FilterableStream, self).__init__(upstream_node, upstream_label, {InputNode, FilterNode}, super(FilterableStream, self).__init__(
upstream_selector) upstream_node, upstream_label, {InputNode, FilterNode}, upstream_selector
)
# noinspection PyMethodOverriding # noinspection PyMethodOverriding
@ -220,7 +252,7 @@ class InputNode(Node):
min_inputs=0, min_inputs=0,
max_inputs=0, max_inputs=0,
args=args, args=args,
kwargs=kwargs kwargs=kwargs,
) )
@property @property
@ -239,7 +271,7 @@ class FilterNode(Node):
min_inputs=1, min_inputs=1,
max_inputs=max_inputs, max_inputs=max_inputs,
args=args, args=args,
kwargs=kwargs kwargs=kwargs,
) )
"""FilterNode""" """FilterNode"""
@ -279,7 +311,7 @@ class OutputNode(Node):
min_inputs=1, min_inputs=1,
max_inputs=None, max_inputs=None,
args=args, args=args,
kwargs=kwargs kwargs=kwargs,
) )
@property @property
@ -289,8 +321,12 @@ class OutputNode(Node):
class OutputStream(Stream): class OutputStream(Stream):
def __init__(self, upstream_node, upstream_label, upstream_selector=None): def __init__(self, upstream_node, upstream_label, upstream_selector=None):
super(OutputStream, self).__init__(upstream_node, upstream_label, {OutputNode, GlobalNode, MergeOutputsNode}, super(OutputStream, self).__init__(
upstream_selector=upstream_selector) upstream_node,
upstream_label,
{OutputNode, GlobalNode, MergeOutputsNode},
upstream_selector=upstream_selector,
)
# noinspection PyMethodOverriding # noinspection PyMethodOverriding
@ -302,7 +338,7 @@ class MergeOutputsNode(Node):
incoming_stream_types={OutputStream}, incoming_stream_types={OutputStream},
outgoing_stream_type=OutputStream, outgoing_stream_type=OutputStream,
min_inputs=1, min_inputs=1,
max_inputs=None max_inputs=None,
) )
@ -317,7 +353,7 @@ class GlobalNode(Node):
min_inputs=1, min_inputs=1,
max_inputs=1, max_inputs=1,
args=args, args=args,
kwargs=kwargs kwargs=kwargs,
) )
@ -338,6 +374,4 @@ def output_operator(name=None):
return stream_operator(stream_classes={OutputStream}, name=name) return stream_operator(stream_classes={OutputStream}, name=name)
__all__ = [ __all__ = ['Stream']
'Stream',
]

376
ffmpeg/tests/test_ffmpeg.py
View File

@ -30,7 +30,10 @@ subprocess.check_call(['ffmpeg', '-version'])
def test_escape_chars(): 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', ':\\') == '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('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(123, ':\\') == '123'
@ -62,23 +65,16 @@ def test_fluent_concat():
def test_fluent_output(): def test_fluent_output():
(ffmpeg ffmpeg.input('dummy.mp4').trim(start_frame=10, end_frame=20).output('dummy2.mp4')
.input('dummy.mp4')
.trim(start_frame=10, end_frame=20)
.output('dummy2.mp4')
)
def test_fluent_complex_filter(): def test_fluent_complex_filter():
in_file = ffmpeg.input('dummy.mp4') in_file = ffmpeg.input('dummy.mp4')
return (ffmpeg return ffmpeg.concat(
.concat( in_file.trim(start_frame=10, end_frame=20),
in_file.trim(start_frame=10, end_frame=20), in_file.trim(start_frame=30, end_frame=40),
in_file.trim(start_frame=30, end_frame=40), in_file.trim(start_frame=50, end_frame=60),
in_file.trim(start_frame=50, end_frame=60) ).output('dummy2.mp4')
)
.output('dummy2.mp4')
)
def test_node_repr(): def test_node_repr():
@ -88,21 +84,35 @@ def test_node_repr():
trim3 = ffmpeg.trim(in_file, start_frame=50, end_frame=60) trim3 = ffmpeg.trim(in_file, start_frame=50, end_frame=60)
concatted = ffmpeg.concat(trim1, trim2, trim3) concatted = ffmpeg.concat(trim1, trim2, trim3)
output = ffmpeg.output(concatted, 'dummy2.mp4') output = ffmpeg.output(concatted, 'dummy2.mp4')
assert repr(in_file.node) == 'input(filename={!r}) <{}>'.format('dummy.mp4', in_file.node.short_hash) assert repr(in_file.node) == 'input(filename={!r}) <{}>'.format(
assert repr(trim1.node) == 'trim(end_frame=20, start_frame=10) <{}>'.format(trim1.node.short_hash) 'dummy.mp4', in_file.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(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(concatted.node) == 'concat(n=3) <{}>'.format(concatted.node.short_hash)
assert repr(output.node) == 'output(filename={!r}) <{}>'.format('dummy2.mp4', output.node.short_hash) assert repr(output.node) == 'output(filename={!r}) <{}>'.format(
'dummy2.mp4', output.node.short_hash
)
def test_stream_repr(): def test_stream_repr():
in_file = ffmpeg.input('dummy.mp4') in_file = ffmpeg.input('dummy.mp4')
assert repr(in_file) == 'input(filename={!r})[None] <{}>'.format('dummy.mp4', in_file.node.short_hash) assert repr(in_file) == 'input(filename={!r})[None] <{}>'.format(
'dummy.mp4', in_file.node.short_hash
)
split0 = in_file.filter_multi_output('split')[0] split0 = in_file.filter_multi_output('split')[0]
assert repr(split0) == 'split()[0] <{}>'.format(split0.node.short_hash) assert repr(split0) == 'split()[0] <{}>'.format(split0.node.short_hash)
dummy_out = in_file.filter_multi_output('dummy')['out'] dummy_out = in_file.filter_multi_output('dummy')['out']
assert repr(dummy_out) == 'dummy()[{!r}] <{}>'.format(dummy_out.label, dummy_out.node.short_hash) assert repr(dummy_out) == 'dummy()[{!r}] <{}>'.format(
dummy_out.label, dummy_out.node.short_hash
)
def test__get_args__simple(): def test__get_args__simple():
@ -111,8 +121,18 @@ def test__get_args__simple():
def test_global_args(): def test_global_args():
out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4').global_args('-progress', 'someurl') out_file = (
assert out_file.get_args() == ['-i', 'dummy.mp4', 'dummy2.mp4', '-progress', 'someurl'] 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(): def _get_simple_example():
@ -120,18 +140,14 @@ def _get_simple_example():
def _get_complex_filter_example(): def _get_complex_filter_example():
split = (ffmpeg split = ffmpeg.input(TEST_INPUT_FILE1).vflip().split()
.input(TEST_INPUT_FILE1)
.vflip()
.split()
)
split0 = split[0] split0 = split[0]
split1 = split[1] split1 = split[1]
overlay_file = ffmpeg.input(TEST_OVERLAY_FILE) overlay_file = ffmpeg.input(TEST_OVERLAY_FILE)
overlay_file = ffmpeg.crop(overlay_file, 10, 10, 158, 112) overlay_file = ffmpeg.crop(overlay_file, 10, 10, 158, 112)
return (ffmpeg return (
.concat( ffmpeg.concat(
split0.trim(start_frame=10, end_frame=20), split0.trim(start_frame=10, end_frame=20),
split1.trim(start_frame=30, end_frame=40), split1.trim(start_frame=30, end_frame=40),
) )
@ -145,20 +161,25 @@ def _get_complex_filter_example():
def test__get_args__complex_filter(): def test__get_args__complex_filter():
out = _get_complex_filter_example() out = _get_complex_filter_example()
args = ffmpeg.get_args(out) args = ffmpeg.get_args(out)
assert args == ['-i', TEST_INPUT_FILE1, assert args == [
'-i', TEST_OVERLAY_FILE, '-i',
TEST_INPUT_FILE1,
'-i',
TEST_OVERLAY_FILE,
'-filter_complex', '-filter_complex',
'[0]vflip[s0];' \ '[0]vflip[s0];'
'[s0]split=2[s1][s2];' \ '[s0]split=2[s1][s2];'
'[s1]trim=end_frame=20:start_frame=10[s3];' \ '[s1]trim=end_frame=20:start_frame=10[s3];'
'[s2]trim=end_frame=40:start_frame=30[s4];' \ '[s2]trim=end_frame=40:start_frame=30[s4];'
'[s3][s4]concat=n=2[s5];' \ '[s3][s4]concat=n=2[s5];'
'[1]crop=158:112:10:10[s6];' \ '[1]crop=158:112:10:10[s6];'
'[s6]hflip[s7];' \ '[s6]hflip[s7];'
'[s5][s7]overlay=eof_action=repeat[s8];' \ '[s5][s7]overlay=eof_action=repeat[s8];'
'[s8]drawbox=50:50:120:120:red:t=5[s9]', '[s8]drawbox=50:50:120:120:red:t=5[s9]',
'-map', '[s9]', TEST_OUTPUT_FILE1, '-map',
'-y' '[s9]',
TEST_OUTPUT_FILE1,
'-y',
] ]
@ -167,11 +188,15 @@ def test_combined_output():
i2 = ffmpeg.input(TEST_OVERLAY_FILE) i2 = ffmpeg.input(TEST_OVERLAY_FILE)
out = ffmpeg.output(i1, i2, TEST_OUTPUT_FILE1) out = ffmpeg.output(i1, i2, TEST_OUTPUT_FILE1)
assert out.get_args() == [ assert out.get_args() == [
'-i', TEST_INPUT_FILE1, '-i',
'-i', TEST_OVERLAY_FILE, TEST_INPUT_FILE1,
'-map', '0', '-i',
'-map', '1', TEST_OVERLAY_FILE,
TEST_OUTPUT_FILE1 '-map',
'0',
'-map',
'1',
TEST_OUTPUT_FILE1,
] ]
@ -186,16 +211,18 @@ def test_filter_with_selector(use_shorthand):
a1 = i['a'].filter('aecho', 0.8, 0.9, 1000, 0.3) a1 = i['a'].filter('aecho', 0.8, 0.9, 1000, 0.3)
out = ffmpeg.output(a1, v1, TEST_OUTPUT_FILE1) out = ffmpeg.output(a1, v1, TEST_OUTPUT_FILE1)
assert out.get_args() == [ assert out.get_args() == [
'-i', TEST_INPUT_FILE1, '-i',
TEST_INPUT_FILE1,
'-filter_complex', '-filter_complex',
'[0:a]aecho=0.8:0.9:1000:0.3[s0];' \ '[0:a]aecho=0.8:0.9:1000:0.3[s0];' '[0:v]hflip[s1]',
'[0:v]hflip[s1]', '-map',
'-map', '[s0]', '-map', '[s1]', '[s0]',
TEST_OUTPUT_FILE1 '-map',
'[s1]',
TEST_OUTPUT_FILE1,
] ]
def test_get_item_with_bad_selectors(): def test_get_item_with_bad_selectors():
input = ffmpeg.input(TEST_INPUT_FILE1) input = ffmpeg.input(TEST_INPUT_FILE1)
@ -213,16 +240,12 @@ def test_get_item_with_bad_selectors():
def _get_complex_filter_asplit_example(): def _get_complex_filter_asplit_example():
split = (ffmpeg split = ffmpeg.input(TEST_INPUT_FILE1).vflip().asplit()
.input(TEST_INPUT_FILE1)
.vflip()
.asplit()
)
split0 = split[0] split0 = split[0]
split1 = split[1] split1 = split[1]
return (ffmpeg return (
.concat( ffmpeg.concat(
split0.filter('atrim', start=10, end=20), split0.filter('atrim', start=10, end=20),
split1.filter('atrim', start=30, end=40), split1.filter('atrim', start=30, end=40),
) )
@ -234,12 +257,7 @@ def _get_complex_filter_asplit_example():
def test_filter_concat__video_only(): def test_filter_concat__video_only():
in1 = ffmpeg.input('in1.mp4') in1 = ffmpeg.input('in1.mp4')
in2 = ffmpeg.input('in2.mp4') in2 = ffmpeg.input('in2.mp4')
args = ( args = ffmpeg.concat(in1, in2).output('out.mp4').get_args()
ffmpeg
.concat(in1, in2)
.output('out.mp4')
.get_args()
)
assert args == [ assert args == [
'-i', '-i',
'in1.mp4', 'in1.mp4',
@ -256,12 +274,7 @@ def test_filter_concat__video_only():
def test_filter_concat__audio_only(): def test_filter_concat__audio_only():
in1 = ffmpeg.input('in1.mp4') in1 = ffmpeg.input('in1.mp4')
in2 = ffmpeg.input('in2.mp4') in2 = ffmpeg.input('in2.mp4')
args = ( args = ffmpeg.concat(in1, in2, v=0, a=1).output('out.mp4').get_args()
ffmpeg
.concat(in1, in2, v=0, a=1)
.output('out.mp4')
.get_args()
)
assert args == [ assert args == [
'-i', '-i',
'in1.mp4', 'in1.mp4',
@ -271,7 +284,7 @@ def test_filter_concat__audio_only():
'[0][1]concat=a=1:n=2:v=0[s0]', '[0][1]concat=a=1:n=2:v=0[s0]',
'-map', '-map',
'[s0]', '[s0]',
'out.mp4' 'out.mp4',
] ]
@ -279,11 +292,7 @@ def test_filter_concat__audio_video():
in1 = ffmpeg.input('in1.mp4') in1 = ffmpeg.input('in1.mp4')
in2 = ffmpeg.input('in2.mp4') in2 = ffmpeg.input('in2.mp4')
joined = ffmpeg.concat(in1.video, in1.audio, in2.hflip(), in2['a'], v=1, a=1).node joined = ffmpeg.concat(in1.video, in1.audio, in2.hflip(), in2['a'], v=1, a=1).node
args = ( args = ffmpeg.output(joined[0], joined[1], 'out.mp4').get_args()
ffmpeg
.output(joined[0], joined[1], 'out.mp4')
.get_args()
)
assert args == [ assert args == [
'-i', '-i',
'in1.mp4', 'in1.mp4',
@ -304,8 +313,10 @@ def test_filter_concat__wrong_stream_count():
in2 = ffmpeg.input('in2.mp4') in2 = ffmpeg.input('in2.mp4')
with pytest.raises(ValueError) as excinfo: with pytest.raises(ValueError) as excinfo:
ffmpeg.concat(in1.video, in1.audio, in2.hflip(), v=1, a=1).node ffmpeg.concat(in1.video, in1.audio, in2.hflip(), v=1, a=1).node
assert str(excinfo.value) == \ assert (
'Expected concat input streams to have length multiple of 2 (v=1, a=1); got 3' str(excinfo.value)
== 'Expected concat input streams to have length multiple of 2 (v=1, a=1); got 3'
)
def test_filter_asplit(): def test_filter_asplit():
@ -320,14 +331,13 @@ def test_filter_asplit():
'-map', '-map',
'[s5]', '[s5]',
TEST_OUTPUT_FILE1, TEST_OUTPUT_FILE1,
'-y' '-y',
] ]
def test__output__bitrate(): def test__output__bitrate():
args = ( args = (
ffmpeg ffmpeg.input('in')
.input('in')
.output('out', video_bitrate=1000, audio_bitrate=200) .output('out', video_bitrate=1000, audio_bitrate=200)
.get_args() .get_args()
) )
@ -336,28 +346,26 @@ def test__output__bitrate():
@pytest.mark.parametrize('video_size', [(320, 240), '320x240']) @pytest.mark.parametrize('video_size', [(320, 240), '320x240'])
def test__output__video_size(video_size): def test__output__video_size(video_size):
args = ( args = ffmpeg.input('in').output('out', video_size=video_size).get_args()
ffmpeg
.input('in')
.output('out', video_size=video_size)
.get_args()
)
assert args == ['-i', 'in', '-video_size', '320x240', 'out'] assert args == ['-i', 'in', '-video_size', '320x240', 'out']
def test_filter_normal_arg_escape(): def test_filter_normal_arg_escape():
"""Test string escaping of normal filter args (e.g. ``font`` param of ``drawtext`` filter).""" """Test string escaping of normal filter args (e.g. ``font`` param of ``drawtext`` filter)."""
def _get_drawtext_font_repr(font): def _get_drawtext_font_repr(font):
"""Build a command-line arg using drawtext ``font`` param and extract the ``-filter_complex`` arg.""" """Build a command-line arg using drawtext ``font`` param and extract the ``-filter_complex`` arg."""
args = (ffmpeg args = (
.input('in') ffmpeg.input('in')
.drawtext('test', font='a{}b'.format(font)) .drawtext('test', font='a{}b'.format(font))
.output('out') .output('out')
.get_args() .get_args()
) )
assert args[:3] == ['-i', 'in', '-filter_complex'] assert args[:3] == ['-i', 'in', '-filter_complex']
assert args[4:] == ['-map', '[s0]', 'out'] assert args[4:] == ['-map', '[s0]', 'out']
match = re.match(r'\[0\]drawtext=font=a((.|\n)*)b:text=test\[s0\]', args[3], re.MULTILINE) 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]) assert match is not None, 'Invalid -filter_complex arg: {!r}'.format(args[3])
return match.group(1) return match.group(1)
@ -381,14 +389,10 @@ def test_filter_normal_arg_escape():
def test_filter_text_arg_str_escape(): def test_filter_text_arg_str_escape():
"""Test string escaping of normal filter args (e.g. ``text`` param of ``drawtext`` filter).""" """Test string escaping of normal filter args (e.g. ``text`` param of ``drawtext`` filter)."""
def _get_drawtext_text_repr(text): def _get_drawtext_text_repr(text):
"""Build a command-line arg using drawtext ``text`` param and extract the ``-filter_complex`` arg.""" """Build a command-line arg using drawtext ``text`` param and extract the ``-filter_complex`` arg."""
args = (ffmpeg args = ffmpeg.input('in').drawtext('a{}b'.format(text)).output('out').get_args()
.input('in')
.drawtext('a{}b'.format(text))
.output('out')
.get_args()
)
assert args[:3] == ['-i', 'in', '-filter_complex'] assert args[:3] == ['-i', 'in', '-filter_complex']
assert args[4:] == ['-map', '[s0]', 'out'] assert args[4:] == ['-map', '[s0]', 'out']
match = re.match(r'\[0\]drawtext=text=a((.|\n)*)b\[s0\]', args[3], re.MULTILINE) match = re.match(r'\[0\]drawtext=text=a((.|\n)*)b\[s0\]', args[3], re.MULTILINE)
@ -413,14 +417,19 @@ def test_filter_text_arg_str_escape():
assert expected == actual assert expected == actual
#def test_version(): # def test_version():
# subprocess.check_call(['ffmpeg', '-version']) # subprocess.check_call(['ffmpeg', '-version'])
def test__compile(): def test__compile():
out_file = ffmpeg.input('dummy.mp4').output('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() == ['ffmpeg', '-i', 'dummy.mp4', 'dummy2.mp4']
assert out_file.compile(cmd='ffmpeg.old') == ['ffmpeg.old', '-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_stdin', [True, False])
@ -431,15 +440,18 @@ def test__run_async(mocker, pipe_stdin, pipe_stdout, pipe_stderr):
popen__mock = mocker.patch.object(subprocess, 'Popen', return_value=process__mock) popen__mock = mocker.patch.object(subprocess, 'Popen', return_value=process__mock)
stream = _get_simple_example() stream = _get_simple_example()
process = ffmpeg.run_async( process = ffmpeg.run_async(
stream, pipe_stdin=pipe_stdin, pipe_stdout=pipe_stdout, pipe_stderr=pipe_stderr) stream, pipe_stdin=pipe_stdin, pipe_stdout=pipe_stdout, pipe_stderr=pipe_stderr
)
assert process is process__mock assert process is process__mock
expected_stdin = subprocess.PIPE if pipe_stdin else None expected_stdin = subprocess.PIPE if pipe_stdin else None
expected_stdout = subprocess.PIPE if pipe_stdout else None expected_stdout = subprocess.PIPE if pipe_stdout else None
expected_stderr = subprocess.PIPE if pipe_stderr else None expected_stderr = subprocess.PIPE if pipe_stderr else None
(args,), kwargs = popen__mock.call_args (args,), kwargs = popen__mock.call_args
assert args == ffmpeg.compile(stream) assert args == ffmpeg.compile(stream)
assert kwargs == dict(stdin=expected_stdin, stdout=expected_stdout, stderr=expected_stderr) assert kwargs == dict(
stdin=expected_stdin, stdout=expected_stdout, stderr=expected_stderr
)
def test__run(): def test__run():
@ -454,7 +466,9 @@ def test__run():
def test__run__capture_out(mocker, capture_stdout, capture_stderr): def test__run__capture_out(mocker, capture_stdout, capture_stderr):
mocker.patch.object(ffmpeg._run, 'compile', return_value=['echo', 'test']) mocker.patch.object(ffmpeg._run, 'compile', return_value=['echo', 'test'])
stream = _get_simple_example() stream = _get_simple_example()
out, err = ffmpeg.run(stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr) out, err = ffmpeg.run(
stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr
)
if capture_stdout: if capture_stdout:
assert out == 'test\n'.encode() assert out == 'test\n'.encode()
else: else:
@ -479,7 +493,9 @@ def test__run__error(mocker, capture_stdout, capture_stderr):
mocker.patch.object(ffmpeg._run, 'compile', return_value=['ffmpeg']) mocker.patch.object(ffmpeg._run, 'compile', return_value=['ffmpeg'])
stream = _get_complex_filter_example() stream = _get_complex_filter_example()
with pytest.raises(ffmpeg.Error) as excinfo: with pytest.raises(ffmpeg.Error) as excinfo:
out, err = ffmpeg.run(stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr) out, err = ffmpeg.run(
stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr
)
assert str(excinfo.value) == 'ffmpeg error (see stderr output for detail)' assert str(excinfo.value) == 'ffmpeg error (see stderr output for detail)'
out = excinfo.value.stdout out = excinfo.value.stdout
err = excinfo.value.stderr err = excinfo.value.stderr
@ -515,24 +531,30 @@ def test__filter__custom():
stream = ffmpeg.filter(stream, 'custom_filter', 'a', 'b', kwarg1='c') stream = ffmpeg.filter(stream, 'custom_filter', 'a', 'b', kwarg1='c')
stream = ffmpeg.output(stream, 'dummy2.mp4') stream = ffmpeg.output(stream, 'dummy2.mp4')
assert stream.get_args() == [ assert stream.get_args() == [
'-i', 'dummy.mp4', '-i',
'-filter_complex', '[0]custom_filter=a:b:kwarg1=c[s0]', 'dummy.mp4',
'-map', '[s0]', '-filter_complex',
'dummy2.mp4' '[0]custom_filter=a:b:kwarg1=c[s0]',
'-map',
'[s0]',
'dummy2.mp4',
] ]
def test__filter__custom_fluent(): def test__filter__custom_fluent():
stream = (ffmpeg stream = (
.input('dummy.mp4') ffmpeg.input('dummy.mp4')
.filter('custom_filter', 'a', 'b', kwarg1='c') .filter('custom_filter', 'a', 'b', kwarg1='c')
.output('dummy2.mp4') .output('dummy2.mp4')
) )
assert stream.get_args() == [ assert stream.get_args() == [
'-i', 'dummy.mp4', '-i',
'-filter_complex', '[0]custom_filter=a:b:kwarg1=c[s0]', 'dummy.mp4',
'-map', '[s0]', '-filter_complex',
'dummy2.mp4' '[0]custom_filter=a:b:kwarg1=c[s0]',
'-map',
'[s0]',
'dummy2.mp4',
] ]
@ -541,16 +563,29 @@ def test__merge_outputs():
out1 = in_.output('out1.mp4') out1 = in_.output('out1.mp4')
out2 = in_.output('out2.mp4') out2 = in_.output('out2.mp4')
assert ffmpeg.merge_outputs(out1, out2).get_args() == [ assert ffmpeg.merge_outputs(out1, out2).get_args() == [
'-i', 'in.mp4', 'out1.mp4', 'out2.mp4' '-i',
] 'in.mp4',
assert ffmpeg.get_args([out1, out2]) == [ 'out1.mp4',
'-i', 'in.mp4', 'out2.mp4', 'out1.mp4' 'out2.mp4',
] ]
assert ffmpeg.get_args([out1, out2]) == ['-i', 'in.mp4', 'out2.mp4', 'out1.mp4']
def test__input__start_time(): 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=10.5).output('out').get_args() == [
assert ffmpeg.input('in', ss=0.0).output('out').get_args() == ['-ss', '0.0', '-i', 'in', 'out'] '-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(): def test_multi_passthrough():
@ -558,49 +593,53 @@ def test_multi_passthrough():
out2 = ffmpeg.input('in2.mp4').output('out2.mp4') out2 = ffmpeg.input('in2.mp4').output('out2.mp4')
out = ffmpeg.merge_outputs(out1, out2) out = ffmpeg.merge_outputs(out1, out2)
assert ffmpeg.get_args(out) == [ assert ffmpeg.get_args(out) == [
'-i', 'in1.mp4', '-i',
'-i', 'in2.mp4', 'in1.mp4',
'-i',
'in2.mp4',
'out1.mp4', 'out1.mp4',
'-map', '1', '-map',
'out2.mp4' '1',
'out2.mp4',
] ]
assert ffmpeg.get_args([out1, out2]) == [ assert ffmpeg.get_args([out1, out2]) == [
'-i', 'in2.mp4', '-i',
'-i', 'in1.mp4', 'in2.mp4',
'-i',
'in1.mp4',
'out2.mp4', 'out2.mp4',
'-map', '1', '-map',
'out1.mp4' '1',
'out1.mp4',
] ]
def test_passthrough_selectors(): def test_passthrough_selectors():
i1 = ffmpeg.input(TEST_INPUT_FILE1) i1 = ffmpeg.input(TEST_INPUT_FILE1)
args = ( args = ffmpeg.output(i1['1'], i1['2'], TEST_OUTPUT_FILE1).get_args()
ffmpeg
.output(i1['1'], i1['2'], TEST_OUTPUT_FILE1)
.get_args()
)
assert args == [ assert args == [
'-i', TEST_INPUT_FILE1, '-i',
'-map', '0:1', TEST_INPUT_FILE1,
'-map', '0:2', '-map',
'0:1',
'-map',
'0:2',
TEST_OUTPUT_FILE1, TEST_OUTPUT_FILE1,
] ]
def test_mixed_passthrough_selectors(): def test_mixed_passthrough_selectors():
i1 = ffmpeg.input(TEST_INPUT_FILE1) i1 = ffmpeg.input(TEST_INPUT_FILE1)
args = ( args = ffmpeg.output(i1['1'].hflip(), i1['2'], TEST_OUTPUT_FILE1).get_args()
ffmpeg
.output(i1['1'].hflip(), i1['2'], TEST_OUTPUT_FILE1)
.get_args()
)
assert args == [ assert args == [
'-i', TEST_INPUT_FILE1, '-i',
TEST_INPUT_FILE1,
'-filter_complex', '-filter_complex',
'[0:1]hflip[s0]', '[0:1]hflip[s0]',
'-map', '[s0]', '-map',
'-map', '0:2', '[s0]',
'-map',
'0:2',
TEST_OUTPUT_FILE1, TEST_OUTPUT_FILE1,
] ]
@ -612,36 +651,53 @@ def test_pipe():
frame_count = 10 frame_count = 10
start_frame = 2 start_frame = 2
out = (ffmpeg out = (
.input('pipe:0', format='rawvideo', pixel_format='rgb24', video_size=(width, height), framerate=10) ffmpeg.input(
'pipe:0',
format='rawvideo',
pixel_format='rgb24',
video_size=(width, height),
framerate=10,
)
.trim(start_frame=start_frame) .trim(start_frame=start_frame)
.output('pipe:1', format='rawvideo') .output('pipe:1', format='rawvideo')
) )
args = out.get_args() args = out.get_args()
assert args == [ assert args == [
'-f', 'rawvideo', '-f',
'-video_size', '{}x{}'.format(width, height), 'rawvideo',
'-framerate', '10', '-video_size',
'-pixel_format', 'rgb24', '{}x{}'.format(width, height),
'-i', 'pipe:0', '-framerate',
'10',
'-pixel_format',
'rgb24',
'-i',
'pipe:0',
'-filter_complex', '-filter_complex',
'[0]trim=start_frame=2[s0]', '[0]trim=start_frame=2[s0]',
'-map', '[s0]', '-map',
'-f', 'rawvideo', '[s0]',
'pipe:1' '-f',
'rawvideo',
'pipe:1',
] ]
cmd = ['ffmpeg'] + args cmd = ['ffmpeg'] + args
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 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)])) 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.write(in_data) # note: this could block, in which case need to use threads
p.stdin.close() p.stdin.close()
out_data = p.stdout.read() out_data = p.stdout.read()
assert len(out_data) == frame_size * (frame_count - start_frame) assert len(out_data) == frame_size * (frame_count - start_frame)
assert out_data == in_data[start_frame*frame_size:] assert out_data == in_data[start_frame * frame_size :]
def test__probe(): def test__probe():

1
requirements.txt
View File

@ -13,6 +13,7 @@ importlib-metadata==0.17
Jinja2==2.10.1 Jinja2==2.10.1
MarkupSafe==1.1.1 MarkupSafe==1.1.1
more-itertools==7.0.0 more-itertools==7.0.0
numpy==1.16.4
packaging==19.0 packaging==19.0
pluggy==0.12.0 pluggy==0.12.0
py==1.8.0 py==1.8.0

15
setup.py
View File

@ -2,22 +2,26 @@ from setuptools import setup
from textwrap import dedent from textwrap import dedent
version = '0.1.18' version = '0.1.18'
download_url = 'https://github.com/kkroening/ffmpeg-python/archive/v{}.zip'.format(version) download_url = 'https://github.com/kkroening/ffmpeg-python/archive/v{}.zip'.format(
version
)
long_description = dedent("""\ long_description = dedent(
'''\
ffmpeg-python: Python bindings for FFmpeg ffmpeg-python: Python bindings for FFmpeg
========================================= =========================================
:Github: https://github.com/kkroening/ffmpeg-python :Github: https://github.com/kkroening/ffmpeg-python
:API Reference: https://kkroening.github.io/ffmpeg-python/ :API Reference: https://kkroening.github.io/ffmpeg-python/
""") '''
)
file_formats = [ file_formats = [
'aac', 'aac',
'ac3', 'ac3',
'avi', 'avi',
'bmp' 'bmp',
'flac', 'flac',
'gif', 'gif',
'mov', 'mov',
@ -70,11 +74,12 @@ setup(
extras_require={ extras_require={
'dev': [ 'dev': [
'future==0.17.1', 'future==0.17.1',
'numpy==1.16.4',
'pytest-mock==1.10.4', 'pytest-mock==1.10.4',
'pytest==4.6.1', 'pytest==4.6.1',
'Sphinx==2.1.0', 'Sphinx==2.1.0',
'tox==3.12.1', 'tox==3.12.1',
], ]
}, },
classifiers=[ classifiers=[
'Intended Audience :: Developers', 'Intended Audience :: Developers',