通过文件类对象支持流式传输数据

在此示例中，我们将展示如何解码流式数据。也就是说，当文件不在本地时，我们将展示如何仅下载解码所需帧所需的数据段。我们通过Python 文件类对象实现此功能。我们的示例使用视频文件，因此我们使用VideoDecoder类进行解码。但这里的原理也适用于音频文件和AudioDecoder类。

首先是一些样板代码。我们定义了两个函数：一个用于从给定URL下载内容，另一个用于测量给定函数的执行时间。

import torch
import requests
from time import perf_counter_ns


def get_url_content(url):
    response = requests.get(url, headers={"User-Agent": ""})
    if response.status_code != 200:
        raise RuntimeError(f"Failed to download video. {response.status_code = }.")
    return response.content


def bench(f, average_over=10, warmup=2):
    for _ in range(warmup):
        f()

    times = []
    for _ in range(average_over):
        start = perf_counter_ns()
        f()
        end = perf_counter_ns()
        times.append(end - start)

    times = torch.tensor(times) * 1e-6  # ns to ms
    std = times.std().item()
    med = times.median().item()
    print(f"{med = :.2f}ms +- {std:.2f}")

性能：先下载 vs. 流式传输

我们将研究在解码任何帧之前必须下载整个视频的成本，以及在解码时能够流式传输视频数据的成本。为了演示一个极端情况，我们将始终只解码视频的第一帧，同时改变获取视频数据的方式。

本教程中使用的视频可在互联网上公开获取。我们对其进行初步下载，以便了解其大小和内容

from torchcodec.decoders import VideoDecoder

nasa_url = "https://download.pytorch.org/torchaudio/tutorial-assets/stream-api/NASAs_Most_Scientifically_Complex_Space_Observatory_Requires_Precision-MP4.mp4"

pre_downloaded_raw_video_bytes = get_url_content(nasa_url)
decoder = VideoDecoder(pre_downloaded_raw_video_bytes)

print(f"Video size in MB: {len(pre_downloaded_raw_video_bytes) // 1024 // 1024}")
print(decoder.metadata)

Video size in MB: 253
VideoStreamMetadata:
  duration_seconds_from_header: 206.039167
  begin_stream_seconds_from_header: 0.0
  bit_rate: 9958354.0
  codec: h264
  stream_index: 0
  begin_stream_seconds_from_content: 0.0
  end_stream_seconds_from_content: 206.039167
  width: 1920
  height: 1080
  num_frames_from_header: 6175
  num_frames_from_content: 6175
  average_fps_from_header: 29.97003
  duration_seconds: 206.039167
  begin_stream_seconds: 0.0
  end_stream_seconds: 206.039167
  num_frames: 6175
  average_fps: 29.970029921543997

我们可以看到该视频约为253 MB，分辨率为1920x1080，帧率约为30帧/秒，时长近3分半钟。由于我们只想解码第一帧，显然不必下载整个视频会带来好处！

我们首先测试三种场景

1. 从我们刚刚下载的*现有*视频中解码。这是我们的基准性能，因为我们将下载成本降至0。

2. 在解码前下载整个视频。这是我们想要避免的最坏情况。

3. 直接将URL提供给VideoDecoder类，它会将URL传递给FFmpeg。然后FFmpeg将决定在解码前下载多少视频数据。

注意，在我们的场景中，我们总是将VideoDecoder类的seek_mode参数设置为"approximate"。这样做是为了避免在初始化期间扫描整个视频，这会要求下载整个视频，即使我们只想解码第一帧。更多信息请参阅精确与近似seek模式：性能与准确度比较。

def decode_from_existing_download():
    decoder = VideoDecoder(
        source=pre_downloaded_raw_video_bytes,
        seek_mode="approximate",
    )
    return decoder[0]


def download_before_decode():
    raw_video_bytes = get_url_content(nasa_url)
    decoder = VideoDecoder(
        source=raw_video_bytes,
        seek_mode="approximate",
    )
    return decoder[0]


def direct_url_to_ffmpeg():
    decoder = VideoDecoder(
        source=nasa_url,
        seek_mode="approximate",
    )
    return decoder[0]


print("Decode from existing download:")
bench(decode_from_existing_download)
print()

print("Download before decode:")
bench(download_before_decode)
print()

print("Direct url to FFmpeg:")
bench(direct_url_to_ffmpeg)

Decode from existing download:
med = 238.16ms +- 0.38

Download before decode:
med = 1547.66ms +- 112.30

Direct url to FFmpeg:
med = 289.60ms +- 11.16

解码已经下载的视频显然是最快的。每次只想解码第一帧却必须下载整个视频比解码现有视频慢很多倍。提供直接URL要好得多，但我们可能仍然下载了超出所需的数据。

我们可以做得更好，方法是使用文件类对象，它实现了自己的read和seek方法，仅在需要时从URL下载数据。我们可以使用fsspec模块中的此类对象，该模块提供适用于Python的文件系统接口。请注意，使用fsspec库的这些功能还需要aiohttp模块。您可以使用pip install fsspec aiohttp安装两者。

import fsspec


def stream_while_decode():
    # The `client_kwargs` are passed down to the aiohttp module's client
    # session; we need to indicate that we need to trust the environment
    # settings for proxy configuration. Depending on your environment, you may
    # not need this setting.
    with fsspec.open(nasa_url, client_kwargs={'trust_env': True}) as file_like:
        decoder = VideoDecoder(file_like, seek_mode="approximate")
        return decoder[0]


print("Stream while decode: ")
bench(stream_while_decode)

Stream while decode:
med = 261.20ms +- 0.67

通过文件类对象流式传输数据比先下载视频要快得多。而且它不仅比提供直接URL更快，而且更通用。VideoDecoder支持直接URL是因为底层的FFmpeg函数支持它们。但支持的协议类型取决于该版本FFmpeg支持的内容。文件类对象可以适应任何类型的资源，包括您自己的基础设施特有的且FFmpeg未知资源。

工作原理

在Python中，文件类对象是任何暴露了用于读、写和seek的特殊方法的对象。虽然这些方法显然是面向文件的，但文件类对象不一定由实际文件支持。对于Python而言，如果一个对象行为像文件，那么它就是文件。这是一个强大的概念，因为它使得读写数据的库可以假定文件类接口。其他呈现新资源的库随后可以通过为其资源提供文件类包装轻松使用。

对于我们的情况，解码只需要read和seek方法。所需的精确方法签名在下面的示例中。我们不包装新资源，而是通过包装实际文件并计算每个方法被调用的次数来演示此功能。

from pathlib import Path
import tempfile

# Create a local file to interact with.
temp_dir = tempfile.mkdtemp()
nasa_video_path = Path(temp_dir) / "nasa_video.mp4"
with open(nasa_video_path, "wb") as f:
    f.write(pre_downloaded_raw_video_bytes)


# A file-like class that is backed by an actual file, but it intercepts reads
# and seeks to maintain counts.
class FileOpCounter:
    def __init__(self, file):
        self._file = file
        self.num_reads = 0
        self.num_seeks = 0

    def read(self, size: int) -> bytes:
        self.num_reads += 1
        return self._file.read(size)

    def seek(self, offset: int, whence: int) -> bytes:
        self.num_seeks += 1
        return self._file.seek(offset, whence)


# Let's now get a file-like object from our class defined above, providing it a
# reference to the file we created. We pass our file-like object to the decoder
# rather than the file itself.
file_op_counter = FileOpCounter(open(nasa_video_path, "rb"))
counter_decoder = VideoDecoder(file_op_counter, seek_mode="approximate")

print("Decoder initialization required "
      f"{file_op_counter.num_reads} reads and "
      f"{file_op_counter.num_seeks} seeks.")

init_reads = file_op_counter.num_reads
init_seeks = file_op_counter.num_seeks

first_frame = counter_decoder[0]

print("Decoding the first frame required "
      f"{file_op_counter.num_reads - init_reads} additional reads and "
      f"{file_op_counter.num_seeks - init_seeks} additional seeks.")

Decoder initialization required 9 reads and 11 seeks.
Decoding the first frame required 2 additional reads and 1 additional seeks.

虽然我们主要为了演示定义了一个简单的类，但它实际上对于诊断不同解码操作需要多少read和seek操作非常有用。我们还引入了一个谜题，需要回答：为什么解码器的初始化比解码第一帧需要更多的read和seek操作？答案是，在我们的解码器实现中，我们实际上调用了一个特殊的FFmpeg函数，该函数解码前几帧以返回更可靠的元数据。

还值得注意的是，Python文件类接口只是故事的一半。FFmpeg也有自己的机制，用于在解码期间将read和seek操作导向用户定义的函数。VideoDecoder对象负责将您定义的Python方法连接到FFmpeg。您只需在Python中定义您的方法，其余的工作由我们完成。

性能：本地文件路径 vs. 本地文件类对象

既然我们定义了一个本地文件，我们来做一个额外的性能测试。现在我们有两种向VideoDecoder提供本地文件的方法

1. 通过路径，其中VideoDecoder对象将负责打开该路径下的本地文件。

2. 通过文件类对象，您自己打开文件并将该文件类对象提供给VideoDecoder。

一个显而易见的问题是：哪种方式更快？下面的代码测试了这个问题。

def decode_from_existing_file_path():
    decoder = VideoDecoder(nasa_video_path, seek_mode="approximate")
    return decoder[0]


def decode_from_existing_open_file_object():
    with open(nasa_video_path, "rb") as file:
        decoder = VideoDecoder(file, seek_mode="approximate")
        return decoder[0]


print("Decode from existing file path:")
bench(decode_from_existing_file_path)
print()

print("Decode from existing open file object:")
bench(decode_from_existing_open_file_object)

Decode from existing file path:
med = 238.30ms +- 0.72

Decode from existing open file object:
med = 240.67ms +- 0.66

幸运的是，答案是通过本地文件解码的两种方式花费的时间大致相同。这个结果意味着在您的代码中，可以使用更方便的方法。这个结果意味着实际读取和复制数据的成本在解码时占主导地位，高于调用Python方法的成本。

最后，我们清理创建的本地资源。

import shutil
shutil.rmtree(temp_dir)