File size: 2,483 Bytes
2e0601e | 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 | """Bounded video sampling for SPEV evidence."""
import io
import math
from tempfile import NamedTemporaryFile
import av
from transformers.video_utils import load_video
def decode_video(data, video_processor, *, fps, max_seconds, max_frames, max_bytes):
if len(data) > max_bytes:
raise ValueError("Video exceeds the checkpoint byte limit")
try:
with av.open(io.BytesIO(data)) as container:
if not container.streams.video:
raise ValueError("Video has no visual stream")
stream = container.streams.video[0]
duration = float(stream.duration * stream.time_base) if stream.duration is not None else None
if duration is None and container.duration is not None:
duration = container.duration / av.time_base
if duration is None or not math.isfinite(duration) or not 0 < duration <= max_seconds:
raise ValueError("Video duration is missing or exceeds the checkpoint limit")
frame_rate = float(stream.average_rate) if stream.average_rate is not None else None
if frame_rate is None or not math.isfinite(frame_rate) or frame_rate <= 0:
raise ValueError("Video frame rate is missing or invalid")
frame_count = stream.frames or math.ceil(duration * frame_rate)
if not 0 < frame_count <= max_frames:
raise ValueError("Video frame count is missing or exceeds the checkpoint limit")
except av.FFmpegError as exc:
raise ValueError("Invalid or unsupported video") from exc
def sample_indices(metadata, **kwargs):
metadata.total_num_frames = frame_count
metadata.fps = frame_rate
metadata.duration = frame_count / frame_rate
return video_processor.sample_frames(metadata=metadata, fps=fps)
with NamedTemporaryFile() as source:
source.write(data)
source.flush()
try:
frames, metadata = load_video(source.name, backend="pyav", sample_indices_fn=sample_indices)
except (av.FFmpegError, OSError) as exc:
raise ValueError("Invalid or unsupported video") from exc
if not len(frames) or not metadata.fps or not metadata.total_num_frames:
raise ValueError("Video has no decodable frames")
if metadata.frames_indices is None or len(frames) != len(metadata.frames_indices):
raise ValueError("Video sampled-frame decode is incomplete")
return frames, metadata
|