import json
import subprocess
from ._run import Error
from ._utils import convert_kwargs_to_cmd_line_args
def probe(filename, cmd='ffprobe', **kwargs):
"""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]
# Original: p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# Popen add args: shell=True, stdin=subprocess.PIPE,
p = subprocess.Popen(args, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = p.communicate()
if p.returncode != 0:
raise Error('ffprobe', out, err)
return json.loads(out.decode('utf-8'))
__all__ = ['probe']
然后,进入_run.py文件。你需要添加shell=True,修改stdin=subprocess.PIPE或修改pipe_stdin=True(下面的代码部分仅为代码的一部分):
@output_operator()
def run_async(
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
# Original: return subprocess.Popen(
# args, stdin=pipe_stdin, stdout=stdout_stream, stderr=stderr_stream)
# Add shell=True, modify stdin=subprocess.PIPE or modify pipe_stdin=True
return subprocess.Popen(
args, shell=True, stdin=subprocess.PIPE, stdout=stdout_stream, stderr=stderr_stream
)
将日志级别设置为安静
ffmpeg.input(file).output(filename, loglevel="quiet").run()
添加 "from subprocess import CREATE_NO_WINDOW" 并使用 "creationflags=CREATE_NO_WINDOW" 用于 Popen。下面是 ffmpeg-python 库中 "_run.py" 代码的更新部分,对我有效。
from subprocess import CREATE_NO_WINDOW
@output_operator()
def run_async(
stream_spec,
cmd='ffmpeg',
pipe_stdin=False,
pipe_stdout=False,
pipe_stderr=False,
quiet=False,
overwrite_output=False,
):
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=subprocess.PIPE, stdout=stdout_stream, stderr=stderr_stream, creationflags=CREATE_NO_WINDOW
)
Bradley的答案可以解决使用pyinstaller编译后控制台闪烁的问题。然而,我不太舒服直接更新ffmpeg-python
库本身,因为当PIP有更新时它会被覆盖,而且总体上感觉有点hacky。
最终,我选择劫持函数并在我的类中直接使用它们,这也解决了问题。我认为这样更安全,但如果库以与劫持函数冲突的方式进行更新,则仍存在风险。
"""Run OS command
Function to merge video and
subtitle file(s) into an MKV
"""
def run_os_command(self, os_command):
subprocess.call(os_command, shell=True)
"""FFmpeg probe hi-jack
Customized arguments to Popen to
prevent console flashes after
compiled with PyInstaller
"""
def ffmpeg_probe(self, video_input_path):
command = ['ffprobe', '-show_format', '-show_streams', '-of', 'json']
command += [video_input_path]
process = subprocess.Popen(
command,
shell=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
out, err = process.communicate()
if process.returncode != 0:
raise Exception(f"ffprobe error: {err}")
return json.loads(out.decode('utf-8'))
"""FFmpeg run hi-jack
Uses argument compiler from
library but alternate sub-
process method to run command
to prevent console flashes.
"""
def ffmpeg_run(self, stream):
os_command = ffmpeg.compile(stream, 'ffmpeg', overwrite_output=True)
return self.run_os_command(os_command)
然后使用
probe = ffmpeg_probe(video_input_path) # use like ffmpeg.probe()
ffmpeg_run(stream) # use like ffmpeg.run() can update the function if you pass more than stream