-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathlog.py
More file actions
158 lines (126 loc) · 4.7 KB
/
log.py
File metadata and controls
158 lines (126 loc) · 4.7 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
149
150
151
152
153
154
155
156
157
158
import re
from fractions import Fraction
from .._typing import Sequence
from ..caps import sample_fmts
from . import layout_to_channels
_re_audio = re.compile(r"(?:(\d+) Hz, )?(.+)")
def parse_log_audio_stream(info):
"""parse audio codec info on FFmpeg log
:param info: stream info string after basic codec info
:type info: str
:return: dict containing stream info with keys matching FFmpeg options
:rtype: dict
Output Key Type Description
------------ ---- -------------------------------------
'ar' int Sampling rate in samples/second
'ac' int Number of audio channels
'sample_fmt' str Audio sample format
"""
mm = _re_audio.match(info)
d = {}
if mm[1]:
d["ar"] = int(mm[1])
items = mm[2].split(", ")
i = 1
if len(items):
try:
d["ac"] = layout_to_channels(items[0])
except:
try:
d["ac"] = int(re.match(r"(\d+) channels", items[0])[1])
except:
i = 0
sample_fmt = items[i] if len(items) > i else None
if sample_fmt is not None:
sample_fmt = sample_fmt.split(" ", 1)[0]
if sample_fmt in sample_fmts():
d["sample_fmt"] = sample_fmt
return d
# none|pix_fmt[([%bits_per_raw_sample% bpc, ][%color_range%, ]
# [%color_space_name%/%color_primaries_name%/%color_transfer_name%, |%color_primaries_name%, ]
# [%field_order%, ][%chroma_sample_location%, ])]
# [[(??|, )%width%x%height%[ (coded_widthxcoded_height)][ \[SAR %d:%d DAR %d:%d\]][, %d/%d]]]
# [, q=%d-%d]|[, Closed Captions][, Film Grain][, lossless]
_re_video_codec = re.compile(
r"(none)|([a-z0-9]+)(?:\(.+?\))?(?:, ([0-9]+)x([0-9]+)(?: [0-9]+x[0-9]+)?(?:, \[SAR [0-9]+:[0-9]+ DAR ([0-9]+):([0-9]+)\])?)?"
)
# [(avg_frame_rate) fps, (r_frame_rate) tbr, (1/time_base) tbn]
_re_video_fps = re.compile(r"([0-9.]+)(k)? fps|([0-9.]+)(k)? tbr")
def parse_log_video_stream(info):
"""parse video codec info in FFmpeg log
:param info: stream info string after basic codec info
:type info: str
:return: dict containing stream info with keys matching FFmpeg options
:rtype: dict
Output Key Type Description
------------ ---- -------------------------------------
'r' int Sampling rate in samples/second
'pix_fmt' str Pixel format
's' [int, int] width and height
'aspect' Fraction display aspect ratio
"""
d = {}
mm = _re_video_codec.match(info)
if mm[2]:
d["pix_fmt"] = mm[2]
if mm[3]:
d["s"] = [int(mm[i]) for i in range(3, 5)]
if mm[5]:
d["aspect"] = Fraction(*[int(mm[i]) for i in range(5, 7)])
mm = _re_video_fps.search(info, mm.end())
r = None
if mm[1]:
r = float(mm[1])
if mm[2]:
r *= 1e3
elif mm[3]:
r = float(mm[3])
if mm[4]:
r *= 1e3
if r is not None:
ri = int(r)
d["r"] = ri if ri == r else Fraction(round(r * 1001), 1001)
# [TODO] x1001 is a hack job, revisit
return d
_re_stream = re.compile(
r" Stream #\d+:\d+(?:\[.+?\])?(?:\(.+?\))?: (Audio|Video): ([^, ]+)(?: \(.+?\))*, (.*)"
)
def extract_output_stream(
logs: str | Sequence[str],
file_id: int = 0,
stream_id: int = 0,
hint: int | None = None,
) -> dict:
"""extract output stream info from the log lines
:param logs: lines of FFmpeg log messages
:param file_id: output file id, defaults to 0
:param stream_id: output stream id, defaults to 0
:param hint: starting log line index to search, defaults to None
:return: stream information
"""
if isinstance(logs, str):
logs = re.split(r"[\n\r]+", logs)
fname = f"Output #{file_id}"
sname = f" Stream #{file_id}:{stream_id}"
if hint:
logs = logs[hint:]
i0 = next((i for i, l in enumerate(logs) if l.startswith(fname)), None)
if i0 is None:
raise ValueError("output log is not present")
log = next((l for l in logs[i0:] if l.startswith(sname)), None)
if log is None:
raise ValueError("output stream log is not present")
# https://github.com/FFmpeg/FFmpeg/blob/6af21de373c979bc2087717acb61e834768ebe4b/libavformat/dump.c#L621
# https://github.com/FFmpeg/FFmpeg/blob/cd03a180cb66ca199707ad129a4ab44548711c94/libavcodec/avcodec.c#L519
sinfo = {}
m = _re_stream.match(log)
type = m[1]
sinfo = {"codec": m[2], "type": type.lower()}
info = m[3]
if type == "Audio":
sinfo = {**sinfo, **parse_log_audio_stream(info)}
elif type == "Video":
sinfo = {**sinfo, **parse_log_video_stream(info)}
else:
raise RuntimeError(f"parser for {type.lower()} codec is not defined.")
return sinfo