[CI/Build][MiniCPM-o] Add MiniCPM-o 4.5 accuracy and performance nightly coverage (#5710)

Signed-off-by: psv666 <2693925048@qq.com>
Signed-off-by: psv666 <150513104+psv666@users.noreply.github.com>
Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

vllm_omni/benchmarks/data_modules/daily_omni_dataset.py

@@ -110,47 +110,17 @@ def _daily_omni_find_videos_root_in_extract(tmp: Path) -> Path:
     )
 
 
-def ensure_daily_omni_hub_videos_dir(repo_id: str) -> Path:
-    """Download ``Videos.tar`` from the Hugging Face dataset repo and return the ``Videos`` root.
+def _extract_daily_omni_videos_tar(tar_path: Path, cache_key: str) -> Path:
+    """Extract ``Videos.tar`` once into the shared media cache and return the ``Videos`` root.
 
-    The returned path matches ``--daily-omni-video-dir`` (directory containing ``{{video_id}}/``).
-
-    Cached under ``HF_HOME`` / ``vllm_omni/daily_omni_media/<repo>``. Reuses extraction when the
-    tarball fingerprint matches.
-
-    Raises:
-        ImportError: if ``huggingface_hub`` is not installed.
-        FileNotFoundError / RuntimeError: if the archive is missing or malformed.
+    Cached under ``HF_HOME`` / ``vllm_omni/daily_omni_media/<cache_key>``. Reuses a previous
+    extraction when the tarball fingerprint matches.
     """
-    rid = (repo_id or "").strip()
-    if not rid:
-        raise ValueError("repo_id is required to download Daily-Omni Videos.tar")
-
-    try:
-        from huggingface_hub import hf_hub_download
-    except ImportError as e:
-        raise ImportError(
-            "Daily-Omni Hub media download requires huggingface_hub. "
-            "Install it (e.g. with vLLM) or provide --daily-omni-video-dir with a local extract."
-        ) from e
-
-    safe = rid.replace("/", "__").replace("\\", "_")
+    safe = cache_key.replace("/", "__").replace("\\", "_")
     staging_root = _daily_omni_hf_cache_root() / "vllm_omni" / "daily_omni_media" / safe
     videos_dir = staging_root / "Videos"
     marker = staging_root / ".videos_extracted"
 
-    tar_path: Path | None = None
-    for fname in ("Videos.tar", "videos.tar"):
-        try:
-            tar_path = Path(hf_hub_download(repo_id=rid, filename=fname, repo_type="dataset"))
-            break
-        except Exception:
-            continue
-    if tar_path is None or not tar_path.is_file():
-        raise FileNotFoundError(
-            f"Could not download Videos.tar from Hugging Face dataset {rid!r} (tried Videos.tar / videos.tar)."
-        )
-
     fp = _daily_omni_tar_fingerprint(tar_path)
     if marker.is_file() and videos_dir.is_dir():
         try:
@@ -167,7 +137,7 @@ def ensure_daily_omni_hub_videos_dir(repo_id: str) -> Path:
     shutil.rmtree(work, ignore_errors=True)
     work.mkdir(parents=True)
     try:
-        logger.info("Extracting Daily-Omni Videos.tar from %s (repo=%s)", tar_path, rid)
+        logger.info("Extracting Daily-Omni Videos.tar from %s (cache_key=%s)", tar_path, cache_key)
         with tarfile.open(tar_path, "r:*") as tf:
             tf.extractall(path=work, filter="data")
         found = _daily_omni_find_videos_root_in_extract(work)
@@ -178,10 +148,106 @@ def ensure_daily_omni_hub_videos_dir(repo_id: str) -> Path:
         shutil.rmtree(work, ignore_errors=True)
 
     marker.write_text(fp, encoding="utf-8")
-    logger.info("Daily-Omni Hub media ready at %s", videos_dir)
+    logger.info("Daily-Omni media ready at %s", videos_dir)
     return videos_dir
 
 
+def ensure_daily_omni_hub_videos_dir(repo_id: str) -> Path:
+    """Download ``Videos.tar`` from the Hugging Face dataset repo and return the ``Videos`` root.
+
+    The returned path matches ``--daily-omni-video-dir`` (directory containing ``{{video_id}}/``).
+
+    Raises:
+        ImportError: if ``huggingface_hub`` is not installed.
+        FileNotFoundError / RuntimeError: if the archive is missing or malformed.
+    """
+    rid = (repo_id or "").strip()
+    if not rid:
+        raise ValueError("repo_id is required to download Daily-Omni Videos.tar")
+
+    try:
+        from huggingface_hub import hf_hub_download
+    except ImportError as e:
+        raise ImportError(
+            "Daily-Omni Hub media download requires huggingface_hub. "
+            "Install it (e.g. with vLLM) or provide --daily-omni-video-dir with a local extract."
+        ) from e
+
+    tar_path: Path | None = None
+    for fname in ("Videos.tar", "videos.tar"):
+        try:
+            tar_path = Path(hf_hub_download(repo_id=rid, filename=fname, repo_type="dataset"))
+            break
+        except Exception:
+            continue
+    if tar_path is None or not tar_path.is_file():
+        raise FileNotFoundError(
+            f"Could not download Videos.tar from Hugging Face dataset {rid!r} (tried Videos.tar / videos.tar)."
+        )
+
+    return _extract_daily_omni_videos_tar(tar_path, rid)
+
+
+def _resolve_hf_cache_snapshot(path: Path) -> Path:
+    """Map a HF hub cache dir (``datasets--org--name``) to its checked-out snapshot dir.
+
+    Any other directory is returned unchanged.
+    """
+    snapshots = path / "snapshots"
+    if not snapshots.is_dir():
+        return path
+    ref = path / "refs" / "main"
+    if ref.is_file():
+        try:
+            target = snapshots / ref.read_text(encoding="utf-8").strip()
+        except OSError:
+            target = snapshots
+        if target.is_dir():
+            return target
+    revisions = [p for p in snapshots.iterdir() if p.is_dir()]
+    if revisions:
+        return max(revisions, key=lambda p: p.stat().st_mtime)
+    return path
+
+
+def resolve_daily_omni_local_root(dataset_path: str | None) -> Path | None:
+    """Return the local Daily-Omni mirror root for ``dataset_path``, or ``None`` if not local.
+
+    Accepts a plain directory (``qa.json`` + ``Videos/`` or ``Videos.tar``, i.e. the layout of the
+    Hub repo) as well as a HuggingFace hub cache directory (``datasets--liarliar--Daily-Omni``),
+    which is resolved to its snapshot. Hub ids such as ``liarliar/Daily-Omni`` return ``None``.
+    """
+    raw = (dataset_path or "").strip()
+    if not raw:
+        return None
+    path = Path(raw).expanduser()
+    if not path.is_dir():
+        return None
+    return _resolve_hf_cache_snapshot(path.resolve())
+
+
+def daily_omni_local_qa_json(root: Path) -> Path | None:
+    """Locate ``qa.json`` inside a local Daily-Omni mirror (``None`` when absent)."""
+    for name in ("qa.json", "QA.json"):
+        candidate = root / name
+        if candidate.is_file():
+            return candidate
+    return None
+
+
+def daily_omni_local_videos_dir(root: Path) -> Path | None:
+    """Return the ``Videos`` root of a local mirror, extracting ``Videos.tar`` when needed."""
+    for name in ("Videos", "videos"):
+        candidate = root / name
+        if candidate.is_dir():
+            return candidate
+    for name in ("Videos.tar", "videos.tar"):
+        tar_path = root / name
+        if tar_path.is_file():
+            return _extract_daily_omni_videos_tar(tar_path, f"local__{root.name}")
+    return None
+
+
 class _ListDatasetIterator:
     """Simple iterator wrapper around a list to mimic HuggingFace streaming dataset behavior."""
 
@@ -243,6 +309,73 @@ def _uniform_sample_indices(n: int, k: int) -> list[int]:
         return [int(round(i * (n - 1) / (k - 1))) for i in range(k)]
 
 
+def _probe_video_frames_and_fps(video_path: Path) -> tuple[int, float]:
+    """Frame count and average FPS, replacing ``len(vr)`` / ``vr.get_avg_fps()``.
+
+    PyAV is used instead of ``decord`` because ``decord`` ships x86_64-only Linux wheels
+    and is unmaintained, while ``av`` is already a first-class requirement and has
+    aarch64 wheels (Ascend hosts).
+    """
+    import av
+
+    with av.open(str(video_path)) as container:
+        stream = container.streams.video[0]
+        avg_fps = float(stream.average_rate) if stream.average_rate else 0.0
+        num_frames = stream.frames or 0
+        if num_frames <= 0:
+            # Containers such as raw webm leave ``frames`` at 0; derive it from duration.
+            duration: float | None = None
+            if stream.duration is not None and stream.time_base is not None:
+                duration = float(stream.duration * stream.time_base)
+            elif container.duration is not None:
+                duration = container.duration / av.time_base
+            if duration and avg_fps:
+                num_frames = int(math.floor(duration * avg_fps))
+        if num_frames <= 0:
+            # Last resort for streams with no usable metadata at all.
+            stream.thread_type = "AUTO"
+            num_frames = sum(1 for _ in container.decode(stream))
+    if num_frames <= 0:
+        raise ValueError(f"No decodable video frames in {video_path}")
+    return num_frames, avg_fps or 1.0
+
+
+def _decode_video_frames_rgb(video_path: Path, frame_idx: list[int]) -> list[Any]:
+    """Decode presentation-order ``frame_idx`` into RGB ``PIL.Image``s.
+
+    Mirrors ``decord.VideoReader.get_batch(frame_idx)``: one image per requested index,
+    in the requested order, duplicates repeated. Decoding is a single forward pass
+    (no seeking) so the mapping from index to picture matches decord's indexing.
+    """
+    import av
+
+    wanted = sorted({int(i) for i in frame_idx})
+    if not wanted:
+        return []
+
+    decoded: dict[int, Any] = {}
+    with av.open(str(video_path)) as container:
+        stream = container.streams.video[0]
+        stream.thread_type = "AUTO"
+        cursor = 0
+        for position, frame in enumerate(container.decode(stream)):
+            if wanted[cursor] > position:
+                continue
+            image = frame.to_image()
+            # ``<=`` also absorbs indices that overshoot the real frame count.
+            while cursor < len(wanted) and wanted[cursor] <= position:
+                decoded[wanted[cursor]] = image
+                cursor += 1
+            if cursor >= len(wanted):
+                break
+
+    if not decoded:
+        raise ValueError(f"Decoded no frames from {video_path} for indices {wanted[:8]}")
+    # Metadata may promise more frames than the stream actually decodes; clamp to the last one.
+    last_image = decoded[max(decoded)]
+    return [decoded.get(int(i), last_image) for i in frame_idx]
+
+
 def _numpy_to_wav_bytes(audio_np: Any, sample_rate: int = _MINICPM_AUDIO_SR) -> bytes:
     """Encode a mono float/int numpy waveform as 16-bit PCM WAV bytes."""
     import numpy as np
@@ -1029,28 +1162,24 @@ class DailyOmniDataset(BenchmarkDataset):
     ) -> tuple[list[Any], list[Any]]:
         """Port of MiniCPM ``get_video_frame_audio_segments`` (stack_frames=1, 1fps)."""
         import numpy as np
-        from decord import VideoReader, cpu
-        from PIL import Image
         from vllm.multimodal.media.audio import load_audio
 
-        vr = VideoReader(str(video_path), ctx=cpu(0))
-        avg_fps = float(vr.get_avg_fps()) or 1.0
-        duration = len(vr) / avg_fps
+        num_video_frames, avg_fps = _probe_video_frames_and_fps(video_path)
+        duration = num_video_frames / avg_fps
         num_seconds = max(1, int(math.ceil(duration)))
         second_timestamps = list(range(num_seconds))
 
         if duration > max_num_frames:
             timestamps = [round(i * 0.1, 1) for i in range(int(duration / 0.1))]
-            frame_idx = [min(int(ts * avg_fps), len(vr) - 1) for ts in timestamps]
+            frame_idx = [min(int(ts * avg_fps), num_video_frames - 1) for ts in timestamps]
             pick = _uniform_sample_indices(len(frame_idx), max_num_frames)
             frame_idx = [frame_idx[i] for i in pick]
             timestamps = [timestamps[i] for i in pick]
         else:
-            frame_idx = [min(int(i * avg_fps), len(vr) - 1) for i in range(num_seconds)]
+            frame_idx = [min(int(i * avg_fps), num_video_frames - 1) for i in range(num_seconds)]
             timestamps = second_timestamps
 
-        video = vr.get_batch(frame_idx).asnumpy()
-        frames = [Image.fromarray(v.astype("uint8")).convert("RGB") for v in video]
+        frames = _decode_video_frames_rgb(video_path, frame_idx)
 
         audio_segments: list[Any] = []
         if include_audio: