通过ffmpeg子进程进行视频数据IO

当我重新开始找工作时（是的，我仍然#OpenToWork，请告诉我！），在一份工作申请中，我被要求实现一个处理视频数据的原型。在完成该项目的过程中，由于我在该领域相对缺乏经验，我意外地从生成式 AI 聊天机器人那里得到了很多帮助。

如标题所述，使用ffmpeg进行一些预处理工作。该项目的目标之一是能够依次播放多个视频文件。虽然有多种方法可以实现这一目标，但我决定采用最明显的解决方案，将它们连接在一起。

$ 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 是一个非常复杂的实用程序，可以做很多事情。有多次尝试提供一些界面来帮助用户使用它（我真的很想尝试这个，但显然它现在已经死了）。然而，如今生成式人工智能的帮助非常大，只需几个提示即可获得正确的命令。

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，所以这将是一个异步应用程序。我们这次使用的http客户端库是httpx，它有一个小批量获取下载的方法：

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

我们稍后担心实际处理，现在我们只需获取将块打印到屏幕上的代码。

接下来我们编写一个函数来调用ffmpeg，通过asyncio.create_subprocess_exec

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_angled()）

回到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. 如果 chunk 是空字符串，则退出循环
3. 否则，将块推送到队列（通过 asyncio.to_thread，因为我们在这里使用进程安全版本）
4. 然后我们通过process.wait()等待命令优雅退出

现在看起来很简单，但我花了整整一个晚上才真正正确地完成了这件事（而且我在写这篇文章时仍在修改代码）。有一半的时间我会检查文档以确保没有遗漏任何内容，其他时间我会让 Gemini 审查我的代码。

希望您觉得本文有用，今天就到此为止，希望我们下周能回到之前承诺的“代码降临”内容。

以上是通过ffmpeg子进程进行视频数据IO的详细内容。更多信息请关注PHP中文网其他相关文章！