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 #283 from magnusvmt/master

Browse Source 
Add optional timeout argument to probe
This commit is contained in:
Karl Kroening 2019-12-30 02:23:02 -06:00 committed by GitHub
commit 4cb7d26f55
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 13 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
ffmpeg/_probe.py
View File

@ -4,7 +4,7 @@ from ._run import Error
from ._utils import convert_kwargs_to_cmd_line_args from ._utils import convert_kwargs_to_cmd_line_args
def probe(filename, cmd='ffprobe', **kwargs): 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: Raises:
@ -18,7 +18,10 @@ def probe(filename, cmd='ffprobe', **kwargs):
args += [filename] args += [filename]
p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = p.communicate() communicate_kwargs = {}
if timeout is not None:
communicate_kwargs['timeout'] = timeout
out, err = p.communicate(**communicate_kwargs)
if p.returncode != 0: if p.returncode != 0:
raise Error('ffprobe', out, err) raise Error('ffprobe', out, err)
return json.loads(out.decode('utf-8')) return json.loads(out.decode('utf-8'))

8
ffmpeg/tests/test_ffmpeg.py
View File

@ -8,6 +8,7 @@ import pytest
import random import random
import re import re
import subprocess import subprocess
import sys
try: try:
@ -711,6 +712,13 @@ def test__probe():
assert data['format']['duration'] == '7.036000' 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)
def test__probe__exception(): def test__probe__exception():
with pytest.raises(ffmpeg.Error) as excinfo: with pytest.raises(ffmpeg.Error) as excinfo:
ffmpeg.probe(BOGUS_INPUT_FILE) ffmpeg.probe(BOGUS_INPUT_FILE)