-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathstd_runners.py
More file actions
148 lines (124 loc) · 4.22 KB
/
std_runners.py
File metadata and controls
148 lines (124 loc) · 4.22 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
"""FFmpeg runner functions for SISO operations over standard pipes"""
from __future__ import annotations
import logging
from . import configure
from . import ffmpegprocess as fp
from ._typing import (
TYPE_CHECKING,
Any,
EncodedInputInfoDict,
EncodedOutputInfoDict,
ProgressCallable,
RawInputInfoDict,
RawOutputInfoDict,
)
from .errors import FFmpegError, FFmpegioError
if TYPE_CHECKING:
from .configure import FFmpegArgs
logger = logging.getLogger("ffmpegio")
__all__ = ["run_and_return_raw", "run_and_return_encoded"]
def run_and_return_raw(
args: FFmpegArgs,
input_info: list[RawInputInfoDict | EncodedInputInfoDict],
output_info: list[RawOutputInfoDict | EncodedOutputInfoDict],
progress: ProgressCallable | None,
show_log: bool | None,
sp_kwargs: dict[str, Any] | None,
):
# check configuration yields at most one piped input
# check configuration yields at most one piped output
n_piped_inputs = sum(
info["src_type"] in ("buffer", "fileobj") for info in input_info
)
if n_piped_inputs > 1:
raise ValueError(
"Only at most one input source can be a pipe or a file-stream object."
)
# check configuration yields exactly one piped audio output
if len(output_info) == 0:
raise FFmpegioError("No audio stream found.")
if len(output_info) > 1:
raise ValueError("Too many audio stream found.")
if output_info[0]["dst_type"] != "buffer":
raise ValueError("Not outputting to pipe")
n_piped_outputs = sum(
info["dst_type"] in ("buffer", "fileobj") for info in output_info
)
if n_piped_outputs > 1:
raise ValueError(
"Only at most one output destination can be a pipe or a file-stream object."
)
# assign the stdin and stdout pipes
kwargs = {
**configure.assign_input_pipes(args, input_info, True, True)[1],
**configure.assign_output_pipes(args, output_info, True)[1],
}
if sp_kwargs is not None:
# ignore user's stdin, stdout, stdout if specified
kwargs = {**sp_kwargs, **kwargs}
out = fp.run(
args,
progress=progress,
capture_log=None if show_log else True,
**kwargs,
)
if out.returncode:
raise FFmpegError(out.stderr, show_log)
oinfo = output_info[0]
dtype, shape, rate = oinfo["raw_info"]
return rate, oinfo["bytes2data"](
b=out.stdout, dtype=dtype, shape=shape, squeeze=oinfo["squeeze"]
)
def run_and_return_encoded(
progress,
overwrite,
show_log,
sp_kwargs,
args,
input_info,
output_info,
two_pass=False,
pass1_omits=None,
pass1_extras=None,
):
if output_info is None:
raise FFmpegioError("Unknown error occurred to complete FFmpeg configuration.")
# check configuration yields at most one piped output
n_piped_inputs = sum(
info["src_type"] in ("buffer", "fileobj") for info in input_info
)
if n_piped_inputs > 1:
raise ValueError(
"Only at most one input source can be a pipe or a file-stream object."
)
n_piped_outputs = sum(
info["dst_type"] in ("buffer", "fileobj") for info in output_info
)
if n_piped_outputs > 1:
raise ValueError(
"Only at most one output destination can be a pipe or a file-stream object."
)
# assign the stdin and stdout pipes
kwargs = {
**configure.assign_input_pipes(args, input_info, True, True)[1],
**configure.assign_output_pipes(args, output_info, True)[1],
}
if sp_kwargs is not None:
# ignore user's stdin, stdout, stdout if specified
kwargs = {**sp_kwargs, **kwargs}
if two_pass:
if pass1_omits is not None:
kwargs["pass1_omits"] = pass1_omits
if pass1_extras is not None:
kwargs["pass1_extras"] = pass1_extras
out = (fp.run_two_pass if two_pass else fp.run)(
args,
progress=progress,
capture_log=None if show_log else True,
overwrite=overwrite,
**kwargs,
)
if out.returncode:
raise FFmpegError(out.stderr, show_log)
if n_piped_outputs and any(info["dst_type"] == "buffer" for info in output_info):
return out.stdout