# [Bugfix] Replace ffmpeg subprocess with PyAV to avoid HEVC decoder hang (#7364) (#7504)

Signed-off-by: RyanYun09 <318555231+RyanYun09@users.noreply.github.com>

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

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)

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))))