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

Minor improvements (formatting, etc)

Browse Source
This commit is contained in:
Karl Kroening 2018-03-11 20:02:56 -07:00
parent 809ab6cd17
commit 6169b89321
4 changed files with 52 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
ffmpeg/_ffmpeg.py
View File

@ -11,8 +11,6 @@ from .nodes import (
output_operator,
OutputStream)
_py_map = map
def input(filename, **kwargs):
"""Input file URL (ffmpeg ``-i`` option)
@ -56,9 +54,9 @@ def output(*streams_and_filename, **kwargs):
Official documentation: `Synopsis <https://ffmpeg.org/ffmpeg.html#Synopsis>`__
"""
streams_and_filename = list(streams_and_filename)
if "filename" not in kwargs:
if 'filename' not in kwargs:
if not isinstance(streams_and_filename[-1], basestring):
raise ValueError("You must provide a filename")
raise ValueError('A filename must be provided')
kwargs['filename'] = streams_and_filename.pop(-1)
streams = streams_and_filename
@ -78,7 +76,7 @@ def map(*streams):
tail = streams[1:]
if not isinstance(head, OutputStream):
raise ValueError("First argument must be an output stream")
raise ValueError('First argument must be an output stream')
if not tail:
return head

10
ffmpeg/_run.py
View File

@ -53,14 +53,14 @@ def _get_input_args(input_node):
def _format_input_stream_name(stream_name_map, edge):
prefix = stream_name_map[edge.upstream_node, edge.upstream_label]
if not edge.upstream_selector:
suffix = ""
suffix = ''
else:
suffix = ":{}".format(edge.upstream_selector)
return "[{}{}]".format(prefix, suffix)
suffix = ':{}'.format(edge.upstream_selector)
return '[{}{}]'.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):
@ -104,7 +104,7 @@ def _get_output_args(node, stream_name_map):
args = []
if len(node.incoming_edges) == 0:
raise ValueError("Output node {} has no mapped streams")
raise ValueError('Output node {} has no mapped streams'.format(node))
for edge in node.incoming_edges:
# edge = node.incoming_edges[0]

26
ffmpeg/nodes.py
View File

@ -46,22 +46,22 @@ class Stream(object):
def __repr__(self):
node_repr = self.node.long_repr(include_hash=False)
selector = ""
selector = ''
if self.selector:
selector = ":{}".format(self.selector)
selector = ':{}'.format(self.selector)
out = '{}[{!r}{}] <{}>'.format(node_repr, self.label, selector, self.node.short_hash)
return out
def __getitem__(self, item):
def __getitem__(self, index):
"""
Select a component of the stream. `stream[:X]` is analogous to `stream.node.stream(select=X)`.
Please note that this can only be used to select a substream that already exist. If you want to split
the stream, use the `split` filter.
"""
if not isinstance(item, slice) or item.start is not None:
raise ValueError("Invalid syntax. Use 'stream[:\"something\"]', not 'stream[\"something\"]'.")
if not isinstance(index, slice) or index.start is not None:
raise ValueError("Invalid syntax. Use `stream[:\'something\']`, not `stream[\'something\']`.")
return self.node.stream(select=item.stop)
return self.node.stream(select=index.stop)
def get_stream_map(stream_spec):
@ -147,7 +147,7 @@ class Node(KwargReprNode):
def __init_fromnode__(self, old_node, stream_spec):
# Make sure old node and new node are of the same type
if type(self) != type(old_node):
raise ValueError("'old_node' should be of type {}".format(self.__class__.__name__))
raise TypeError('`old_node` should be of type {}'.format(self.__class__.__name__))
# Copy needed data from old node
name = old_node.name
@ -197,8 +197,8 @@ class Node(KwargReprNode):
argc = 1 + len(args) + len(kwargs)
first_arg = None
if "old_node" in kwargs:
first_arg = kwargs["old_node"]
if 'old_node' in kwargs:
first_arg = kwargs['old_node']
elif len(args) > 0:
first_arg = args[0]
@ -207,8 +207,8 @@ class Node(KwargReprNode):
else:
if isinstance(first_arg, Node):
raise ValueError(
"{}.__init__() received an instance of {} as the first argument. If you want to create a "
"copy of an existing node, the types must match and you must provide an additional stream_spec."
'{}.__init__() received an instance of {} as the first argument. If you want to create a '
'copy of an existing node, the types must match and you must provide an additional stream_spec.'
.format(self.__class__.__name__, first_arg.__class__.__name__)
)
self.__init_fromscratch__(*args, **kwargs)
@ -222,8 +222,8 @@ class Node(KwargReprNode):
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:"audio"] returns a stream with label 0 and
selector "audio", which is the same as ``node.stream(label=0, select="audio")``.
It can also be used to apply a selector: e.g. ``node[0:'audio']`` returns a stream with label 0 and
selector ``'audio'``, which is the same as ``node.stream(label=0, select='audio')``.
"""
if isinstance(item, slice):
return self.stream(label=item.start, select=item.stop)

56
ffmpeg/tests/test_ffmpeg.py
View File

@ -82,21 +82,21 @@ def test_node_repr():
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)
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)
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)
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)
assert repr(dummy_out) == 'dummy()[{!r}] <{}>'.format(dummy_out.label, dummy_out.node.short_hash)
def test_get_args_simple():
@ -146,21 +146,25 @@ def test_get_args_complex_filter():
'-y'
]
def _get_filter_with_select_example():
i = ffmpeg.input(TEST_INPUT_FILE1)
v1 = i[:"v"].hflip()
a1 = i[:"a"].filter_("aecho", 0.8, 0.9, 1000, 0.3)
v1 = i[:'v'].hflip()
a1 = i[:'a'].filter_('aecho', 0.8, 0.9, 1000, 0.3)
return ffmpeg.output(a1, v1, TEST_OUTPUT_FILE1)
def test_filter_with_select():
assert _get_filter_with_select_example().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]
assert _get_filter_with_select_example().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_map_same_effect_as_output():
i1 = ffmpeg.input(TEST_INPUT_FILE1)
@ -173,11 +177,13 @@ def test_map_same_effect_as_output():
assert id(o_map) != id(_o_map) # Checks immutability
assert o_map.node.incoming_edge_map == o_nomap.node.incoming_edge_map
assert o_map.get_args() == o_nomap.get_args() == ['-i', TEST_INPUT_FILE1,
'-i', TEST_OVERLAY_FILE,
'-map', '[0]',
'-map', '[1]',
TEST_OUTPUT_FILE1]
assert o_map.get_args() == o_nomap.get_args() == [
'-i', TEST_INPUT_FILE1,
'-i', TEST_OVERLAY_FILE,
'-map', '[0]',
'-map', '[1]',
TEST_OUTPUT_FILE1
]
def _get_complex_filter_asplit_example():
@ -191,8 +197,8 @@ def _get_complex_filter_asplit_example():
return (ffmpeg
.concat(
split0.filter_("atrim", start=10, end=20),
split1.filter_("atrim", start=30, end=40),
split0.filter_('atrim', start=10, end=20),
split1.filter_('atrim', start=30, end=40),
)
.output(TEST_OUTPUT_FILE1)
.overwrite_output()