From d3396fc5fb16c9b4de6c8fb9ed0f67cc732b7af2 Mon Sep 17 00:00:00 2001 From: RyanYun09 Date: Sun, 20 Sep 2026 22:37:09 +0800 Subject: [PATCH] # [Bugfix] Replace ffmpeg subprocess with PyAV to avoid HEVC decoder hang (#7364) (#7504) Signed-off-by: RyanYun09 <318555231+RyanYun09@users.noreply.github.com> --- .../duplex/test_omni_duplex_eval_media.py | 276 +++++++++++++++++- .../duplex/omni_duplex_eval_eval.py | 3 +- .../duplex/omni_duplex_eval_media.py | 133 +++++++-- 3 files changed, 378 insertions(+), 34 deletions(-) diff --git a/tests/benchmarks/duplex/test_omni_duplex_eval_media.py b/tests/benchmarks/duplex/test_omni_duplex_eval_media.py index 28a502428..ad2fffc9d 100644 --- a/tests/benchmarks/duplex/test_omni_duplex_eval_media.py +++ b/tests/benchmarks/duplex/test_omni_duplex_eval_media.py @@ -1,15 +1,289 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Tests for omni_duplex_eval_media — PyAV-backed media helpers.""" + from __future__ import annotations +import io +from pathlib import Path + +import av +import numpy as np import pytest -from vllm_omni.benchmarks.duplex.omni_duplex_eval_media import materialize_media +from vllm_omni.benchmarks.duplex.omni_duplex_eval_media import ( + _ffmpeg_q_to_pil_quality, + extract_jpeg, + materialize_media, + read_audio_pcm16, + video_duration, +) pytestmark = [pytest.mark.core_model, pytest.mark.cpu] +# --------------------------------------------------------------------------- +# Helpers: synthetic test media generated with PyAV (no external files) +# --------------------------------------------------------------------------- + + +def _make_synthetic_mp4(duration: float = 1.0, fps: int = 30) -> bytes: + """Return an MP4 byte blob with a solid-color pattern. + + Uses the built-in ``mpeg4`` encoder — available everywhere PyAV's bundled + FFmpeg is present — so no external codec dependency. + """ + buf = io.BytesIO() + total_frames = int(duration * fps) + with av.open(buf, mode="w", format="mp4") as container: + stream = container.add_stream("mpeg4", rate=fps) + stream.width = 64 + stream.height = 64 + stream.pix_fmt = "yuv420p" + for i in range(total_frames): + frame = av.VideoFrame.from_ndarray(np.full((64, 64, 3), (i * 8) % 256, dtype=np.uint8), format="rgb24") + for pkt in stream.encode(frame): + container.mux(pkt) + for pkt in stream.encode(): + container.mux(pkt) + return buf.getvalue() + + +def _make_synthetic_wav(duration: float = 1.0, sample_rate: int = 16000) -> bytes: + """Return a 16-bit mono PCM WAV byte blob (440 Hz sine wave).""" + import wave as _wave + + n = int(sample_rate * duration) + samples = (np.sin(2 * np.pi * 440 * np.arange(n) / sample_rate) * 32767).astype(np.int16) + buf = io.BytesIO() + with _wave.open(buf, "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(sample_rate) + wf.writeframes(samples.tobytes()) + return buf.getvalue() + + +def _make_synthetic_ogg(duration: float = 1.0, sample_rate: int = 16000) -> bytes: + """Return an OGG Vorbis byte blob (440 Hz sine wave). + + Uses ``libopus`` as a fallback if ``libvorbis`` is unavailable in the + bundled FFmpeg (e.g. PyAV wheels from an older build). + """ + buf = io.BytesIO() + n = int(sample_rate * duration) + samples = (np.sin(2 * np.pi * 440 * np.arange(n) / sample_rate) * 0.3).astype(np.float32) + with av.open(buf, mode="w", format="ogg") as container: + codec_name = "libopus" if "libvorbis" not in av.codecs_available else "libvorbis" + stream = container.add_stream(codec_name, rate=sample_rate, layout="mono") + frame = av.AudioFrame.from_ndarray(samples[np.newaxis, :], format="fltp", layout="mono") + frame.sample_rate = sample_rate + for pkt in stream.encode(frame): + container.mux(pkt) + for pkt in stream.encode(): + container.mux(pkt) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# video_duration +# --------------------------------------------------------------------------- + + +class TestVideoDuration: + @staticmethod + def test_returns_duration(tmp_path: Path) -> None: + mp4 = _make_synthetic_mp4(duration=2.0, fps=30) + path = tmp_path / "test.mp4" + path.write_bytes(mp4) + dur = video_duration(path) + assert abs(dur - 2.0) < 0.05, f"expected ~2.0s, got {dur}" + + @staticmethod + def test_raises_for_missing_file(tmp_path: Path) -> None: + with pytest.raises((OSError, ValueError)): + video_duration(tmp_path / "nonexistent.mp4") + + @staticmethod + def test_raises_for_broken_file(tmp_path: Path) -> None: + broken = tmp_path / "broken.mp4" + broken.write_bytes(b"\x00\x00\x00\x00") + with pytest.raises((OSError, ValueError)): + video_duration(broken) + + +# --------------------------------------------------------------------------- +# extract_jpeg +# --------------------------------------------------------------------------- + + +class TestExtractJpeg: + @staticmethod + def test_returns_valid_jpeg(tmp_path: Path) -> None: + mp4 = _make_synthetic_mp4(duration=1.0, fps=30) + path = tmp_path / "test.mp4" + path.write_bytes(mp4) + jpeg = extract_jpeg(path, timestamp=0.5, quality=3) + assert jpeg.startswith(b"\xff\xd8"), "not a valid JPEG header" + assert len(jpeg) > 100 + + @staticmethod + def test_quality_maps_to_pil_scale(tmp_path: Path) -> None: + """Higher ffmpeg quality (lower q value) produces a larger JPEG.""" + mp4 = _make_synthetic_mp4(duration=1.0, fps=30) + path = tmp_path / "test.mp4" + path.write_bytes(mp4) + high = extract_jpeg(path, timestamp=0.5, quality=1) + low = extract_jpeg(path, timestamp=0.5, quality=31) + assert len(high) > len(low), "quality=1 should produce larger JPEG than quality=31" + + @staticmethod + def test_negative_timestamp_clamped(tmp_path: Path) -> None: + """Negative timestamps are clamped to 0.""" + mp4 = _make_synthetic_mp4(duration=1.0, fps=30) + path = tmp_path / "test.mp4" + path.write_bytes(mp4) + jpeg = extract_jpeg(path, timestamp=-1.0, quality=3) + assert jpeg.startswith(b"\xff\xd8") + + @staticmethod + def test_raises_value_error_for_no_video(tmp_path: Path) -> None: + wav = _make_synthetic_wav() + path = tmp_path / "test.wav" + path.write_bytes(wav) + with pytest.raises(ValueError): + extract_jpeg(path, timestamp=0.0) + + @staticmethod + def test_long_gop_returns_different_frames_for_different_timestamps(tmp_path: Path) -> None: + """Long GOP: different timestamps must return different frame content (P2).""" + # Create a long-GOP video: 3s @ 10fps = 30 frames, gop_size=30 (1 keyframe) + buf = io.BytesIO() + fps = 10 + duration = 3.0 + total_frames = int(duration * fps) + with av.open(buf, mode="w", format="mp4") as container: + stream = container.add_stream("mpeg4", rate=fps) + stream.width = 64 + stream.height = 64 + stream.pix_fmt = "yuv420p" + stream.gop_size = 30 # 3 seconds per keyframe + for i in range(total_frames): + frame = av.VideoFrame.from_ndarray(np.full((64, 64, 3), (i * 64) % 256, dtype=np.uint8), format="rgb24") + for pkt in stream.encode(frame): + container.mux(pkt) + for pkt in stream.encode(): + container.mux(pkt) + mp4 = buf.getvalue() + + path = tmp_path / "long_gop.mp4" + path.write_bytes(mp4) + + # Extract frames at different timestamps + frame0 = extract_jpeg(path, timestamp=0.0, quality=3) + frame1 = extract_jpeg(path, timestamp=1.0, quality=3) + frame2 = extract_jpeg(path, timestamp=2.0, quality=3) + + # Verify all are valid JPEGs + assert frame0.startswith(b"\xff\xd8") + assert frame1.startswith(b"\xff\xd8") + assert frame2.startswith(b"\xff\xd8") + + # Verify content differs: decode JPEGs and compare mean brightness + from PIL import Image + + img0 = np.asarray(Image.open(io.BytesIO(frame0))) + img1 = np.asarray(Image.open(io.BytesIO(frame1))) + img2 = np.asarray(Image.open(io.BytesIO(frame2))) + means = [img.mean() for img in (img0, img1, img2)] + assert len(set(round(m, 1) for m in means)) >= 2, ( + f"timestamps 0/1/2s should give different frames, got means {means}" + ) + + +# --------------------------------------------------------------------------- +# read_audio_pcm16 +# --------------------------------------------------------------------------- + + +class TestReadAudioPcm16: + @staticmethod + def test_wav_fast_path(tmp_path: Path) -> None: + """WAV with matching params uses the fast wave module path.""" + wav = _make_synthetic_wav(duration=1.0, sample_rate=16000) + path = tmp_path / "test.wav" + path.write_bytes(wav) + pcm = read_audio_pcm16(path) + assert len(pcm) == 16000 * 2, f"expected 32000 bytes, got {len(pcm)}" + assert isinstance(pcm, bytes) + + @staticmethod + def test_wav_different_rate_resamples(tmp_path: Path) -> None: + """A 48 kHz WAV is resampled down to 16 kHz — exact sample count after flush.""" + wav = _make_synthetic_wav(duration=1.0, sample_rate=48000) + path = tmp_path / "test.wav" + path.write_bytes(wav) + pcm = read_audio_pcm16(path) + assert len(pcm) > 0 + assert len(pcm) % 2 == 0 + # 1s @ 48kHz → 1s @ 16kHz = 16000 samples = 32000 bytes (±1 sample) + assert abs(len(pcm) - 32000) <= 4, f"expected ~32000 bytes, got {len(pcm)}" + + @staticmethod + def test_non_wav_ogg(tmp_path: Path) -> None: + """Non-WAV audio (OGG) is decoded and resampled via PyAV.""" + ogg = _make_synthetic_ogg(duration=1.0, sample_rate=16000) + path = tmp_path / "test.ogg" + path.write_bytes(ogg) + pcm = read_audio_pcm16(path) + assert len(pcm) > 0 + assert len(pcm) % 2 == 0 + + @staticmethod + def test_non_wav_ogg_resample_accuracy(tmp_path: Path) -> None: + """OGG resample should produce exact sample count after flush.""" + ogg = _make_synthetic_ogg(duration=1.5, sample_rate=48000) + path = tmp_path / "test.ogg" + path.write_bytes(ogg) + pcm = read_audio_pcm16(path) + # 1.5s @ 48kHz → 1.5s @ 16kHz = 24000 samples = 48000 bytes (±1 sample) + expected = int(1.5 * 16000) * 2 + assert abs(len(pcm) - expected) <= 4, f"expected ~{expected} bytes, got {len(pcm)}" + + @staticmethod + def test_raises_for_broken_file(tmp_path: Path) -> None: + """Corrupt audio must raise, not return empty bytes (L1).""" + broken = tmp_path / "broken.mp3" + broken.write_bytes(b"\x00\x00\x00\x00") + with pytest.raises((OSError, ValueError)): + read_audio_pcm16(broken) + + +# --------------------------------------------------------------------------- +# _ffmpeg_q_to_pil_quality +# --------------------------------------------------------------------------- + + +class TestFfmpegQToPilQuality: + @staticmethod + def test_q1_maps_to_best() -> None: + assert _ffmpeg_q_to_pil_quality(1) >= 90 + + @staticmethod + def test_q31_maps_to_lowest() -> None: + assert _ffmpeg_q_to_pil_quality(31) <= 10 + + @staticmethod + def test_monotonic() -> None: + """Higher ffmpeg q → lower PIL quality (monotonically non-increasing).""" + prev = 100 + for q in range(1, 32): + p = _ffmpeg_q_to_pil_quality(q) + assert p <= prev, f"quality.increase at q={q}: {prev} → {p}" + prev = p + + def test_dict_with_bytes_materializes_a_real_file(tmp_path): # A Hugging Face Audio/Video feature value {"bytes": ..., "path": ...} used # to short-circuit on ``path`` (the artifact name inside the dataset, which diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py index a92376bd7..1dfa68246 100644 --- a/vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py +++ b/vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py @@ -6,7 +6,6 @@ from __future__ import annotations import json -import subprocess from pathlib import Path from typing import Any @@ -168,7 +167,7 @@ def _extract_frames(path: str | Path, timestamps: list[float]) -> list[bytes]: for timestamp in timestamps: try: frame = extract_jpeg(path, timestamp=timestamp) - except (OSError, subprocess.CalledProcessError, ValueError): + except (OSError, ValueError): continue if frame.startswith(b"\xff\xd8"): frames.append(frame) diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py index b45686fb8..ae198902a 100644 --- a/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py +++ b/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py @@ -1,12 +1,18 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project -"""Small ffmpeg-backed media helpers used by the duplex runner and judge.""" +"""Small media helpers used by the duplex runner and judge. + +Video and audio decoding is delegated to PyAV (``av``), which bundles a pinned +FFmpeg version, so results do not depend on whatever ffmpeg binary the host +provides (see #7364: host ffmpeg 4.4.2 could hang forever on some HEVC clips). +""" from __future__ import annotations +import io +import logging import os -import subprocess import tempfile import wave from collections.abc import Iterator @@ -15,10 +21,7 @@ from typing import Any from vllm_omni.experimental.fullduplex.client import PCM16_BYTES_PER_SAMPLE, PCM16_SAMPLE_RATE - -def _run_ffmpeg(args: list[str]) -> bytes: - result = subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", *args], capture_output=True, check=True) - return result.stdout +logger = logging.getLogger(__name__) def materialize_media(value: Any, output_dir: str | Path, stem: str, suffix: str) -> Path: @@ -77,31 +80,73 @@ def _write_media_bytes(payload: bytes | bytearray, output_dir: str | Path, stem: def video_duration(path: str | Path) -> float: - result = subprocess.run( - ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)], - capture_output=True, - text=True, - check=True, - ) - return max(0.0, float(result.stdout.strip())) + """Return the video duration in seconds using PyAV (no ffprobe subprocess). + + Raises: + OSError: If the file cannot be opened (missing file, permission denied, etc.). + ValueError: If the file is corrupt or cannot be parsed (e.g. + :class:`av.error.InvalidDataError`). + """ + import av + + try: + with av.open(str(path)) as container: + # Container-level duration is the same source ffprobe's format=duration reads. + if container.duration and container.duration > 0: + return max(0.0, container.duration / av.time_base) + if container.streams.video: + stream = container.streams.video[0] + if stream.duration is not None and stream.time_base is not None: + return max(0.0, float(stream.duration * stream.time_base)) + # Last resort: count decodable frames. + if container.streams.video: + stream = container.streams.video[0] + stream.thread_type = "AUTO" + frame_count = sum(1 for _ in container.decode(stream)) + if stream.average_rate and frame_count > 0: + return frame_count / float(stream.average_rate) + return 0.0 + except (OSError, ValueError) as exc: + logger.warning("Failed to read video duration from %s: %s", path, exc) + raise def extract_jpeg(path: str | Path, *, timestamp: float, quality: int = 3) -> bytes: - return _run_ffmpeg( - [ - "-ss", - f"{max(0.0, timestamp):.3f}", - "-i", - str(path), - "-frames:v", - "1", - "-q:v", - str(quality), - "-f", - "image2", - "pipe:1", - ] - ) + """Extract a single JPEG frame at ``timestamp`` using PyAV. + + ``quality`` follows the ffmpeg ``-q:v`` convention (lower = better, 1..31) + and is mapped to the PIL JPEG quality scale (higher = better). + + After ``seek`` the function continues decoding until a frame with PTS >= + target_pts is found, matching the ``ffmpeg -ss`` semantic of returning the + first decodable frame *at or after* the requested timestamp. + + Raises: + ValueError: If the file contains no video stream or no frame could be decoded. + """ + import av + + with av.open(str(path)) as container: + if not container.streams.video: + raise ValueError(f"no video stream in {path}") + stream = container.streams.video[0] + stream.thread_type = "AUTO" + time_base = stream.time_base + if time_base is None: + time_base = 1 / 90000 + target_pts = int(max(0.0, timestamp) / time_base) + container.seek(target_pts, any_frame=False, backward=True, stream=stream) + closest_frame = None + for frame in container.decode(stream): + closest_frame = frame + if frame.pts is not None and frame.pts >= target_pts: + break + if closest_frame is not None: + image = closest_frame.to_image() + buf = io.BytesIO() + image.save(buf, format="JPEG", quality=_ffmpeg_q_to_pil_quality(quality)) + return buf.getvalue() + raise ValueError(f"no decodable video frames in {path} at timestamp {timestamp}") def iter_jpegs( @@ -113,7 +158,8 @@ def iter_jpegs( while timestamp < end: try: frame = extract_jpeg(path, timestamp=timestamp, quality=quality) - except (OSError, subprocess.CalledProcessError): + except (OSError, ValueError) as exc: + logger.warning("Frame extraction failed at %.3fs for %s: %s", timestamp, path, exc) break if frame: yield timestamp, frame @@ -132,9 +178,28 @@ def read_audio_pcm16(path: str | Path) -> bytes: and wav_file.getcomptype() == "NONE" ): return wav_file.readframes(wav_file.getnframes()) - except wave.Error: + except (wave.Error, OSError): pass - return _run_ffmpeg(["-i", str(source), "-f", "s16le", "-ac", "1", "-ar", str(PCM16_SAMPLE_RATE), "pipe:1"]) + # Non-WAV audio: decode and resample to mono s16 PCM via PyAV. + import av + import numpy as np + + try: + resampler = av.AudioResampler(format="s16", layout="mono", rate=PCM16_SAMPLE_RATE) + chunks: list[Any] = [] + with av.open(str(source)) as container: + for frame in container.decode(audio=0): + for resampled in resampler.resample(frame): + chunks.append(resampled.to_ndarray()) + # Flush the resampler to drain buffered tail samples (P3). + for resampled in resampler.resample(None): + chunks.append(resampled.to_ndarray()) + if not chunks: + return b"" + return np.concatenate([chunk.reshape(-1) for chunk in chunks]).astype(np.int16).tobytes() + except (OSError, ValueError) as exc: + logger.warning("Failed to decode audio in %s: %s", source, exc) + raise def iter_av_units( @@ -150,3 +215,9 @@ def iter_av_units( frame = next_frame[1] next_frame = next(frame_iter, None) yield audio_pcm16[offset : offset + unit_bytes], frame + + +def _ffmpeg_q_to_pil_quality(quality: int) -> int: + """Map ffmpeg ``-q:v`` (1..31, lower = better) to PIL JPEG quality (1..100, higher = better).""" + ffmpeg_q = max(1, min(31, int(quality))) + return max(1, min(100, round(95 - (ffmpeg_q - 1) * (93 / 30))))