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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203 | def mg_info(self, type=None, autoshow=True, overwrite=True):
"""
Returns info about video/audio/format file using ffprobe.
Args:
type (str, optional): Type of information to retrieve. Possible choices are 'summary', 'audio', 'video', 'format' or 'frame'. Defaults to None (which gives info about video, audio and format).
- 'summary': prints a human-readable table of key video properties (resolution, fps, frame count, duration, color mode, audio) and returns a dict.
- 'audio' / 'video' / 'format': returns the matching ffprobe stream as a pandas DataFrame row.
- 'frame': renders a bar chart of I/P/B frame sizes and returns a DataFrame.
- None: returns a DataFrame with all ffprobe stream and format metadata.
autoshow (bool, optional): Whether to show the I/P/B frames figure automatically. Defaults to True. NB: The type argument needs to be set to 'frame'.
overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.
Returns:
dict or pandas.DataFrame: dict when type='summary', DataFrame otherwise.
"""
if type == 'summary':
framecount = get_framecount(self.filename)
# NB: for an MgVideo `self.length` is the frame *count*, not seconds
# (for an MgAudio it is the duration in seconds and there is no fps).
fps = getattr(self, 'fps', None)
duration_secs = self.length / fps if fps else float(self.length)
h = int(duration_secs // 3600)
m = int((duration_secs % 3600) // 60)
s = duration_secs % 60
duration_str = f"{h}:{m:02d}:{s:05.2f}" if h else f"{m}:{s:05.2f}"
filesize = os.path.getsize(self.filename)
if filesize >= 1_000_000:
size_str = f"{filesize / 1_000_000:.1f} MB"
elif filesize >= 1_000:
size_str = f"{filesize / 1_000:.1f} KB"
else:
size_str = f"{filesize} B"
# Query codec/profile details from ffprobe
v = _probe_stream(self.filename, 'v')
a = _probe_stream(self.filename, 'a')
video_codec = v.get('codec_name')
video_profile = v.get('profile')
pix_fmt = v.get('pix_fmt')
color_space = v.get('color_space')
color_profile = ', '.join(x for x in (pix_fmt, color_space) if x and x != 'unknown') or None
audio_codec = a.get('codec_name')
audio_sr = a.get('sample_rate')
audio_br = a.get('bit_rate')
audio_sr_str = f"{int(audio_sr):,} Hz" if audio_sr and audio_sr.isdigit() else None
audio_br_str = f"{int(audio_br) // 1000} kbps" if audio_br and audio_br.isdigit() else None
info_dict = {
'filename': os.path.basename(self.filename),
'width': self.width,
'height': self.height,
'fps': self.fps,
'frames': framecount,
'duration': round(duration_secs, 3),
'color': self.color,
'video_codec': video_codec,
'video_profile': video_profile,
'pixel_format': pix_fmt,
'color_space': color_space,
'has_audio': bool(self.has_audio),
'audio_codec': audio_codec,
'audio_sample_rate': int(audio_sr) if audio_sr and audio_sr.isdigit() else None,
'audio_bit_rate': int(audio_br) if audio_br and audio_br.isdigit() else None,
'filesize': filesize,
}
col = 14
print(f"{'File:':<{col}} {os.path.basename(self.filename)}")
print(f"{'Resolution:':<{col}} {self.width} × {self.height} px")
print(f"{'Frames:':<{col}} {framecount} @ {self.fps:g} fps")
print(f"{'Duration:':<{col}} {duration_str} ({duration_secs:.3f} s)")
print(f"{'Color:':<{col}} {'color' if self.color else 'grayscale'}")
codec_str = video_codec or 'unknown'
if video_profile:
codec_str += f" ({video_profile})"
print(f"{'Video codec:':<{col}} {codec_str}")
if color_profile:
print(f"{'Color profile:':<{col}} {color_profile}")
if self.has_audio:
audio_str = audio_codec or 'unknown'
extras = ', '.join(x for x in (audio_sr_str, audio_br_str) if x)
if extras:
audio_str += f" ({extras})"
print(f"{'Audio:':<{col}} {audio_str}")
else:
print(f"{'Audio:':<{col}} no")
print(f"{'File size:':<{col}} {size_str}")
return info_dict
# Get streams and format information (https://ffmpeg.org/ffprobe.html)
cmd = ["ffprobe", "-hide_banner", "-loglevel", "quiet", "-show_streams", "-show_format", self.filename]
if type == 'frame':
if self.fex != '.mp4':
# Convert video file to mp4
self.filename = convert_to_mp4(self.of + self.fex, overwrite=overwrite)
self.of, self.fex = os.path.splitext(self.filename)
cmd = ["ffprobe", "-hide_banner", "-loglevel", "quiet", "-v", "error", "-select_streams", "v:0", "-show_entries", "frame=pkt_size, pict_type", self.filename]
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
try:
out, _ = process.communicate(timeout=10)
splitted = out.split('\n')
except subprocess.TimeoutExpired:
process.kill()
out, err = process.communicate()
splitted = out.split('\n')
frame = []
# Retrieve information and export it in a dictionary
if type == 'frame':
current_frame = {}
for line in [i for i in splitted if i not in ('[SIDE_DATA]', '[/SIDE_DATA]', '')]:
if line == '[/FRAME]':
frame.append(current_frame)
current_frame = {}
elif line != '[FRAME]':
pair = line.split('=')
current_frame[pair[0]] = pair[1]
else:
pass
ipb_frames = {
'frame index': range(len(frame)),
'size (bytes)': [int(f['pkt_size']) for f in frame],
'type': [f['pict_type'] for f in frame]
}
df = pd.DataFrame.from_dict(ipb_frames)
if not autoshow:
return df
fig, ax = plt.subplots(figsize=(12,4), dpi=300)
fig.patch.set_facecolor('white') # make sure background is white
fig.patch.set_alpha(1)
for i, (label, series) in enumerate(df.groupby('type')):
plot_frames(series, label, index=i)
# Get handles and labels
handles, labels = plt.gca().get_legend_handles_labels()
order = [1,2,0] # specify order of items in legend
# Add legend to plot
ax.legend([handles[idx] for idx in order],[labels[idx] for idx in order])
ax.set_xlabel('Frame index')
ax.set_ylabel('Size (bytes)')
fig.tight_layout()
# Save and close so display goes through the returned MgImage's show()
target_png = self.of + '_frames.png'
if not overwrite:
target_png = generate_outfilename(target_png)
fig.savefig(target_png, facecolor='white')
plt.close(fig)
return MgImage(target_png)
else:
for i, info in enumerate(splitted):
if info == "[STREAM]" or info == "[SIDE_DATA]" or info == "[FORMAT]":
frame.append(dict())
i +=1
elif info == "[/STREAM]" or info == "[/SIDE_DATA]" or info == "[/FORMAT]" or info == "":
i +=1
else:
try:
key, value = splitted[i].split('=')
frame[-1][key] = value
except ValueError:
key = splitted[i]
frame[-1][key] = ''
if len(frame) > 3:
# Merge video stream with side data dictionary
frame[0] = {**frame[0], **frame[1]}
frame.pop(1)
# Create a pandas dataframe
df = pd.DataFrame.from_dict(frame)
df.insert(0, 'codec_type', df.pop('codec_type')) # move codec type column
df.pop('index') # remove index column
df = df[df.codec_type.notna()] # remove rows with nan values in codec_type column
if type is not None:
return df[df.codec_type == type]
else:
return df
|