python-ffmpeg/tests/test_asyncio_pipe.py at 6a66204858886a77e12f73b0b6ededf4a24587a9 · jonghwanhyeon/python-ffmpeg · GitHub

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
from pathlib import Path

import pytest
from helpers import probe

from ffmpeg.asyncio import FFmpeg

epsilon = 0.25


@pytest.mark.asyncio
async def test_asyncio_input_via_stdin(
    assets_path: Path,
    tmp_path: Path,
):
    source_path = assets_path / "pier-39.ts"
    target_path = tmp_path / "pier-39.mp4"

    with open(source_path, "rb") as source_file:
        source_bytes = source_file.read()

    ffmpeg = (
        FFmpeg()
        .option("y")
        .input("pipe:0")
        .output(
            str(target_path),
            codec="copy",
        )
    )
    await ffmpeg.execute(source_bytes)

    source = probe(source_path)
    target = probe(target_path)

    assert abs(float(source["format"]["duration"]) - float(target["format"]["duration"])) <= epsilon
    assert "mp4" in target["format"]["format_name"]

    assert source["streams"][0]["codec_name"] == target["streams"][0]["codec_name"]
    assert source["streams"][1]["codec_name"] == target["streams"][1]["codec_name"]


@pytest.mark.asyncio
async def test_asyncio_output_via_stdout(
    assets_path: Path,
    tmp_path: Path,
):
    source_path = assets_path / "brewing.wav"
    target_path = tmp_path / "brewing.ogg"

    ffmpeg = (
        FFmpeg()
        .option("y")
        .input(source_path)
        .output(
            "pipe:1",
            f="ogg",
        )
    )
    target_bytes = await ffmpeg.execute()
    with open(target_path, "wb") as target_file:
        target_file.write(target_bytes)

    source = probe(source_path)
    target = probe(target_path)

    assert abs(float(source["format"]["duration"]) - float(target["format"]["duration"])) <= epsilon
    assert target["format"]["format_name"] == "ogg"


@pytest.mark.asyncio
async def test_asyncio_output_stream(
        assets_path: Path,
        tmp_path: Path,
):
    source_path = assets_path / "brewing.wav"
    target_path = tmp_path / "brewing.ogg"

    ffmpeg = (
        FFmpeg()
        .option("y")
        .input(source_path)
        .output(
            "pipe:1",
            f="ogg",
        )
    )
    with open(target_path, "wb") as target_file:
        async for chunk in ffmpeg.execute_stream_stdout():
            target_file.write(chunk)

    source = probe(source_path)
    target = probe(target_path)

    assert abs(float(source["format"]["duration"]) - float(target["format"]["duration"])) <= epsilon
    assert target["format"]["format_name"] == "ogg"