-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathtest_media.py
More file actions
104 lines (79 loc) · 3.01 KB
/
Copy pathtest_media.py
File metadata and controls
104 lines (79 loc) · 3.01 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
from os import path
from pprint import pprint
from tempfile import TemporaryDirectory
import pytest
import ffmpegio as ff
url = "tests/assets/testmulti-1m.mp4"
url1 = "tests/assets/testvideo-1m.mp4"
url2 = "tests/assets/testaudio-1m.mp3"
@pytest.mark.parametrize(
"urls,kwargs,nout",
[
((url,), dict(t=1, show_log=True), 4),
((url,), dict(streams=("v:0", "v:1", "a:1", "a:0"), t=1, show_log=True), 4),
((url1, url2), dict(t=1, show_log=True), 2),
((url1,), dict(t=1, filter_complex="[0:0]split[out1][out2]", show_log=True), 2),
],
)
def test_media_read(urls, kwargs, nout):
rates, data = ff.media.read(*urls, **kwargs, timeout=1.0)
assert len(rates) == nout
print(rates)
print([(k, x["shape"], x["dtype"]) for k, x in data.items()])
def test_media_read_filter_complex():
urls = (url2, url) # aud + mul
kwargs = dict(
t=1,
show_log=True,
filter_complex="[0:a]aformat=f=dbl:r=8000:cl=mono;[1:v:1]setpts=0.5*PTS",
)
# kwargs = dict(map=(['[vout]','[aout]']), t=1, show_log=True, filter_complex='[0:a]aformat=f=dbl:r=8000:cl=mono[aout];[1:v:1]setpts=0.5*PTS[vout]')
rates, data = ff.media.read(*urls, **kwargs)
print(rates)
print([(k, x["shape"], x["dtype"]) for k, x in data.items()])
def test_media_write():
fs, x = ff.audio.read("tests/assets/testaudio-1m.mp3")
fps, F = ff.video.read("tests/assets/testvideo-1m.mp4", vframes=120)
outext = ".mp4"
print(f"video: {len(F['buffer'])} bytes | audio: {len(x['buffer'])} bytes")
with TemporaryDirectory() as tmpdirname:
outfile = path.join(tmpdirname, f"out{outext}")
ff.media.write(
outfile, "va", (fps, F), (fs, x), show_log=True, shortest=ff.FLAG
)
pprint(ff.probe.format_basic(outfile))
pprint(ff.probe.streams_basic(outfile))
def test_media_filter():
fs, x = ff.audio.read("tests/assets/testaudio-1m.mp3")
fps, F = ff.video.read("tests/assets/testvideo-1m.mp4", vframes=120)
print(f"video: {len(F['buffer'])} bytes | audio: {len(x['buffer'])} bytes")
outrates, outdata = ff.media.filter(
["[0:V:0][1:V:0]vstack,split[out0]", "[2:a:0][3:a:0]amerge[out2]"],
"vvaa", # 4 inputs
(fps, F),
(fps, F),
(fs, x),
(fs, x),
output_streams=["[out0]", "out1", {"map": "[out2]"}],
show_log=True,
shortest=ff.FLAG,
)
assert all(k in ("out0", "out1", "out2") for k in outrates)
print(outrates)
def test_filter_empty():
import numpy as np
outrates, outdata = ff.media.filter(
[
"[0]format=pix_fmts=gray,fps=10,scale=40x40[v]",
"[1]aresample=8000,pan=mono|c0=c0[a]",
],
"va",
(30, np.empty((0, 100, 100, 3), dtype="uint8")),
(44100, np.empty((0, 2))),
output_streams=["v", "a"],
squeeze=False,
)
assert outrates["v"] == 10
assert outrates["a"] == 8000
assert outdata["v"]["shape"] == (0, 40, 40, 1)
assert outdata["a"]["shape"] == (0, 1)