diff --git a/.buildkite/common/ci_mirror_hardwares.yml b/.buildkite/common/ci_mirror_hardwares.yml index 4368e2396..b7d40c75d 100644 --- a/.buildkite/common/ci_mirror_hardwares.yml +++ b/.buildkite/common/ci_mirror_hardwares.yml @@ -8,8 +8,7 @@ # NPU naming: {queue}_{resource_class} # a2b3_npu_1 — ascend-a2b3, resource_class npu-1 (A2/B3 CI image) # a2b3_npu_4 — ascend-a2b3, resource_class npu-4 -# a3_npu_1 — ascend-a3, resource_class npu-1 (A3 CI image tag …:$BUILDKITE_COMMIT-a3) -# a3_npu_2 — ascend-a3, resource_class npu-2 +# a3_npu_2 — ascend-a3, resource_class npu-2 (A3 CI image tag …:$BUILDKITE_COMMIT-a3) x-npu-k8s-plugins: &npu_k8s_plugins - kubernetes: @@ -211,13 +210,6 @@ mirror_hardwares: image: "swr.cn-southwest-2.myhuaweicloud.com/modelfoundry/vllm-omni-ci-npu:${BUILDKITE_COMMIT}" plugins: *npu_k8s_plugins - a3_npu_1: - agents: - queue: "ascend-a3" - resource_class: "npu-1" - image: "swr.cn-southwest-2.myhuaweicloud.com/modelfoundry/vllm-omni-ci-npu:${BUILDKITE_COMMIT}-a3" - plugins: *npu_k8s_plugins - a3_npu_2: agents: queue: "ascend-a3" diff --git a/.buildkite/npu/test-npu-nightly.yml b/.buildkite/npu/test-npu-nightly.yml index ac8123625..4925f84c6 100644 --- a/.buildkite/npu/test-npu-nightly.yml +++ b/.buildkite/npu/test-npu-nightly.yml @@ -41,7 +41,7 @@ steps: if: build.env("NIGHTLY") == "1" || build.pull_request.labels includes "nightly-test" || build.pull_request.labels includes "omni-test" steps: - label: ":full_moon: Omni · MiniCPM-o 4.5 Test" - mirror_hardwares: a3_npu_1 + mirror_hardwares: a3_npu_2 env: VLLM_WORKER_MULTIPROC_METHOD: spawn VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" @@ -50,21 +50,27 @@ steps: - pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5_expansion.py -m 'full_model' --run-level 'full_model' - label: ":full_moon: Omni · MiniCPM-o 4.5 · Duplex Test" timeout_in_minutes: 180 - mirror_hardwares: a3_npu_1 + mirror_hardwares: a3_npu_2 env: VLLM_WORKER_MULTIPROC_METHOD: spawn VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" HF_TOKEN: "${HF_TOKEN}" commands: + # NPU has no merge-level pipeline, so advanced_model duplex coverage + # rides on nightly alongside full_model. + # + # test_duplex_soft_interrupt is CUDA-only and carries no npu mark, so + # the marker filter below drops it: it needs real-time duplex + # throughput that the current NPU stack does not reach. - >- pytest -s -v tests/e2e/online_serving/test_minicpmo_4_5_duplex.py tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py - -m "npu and A3 and omni" + -m "(full_model or advanced_model) and npu and A3 and omni" --run-level full_model - label: ":full_moon: Omni · MiniCPM-o 4.5 · Accuracy Test" timeout_in_minutes: 180 - mirror_hardwares: a3_npu_1 + mirror_hardwares: a3_npu_2 env: VLLM_WORKER_MULTIPROC_METHOD: spawn VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" @@ -72,7 +78,39 @@ steps: commands: - | set +e - pytest -s -v tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py -m "full_model and npu" --run-level full_model + pytest -s -v tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py -m "full_model and npu and A3" --run-level full_model EXIT=$$? buildkite-agent artifact upload "tests/e2e/accuracy/minicpmo_4_5/results/*.json" exit $$EXIT + - label: ":full_moon: Omni · MiniCPM-o 4.5 · Perf Test" + key: nightly-omni-performance-minicpmo-4-5-npu + timeout_in_minutes: 180 + mirror_hardwares: a3_npu_2 + env: + VLLM_WORKER_MULTIPROC_METHOD: spawn + VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" + HF_TOKEN: "${HF_TOKEN}" + commands: + - export BENCHMARK_DIR=tests/dfx/perf/results + - | + set +e + pytest -s -v tests/dfx/perf/scripts/run_benchmark.py -m "npu and A3" --test-config-file tests/dfx/perf/tests/test_minicpmo_4_5.json + EXIT=$$? + buildkite-agent artifact upload "tests/dfx/perf/results/*.json" + exit $$EXIT + - label: ":full_moon: Omni · MiniCPM-o 4.5 · Duplex Seed-TTS Perf Test" + key: nightly-omni-performance-minicpmo-4-5-duplex-seed-tts-npu + timeout_in_minutes: 180 + mirror_hardwares: a3_npu_2 + env: + VLLM_WORKER_MULTIPROC_METHOD: spawn + VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" + HF_TOKEN: "${HF_TOKEN}" + commands: + - export BENCHMARK_DIR=tests/dfx/perf/results + - | + set +e + pytest -s -v tests/dfx/perf/scripts/run_benchmark.py -m "npu and A3" --test-config-file tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json + EXIT=$$? + buildkite-agent artifact upload "tests/dfx/perf/results/*.json" + exit $$EXIT diff --git a/.buildkite/npu/test-npu-ready.yml b/.buildkite/npu/test-npu-ready.yml index eac7f6f06..53d4075b9 100644 --- a/.buildkite/npu/test-npu-ready.yml +++ b/.buildkite/npu/test-npu-ready.yml @@ -4,7 +4,7 @@ steps: depends_on: upload-ready-pipeline steps: - label: "Omni · MiniCPM-o 4.5 Test" - mirror_hardwares: a3_npu_1 + mirror_hardwares: a3_npu_2 env: VLLM_WORKER_MULTIPROC_METHOD: spawn VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" @@ -12,7 +12,7 @@ steps: commands: - pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5.py -m 'core_model and npu and A3 and omni' --run-level 'core_model' - label: "Omni · MiniCPM-o 4.5 Duplex Test" - mirror_hardwares: a3_npu_1 + mirror_hardwares: a3_npu_2 env: VLLM_WORKER_MULTIPROC_METHOD: spawn VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" diff --git a/docker/Dockerfile.npu.ci b/docker/Dockerfile.npu.ci index 341ed5965..b5aff4174 100644 --- a/docker/Dockerfile.npu.ci +++ b/docker/Dockerfile.npu.ci @@ -41,6 +41,9 @@ RUN source /usr/local/Ascend/ascend-toolkit/set_env.sh && \ "hyperpyyaml>=1.2.0" \ "onnx>=1.16.0" \ "librosa>=0.10.0" \ + "jiwer>=4.0.0" \ + "zhon>=2.1.1" \ + "zhconv>=1.4.3" \ --index-url http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \ --trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \ --default-timeout=60 && \ diff --git a/docker/Dockerfile.npu.ci.a3 b/docker/Dockerfile.npu.ci.a3 index 714e6b43c..d50e1d163 100644 --- a/docker/Dockerfile.npu.ci.a3 +++ b/docker/Dockerfile.npu.ci.a3 @@ -41,6 +41,9 @@ RUN source /usr/local/Ascend/ascend-toolkit/set_env.sh && \ "hyperpyyaml>=1.2.0" \ "onnx>=1.16.0" \ "librosa>=0.10.0" \ + "jiwer>=4.0.0" \ + "zhon>=2.1.1" \ + "zhconv>=1.4.3" \ --index-url http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \ --trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \ --default-timeout=60 && \ diff --git a/tests/benchmarks/test_daily_omni_local_source.py b/tests/benchmarks/test_daily_omni_local_source.py new file mode 100644 index 000000000..35045a04b --- /dev/null +++ b/tests/benchmarks/test_daily_omni_local_source.py @@ -0,0 +1,92 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Offline Daily-Omni sources: local mirrors must not fall back to the Hub id.""" + +import json +import os +import tarfile +from pathlib import Path + +import pytest + +from vllm_omni.benchmarks.data_modules.daily_omni_dataset import ( + daily_omni_local_qa_json, + daily_omni_local_videos_dir, + resolve_daily_omni_local_root, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.benchmark] + + +def _write_qa_json(root: Path) -> Path: + qa = root / "qa.json" + qa.write_text(json.dumps([{"Question": "q", "Answer": "A", "video_id": "vid0"}]), encoding="utf-8") + return qa + + +def test_resolve_local_root_returns_none_for_hub_id() -> None: + assert resolve_daily_omni_local_root("liarliar/Daily-Omni") is None + assert resolve_daily_omni_local_root("") is None + assert resolve_daily_omni_local_root(None) is None + + +def test_plain_local_mirror(tmp_path: Path) -> None: + _write_qa_json(tmp_path) + (tmp_path / "Videos" / "vid0").mkdir(parents=True) + + root = resolve_daily_omni_local_root(str(tmp_path)) + assert root == tmp_path.resolve() + assert daily_omni_local_qa_json(root) == (tmp_path / "qa.json").resolve() + assert daily_omni_local_videos_dir(root) == root / "Videos" + + +def test_hf_cache_dir_resolves_to_snapshot(tmp_path: Path) -> None: + cache_dir = tmp_path / "datasets--liarliar--Daily-Omni" + snapshot = cache_dir / "snapshots" / "deadbeef" + snapshot.mkdir(parents=True) + _write_qa_json(snapshot) + (cache_dir / "refs").mkdir() + (cache_dir / "refs" / "main").write_text("deadbeef", encoding="utf-8") + + root = resolve_daily_omni_local_root(str(cache_dir)) + assert root == snapshot.resolve() + assert daily_omni_local_qa_json(root) == (snapshot / "qa.json").resolve() + + +def test_hf_cache_dir_without_refs_picks_newest_snapshot(tmp_path: Path) -> None: + cache_dir = tmp_path / "datasets--liarliar--Daily-Omni" + older = cache_dir / "snapshots" / "aaa" + newer = cache_dir / "snapshots" / "bbb" + older.mkdir(parents=True) + newer.mkdir(parents=True) + os.utime(older, (1, 1)) + os.utime(newer, (10, 10)) + + assert resolve_daily_omni_local_root(str(cache_dir)) == newer.resolve() + + +def test_videos_tar_is_extracted(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("HF_HOME", str(tmp_path / "hf_home")) + root = tmp_path / "mirror" + (root / "Videos" / "vid0").mkdir(parents=True) + (root / "Videos" / "vid0" / "vid0_video.mp4").write_bytes(b"fake") + with tarfile.open(root / "Videos.tar", "w") as tf: + tf.add(root / "Videos", arcname="Videos") + # Only the tarball is left in the mirror. + for child in (root / "Videos").rglob("*"): + if child.is_file(): + child.unlink() + (root / "Videos" / "vid0").rmdir() + (root / "Videos").rmdir() + + videos_dir = daily_omni_local_videos_dir(root) + assert videos_dir is not None + assert (videos_dir / "vid0" / "vid0_video.mp4").is_file() + + # Second call reuses the extraction instead of redoing it. + assert daily_omni_local_videos_dir(root) == videos_dir + + +def test_missing_qa_json_reports_none(tmp_path: Path) -> None: + root = resolve_daily_omni_local_root(str(tmp_path)) + assert daily_omni_local_qa_json(root) is None + assert daily_omni_local_videos_dir(root) is None diff --git a/tests/dfx/perf/tests/test_minicpmo_4_5.json b/tests/dfx/perf/tests/test_minicpmo_4_5.json index 8a21b7971..19b8540d3 100644 --- a/tests/dfx/perf/tests/test_minicpmo_4_5.json +++ b/tests/dfx/perf/tests/test_minicpmo_4_5.json @@ -5,7 +5,8 @@ { "hardware_marks": { "res": { - "cuda": "H100" + "cuda": "H100", + "npu": "A3" }, "num_cards": 1 } @@ -30,14 +31,12 @@ "num_prompts": [ 32, 64, - 128, - 256 + 128 ], "max_concurrency": [ 1, 4, - 8, - 16 + 8 ], "no_oversample": true, "trust_remote_code": true, @@ -55,34 +54,29 @@ "percentile-metrics": "ttft,e2el,audio_ttfp,audio_rtf,audio_duration", "baseline": { "request_throughput": [ - 0.7803, - 1.1008, - 1.2027, - 1.3254 + 0.5383, + 0.6042, + 0.7547 ], "mean_ttft_ms": [ - 166.9764, - 307.7133, - 323.584, - 356.6077 + 333.2633, + 533.8656, + 547.3388 ], "mean_e2el_ms": [ - 1361.1682, - 3800.6332, - 6961.8636, - 12609.4295 + 1857.2154, + 6600.5101, + 10450.6811 ], "mean_audio_ttfp_ms": [ - 635.8099, - 1615.6591, - 2888.9831, - 8320.5422 + 986.4666, + 3411.0754, + 3352.7515 ], "mean_audio_rtf": [ - 0.3151, - 0.8568, - 1.5425, - 2.8224 + 0.4423, + 1.5734, + 2.3024 ] } } diff --git a/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json b/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json index 0df82a8b7..fd609f86e 100644 --- a/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json +++ b/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json @@ -5,7 +5,8 @@ { "hardware_marks": { "res": { - "cuda": "H100" + "cuda": "H100", + "npu": "A3" }, "num_cards": 1 } diff --git a/tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py b/tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py index f917db8e2..941a87c5b 100644 --- a/tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py +++ b/tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py @@ -39,10 +39,7 @@ _MAX_SEED_TTS_MEAN_WER = 0.05 # Match the validated Daily-Omni client body from daily_omni_bench.sh. _DAILY_EXTRA_BODY = { "modalities": ["text"], - "chat_template_kwargs": { - "enable_thinking": False, - "use_tts_template": True, - }, + "chat_template_kwargs": {"enable_thinking": False}, } _SEED_EXTRA_BODY = { "modalities": ["text", "audio"], @@ -151,6 +148,7 @@ def test_minicpmo_4_5_seed_tts_wer_bench(omni_server) -> None: omni_server, skip_seed=False, skip_daily=True, + max_concurrency=4, ) argv.extend( [ diff --git a/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py b/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py index 5fac91ddd..dada64605 100644 --- a/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py +++ b/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py @@ -963,12 +963,23 @@ async def _send_clean_turn( if outcome == "listen": return None, "listen" else: - await _wait_for( - state, - lambda: state.count("response.created") > before_created, - timeout_s=timeout_s, - label=f"{transcript} response.created", - ) + try: + await _wait_for( + state, + lambda: state.count("response.created") > before_created, + timeout_s=timeout_s, + label=f"{transcript} response.created", + ) + except TimeoutError: + # ``auto_response`` sessions only get ``response.created`` once the model + # decides to speak, so report the listen decisions that took its place. + raise TimeoutError( + f"Timed out waiting for {transcript} response.created after " + f"{state.count('input_audio_buffer.committed')} committed input turns; " + f"model_listen={state.model_listen_count - before_model_listen} " + f"buffering_listen={state.buffering_listen_count} " + f"errors={state.count('error')}" + ) from None await _wait_for( state, lambda: eligible_response_id() is not None, diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py index e6359aa49..9e50e8e86 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py +++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py @@ -133,6 +133,10 @@ def test_duplex_single_session_response_required(omni_server, model_prefix: str, output_dir=tmp_path / "single_session", ) args.turns = 2 + # Every turn replays the same active-speech window as the first one. The default + # shorter follow-up window is a different mid-utterance slice, which the native + # duplex model may legitimately answer with "listen" instead of a response. + args.turn_duration_ms = [args.first_turn_ms] * args.turns result = asyncio.run(run_demo(args)) assert result["ok"] is True assert result["audio_delta_count"] > 0 diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py index acec7faec..e37d348a9 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py +++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py @@ -53,7 +53,11 @@ def test_duplex_admission_and_expiry_reaper(omni_server, model_prefix: str, tmp_ assert result["admission"]["overflow_error_code"] == "resource_exhausted" -@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1) +# CUDA-only for now: this contract asserts the model speaks while the user is +# still talking, which only holds when the duplex pipeline sustains real-time +# throughput. The current NPU stack runs several times slower than real time, +# so it never reaches a mid-stream decision point. +@hardware_test(res={"cuda": "H100"}, num_cards=1) @pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True) def test_duplex_soft_interrupt(omni_server, model_prefix: str, tmp_path: Path) -> None: input_wav = validated_soft_interrupt_wav() diff --git a/tests/helpers/minicpmo_4_5_duplex.py b/tests/helpers/minicpmo_4_5_duplex.py index 4362b68d6..ddef6e98d 100644 --- a/tests/helpers/minicpmo_4_5_duplex.py +++ b/tests/helpers/minicpmo_4_5_duplex.py @@ -74,9 +74,11 @@ def validated_soft_interrupt_wav() -> Path: def resolve_ref_audio(model_prefix: str) -> Path: - if model_prefix: - model_root = Path(model_prefix) / MODEL - else: + # ``MODEL`` may be a local checkpoint directory instead of a Hub repo id + # (``Path("/prefix") / "/abs/path"`` collapses to the absolute path), so only + # fall back to the Hub cache when no such directory exists on disk. + model_root = Path(model_prefix) / MODEL if model_prefix else Path(MODEL) + if not model_root.is_dir(): model_root = Path(snapshot_download(MODEL, local_files_only=True)) ref_audio = model_root / REF_AUDIO_RELATIVE_PATH if not ref_audio.is_file(): diff --git a/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py b/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py index fd8f32f30..512e86538 100644 --- a/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py +++ b/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/``. 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/``. 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: diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index cfc5dced3..ccedfb75e 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -14,6 +14,7 @@ import wave from collections.abc import Iterable from dataclasses import dataclass from datetime import datetime +from pathlib import Path from typing import Any, Literal from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -38,7 +39,13 @@ from vllm.logger import init_logger from vllm.tokenizers import TokenizerLike from vllm_omni.benchmarks.audio_continuity import compute_continuity_stats -from vllm_omni.benchmarks.data_modules.daily_omni_dataset import DailyOmniDataset, DailyOmniSampleRequest +from vllm_omni.benchmarks.data_modules.daily_omni_dataset import ( + DailyOmniDataset, + DailyOmniSampleRequest, + daily_omni_local_qa_json, + daily_omni_local_videos_dir, + resolve_daily_omni_local_root, +) from vllm_omni.benchmarks.data_modules.random_multi_modal_dataset import OmniRandomMultiModalDataset from vllm_omni.benchmarks.data_modules.seed_tts_dataset import ( SEED_TTS_DEFAULT_OMNI_SYSTEM_PROMPT, @@ -282,6 +289,26 @@ def get_samples(args, tokenizer): if isinstance(qa_json, str): qa_json = qa_json.strip() or None + # A local mirror of the dataset repo (plain directory or HF hub cache dir) is used + # directly: no Hub round-trip, so offline / air-gapped runs work without extra flags. + local_root: Path | None = None + if qa_json is None: + local_root = resolve_daily_omni_local_root(getattr(args, "dataset_path", None)) or ( + resolve_daily_omni_local_root(getattr(args, "hf_name", None)) + ) + if local_root is not None: + local_qa = daily_omni_local_qa_json(local_root) + if local_qa is not None: + qa_json = str(local_qa) + if video_dir is None: + video_dir = daily_omni_local_videos_dir(local_root) + logger.info("Using local Daily-Omni mirror: root=%s", local_root) + else: + logger.info( + "Local Daily-Omni path %s has no qa.json; loading it with `datasets` instead", + local_root, + ) + if qa_json is not None: logger.info( "Loading Daily-Omni dataset: qa_json=%s, video_dir=%s (Hub not used for QA)", @@ -304,7 +331,8 @@ def get_samples(args, tokenizer): repo_id = _daily_omni_repo_from_args(args) if args.dataset_name == "daily-omni": if repo_id is None: - repo_id = _DEFAULT_DAILY_OMNI_REPO + # Prefer an on-disk copy over the Hub id so offline runs keep working. + repo_id = str(local_root) if local_root is not None else _DEFAULT_DAILY_OMNI_REPO elif repo_id is None: raise ValueError( "Daily-Omni with --dataset-name hf requires " @@ -700,6 +728,29 @@ def _image_generation_ms_from_content(content: Any) -> float: return 0.0 +#: Non-200 responses are otherwise silent (only ``reason`` was kept), which turns a fully failed +#: run into "Successful requests: 0" with no cause. Log the first few bodies, keep all of them on +#: ``output.error`` for ``--save-detailed``. +_HTTP_ERROR_LOG_BUDGET = 5 +_http_error_logged = 0 + + +async def _record_http_error(output: "MixRequestFuncOutput", response: aiohttp.ClientResponse, context: str) -> None: + """Store status + response body on ``output.error`` and log the first few occurrences.""" + global _http_error_logged + + try: + body = (await response.text())[:512] + except Exception: + body = "" + detail = f"HTTP {response.status} {response.reason or ''}".strip() + output.error = f"{detail}: {body}" if body else detail + output.success = False + if _http_error_logged < _HTTP_ERROR_LOG_BUDGET: + _http_error_logged += 1 + logger.warning("%s request failed: %s", context, output.error) + + async def async_request_openai_chat_omni_completions( request_func_input: RequestFuncInput, session: aiohttp.ClientSession, @@ -997,8 +1048,7 @@ async def async_request_openai_chat_omni_completions( logger.warning("seed_tts WER PCM export failed: %s", ex) output.success = True else: - output.error = response.reason or "" - output.success = False + await _record_http_error(output, response, "openai-chat-omni") break except aiohttp.ClientError as e: # transient transport error: may retry @@ -1272,8 +1322,7 @@ async def async_request_openai_audio_speech( ) output.success = True else: - output.error = response.reason or "" - output.success = False + await _record_http_error(output, response, "openai-audio-speech") except Exception: output.success = False output.error = traceback.format_exc()