ffmpeg サブプロセスによるビデオ データ IO

求職活動を再開したとき (はい、私はまだ #OpenToWork です。連絡してください!)、求人応募の 1 つで、ビデオ データを処理するプロトタイプを実装するように求められました。このプロジェクトに取り組んでいる間、私はこの分野で比較的経験が浅かったため、思いがけず生成 AI チャットボットから多くの助けを得ることができました。

タイトルで述べたように、いくつかの前処理作業を実行するために ffmpeg が使用されました。プロジェクトの目標の 1 つは、複数のビデオ ファイルを次々に再生できるようにすることでした。これを実現するには複数の方法がありますが、私はそれらを連結するという最も明白な解決策を採用することにしました。

$ cat video1 video2 video3 | python further-work.py

これを実現するには、まずファイルをそれが可能な形式に再エンコードする必要がありました。これについて Google Gemini と「話し合った」後、チャットボットは、この目的のために MPEG-TS を使用することを推奨しました。

MPEG トランスポート ストリーム (MPEG-TS) は、パケット化されたエレメンタリ ストリームをカプセル化することで機能します。これらのストリームには、小さなセグメントにパケット化されたオーディオ、ビデオ、および PSIP データが含まれます。各ストリームは 188 バイトのセクションに分割され、インターリーブされます。このプロセスにより、レイテンシが短縮され、エラー耐性が向上するため、大きなフレームによって音声遅延が発生する可能性があるビデオ会議に最適です。

https://castr.com/blog/mpeg-transport-stream-mp​​eg-ts/

より引用

この目的に使用できるファイル形式は他にもありますが、それらは議論とは無関係です。ビデオをこの形式に再エンコードした後、ビデオ データはキューに送信され、他のプロセスで実行されている他のモジュールによって消費されます。

入力 (オンラインで取得するビデオ ファイルのリスト) と出力 (再エンコードされたビデオ ファイル コンテンツ) の両方を定義したら、次はそれを行う方法を考えます。残念ながら、ffmpeg は非常に多くのことを行う非常に複雑なユーティリティです。ユーザーを支援するインターフェイスを提供する試みが複数回ありました (私は本当にこれを試してみたかったのですが、どうやら現在は廃止されているようです)。ただし、最近では生成 AI が非常に便利になっているため、適切なコマンドを数回入力するだけですぐに使用できるようになります。

ffmpeg -hwaccel cuda -i pipe:0 -c:v h264_nvenc -b:v 1.5M -c:a aac -b:a 128k -f mpegts -y pipe:1

以下のスクリーンショットに示すように、各引数の意味についても説明されています。

Gemini による ffmpeg コマンドの説明

つまり、このコマンドは stdin 経由でビデオ ファイル コンテンツを受け取り、再エンコードされたビデオ ファイル コンテンツを stdout として出力します。

今度は実装をコーディングします。ffmpeg の読み取りと書き込みを同時に行いたいので、これは asyncio アプリケーションになります。今回使用している http クライアント ライブラリは httpx です。これには、ダウンロードを小さいバッチで取得するメソッドがあります:

$ cat video1 video2 video3 | python further-work.py

実際の処理については後で考えますが、今のところは、チャンクを画面に出力するコードを取得するだけです。

次に、asyncio.create_subprocess_exec を通じて ffmpeg を呼び出す関数を作成します

ffmpeg -hwaccel cuda -i pipe:0 -c:v h264_nvenc -b:v 1.5M -c:a aac -b:a 128k -f mpegts -y pipe:1

理想的には、ドキュメントでアドバイスされているように、ここで process.communicate(file_content) を使用しますが、残念ながら、それを実行すると、最初にファイル全体をダウンロードする必要があり、必然的に応答が遅くなり、理想的ではありませんでした。

代わりに、process.stdin.write() を使用できます。元の write_input 関数を更新しましょう。

import httpx

client = httpx.AsyncClient()

async def write_input(
    client: httpx.AsyncClient, video_link: str, process: asyncio.subprocess.Process
) -> None:
    async with client.stream("GET", video_link) as response:
        async for chunk in response.aiter_raw(1024):
            print(chunk) # this is the downloaded video file, in chunks

ダウンロードされたチャンクごとに、

1. process.stdin.write(chunk) を通じてプロセスにそれをフィードします。
2. 完了したら、ファイル入力の終わりを示す EOF (process.stdin.write_eof()) を書き込みます。
3. その後に .close() (および対応する await .wait_closed()) が続きます

video_send 関数に戻り、process.stdout を読み取って関数を続行します。読み取りと書き込みの両方を実行できることが、まさに asyncio を通じてこれを実行している理由です。以前の同期設定では決まった順番で次々と実行することしかできませんでしたが、順番をスケジューラに気にさせることができるようになりました。これで、関数には、再エンコードされたファイルのコンテンツを読み取り、キューにポストするための次のコードが追加されました。

async def video_send(client: httpx.AsyncClient, video_link: str) -> None:
    logger.info("DATA: Fetching video from link", link=video_link)
    process = await asyncio.create_subprocess_exec(
        "ffmpeg",
        "-hwaccel",
        "cuda",
        "-i",
        "pipe:0",
        "-c:v",
        "h264_nvenc",
        # "libx264",
        "-b:v",
        "1.5M",
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        "-f",
        "mpegts",
        "-y",
        "pipe:1",
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
    )

    asyncio.create_task(write_input(client, video_link, process))

ループの中で、私たちは

1. ffmpeg stdout からデータのチャンクを取得します
2. チャンクが空の文字列の場合、ループを終了します
3. それ以外の場合は、チャンクをキューにプッシュします (ここではプロセスセーフバージョンを使用しているため、asyncio.to_thread を通じて)
4. その後、process.wait() を通じてコマンドが正常に終了するのを待ちます

今では非常に簡単に思えますが、これを実際に正しく実行するのに一晩かかりました (そして、これを書いている間、まだコードを修正していました)。半分の時間は、ドキュメントに目を通し、何も抜けていないことを確認していました。残りの時間は、Gemini にコードをレビューしてもらいました。

これがお役に立てば幸いです。今日はここまでです。来週は、以前に約束した Advent of Code のコンテンツに戻ることを願っています。

以上がffmpeg サブプロセスによるビデオ データ IOの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。