diff --git a/tests/benchmarks/duplex/test_omni_duplex_eval.py b/tests/benchmarks/duplex/test_omni_duplex_eval.py index 5f77aacc0..f5d3ebcd8 100644 --- a/tests/benchmarks/duplex/test_omni_duplex_eval.py +++ b/tests/benchmarks/duplex/test_omni_duplex_eval.py @@ -64,6 +64,177 @@ def test_hugging_face_subset_names_are_splits(monkeypatch): assert sample.task_type == "correction" +def test_local_hf_dataset_directory_is_routed_to_hf_loader(tmp_path, monkeypatch): + # An existing local Hugging Face dataset layout (directory with a + # ``data/*.parquet`` file) must be loaded through ``datasets.load_dataset`` + # instead of being parsed as a JSON/JSONL manifest. + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + (data_dir / "train-00000-of-00001.parquet").write_bytes(b"PAR1") + calls = [] + + def fake_load_dataset(name, **kwargs): + calls.append((name, kwargs)) + return [{"id": "pr", "split": "PR_correction", "question_text": "Correct this."}] + + monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset)) + samples = load_samples(hf_dir) + assert calls == [(str(hf_dir), {})] + assert len(samples) == 1 + assert samples[0].task_type == "correction" + + +def test_local_parquet_file_is_routed_to_hf_loader(tmp_path, monkeypatch): + parquet = tmp_path / "samples.parquet" + parquet.write_bytes(b"PAR1") + calls = [] + + def fake_load_dataset(name, **kwargs): + calls.append((name, kwargs)) + return [{"id": "pr", "split": "PR_correction", "question_text": "Correct this."}] + + monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset)) + samples = load_samples(str(parquet)) + assert calls == [("parquet", {"data_files": str(parquet)})] + assert len(samples) == 1 + assert samples[0].family == "pr" + + +def test_hf_collapse_to_train_keeps_row_split_identity(tmp_path, monkeypatch): + # datasets.load_dataset collapses a config-less local mirror (a bare data/ + # directory) into a single generic ``train`` split. Rows that already carry + # their own split/subset identity must keep it so family inference is not + # poisoned by the generic name (regression for the direct-read path). + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + (data_dir / "train-00000-of-00021.parquet").write_bytes(b"PAR1") + + def fake_load_dataset(name, **kwargs): + return { + "train": [ + {"id": "pr", "split": "PR_correction", "question_text": "Correct this."}, + {"id": "pr-subset", "subset": "PR_event_reminder", "question_text": "Remind me."}, + {"id": "rtd", "split": "RTD_OCR", "question_text": "Read this."}, + ] + } + + monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset)) + samples = load_samples(hf_dir) + by_id = {sample.id: sample for sample in samples} + assert by_id["pr"].split == "PR_correction" + assert by_id["pr"].family == "pr" + assert by_id["pr"].task_type == "correction" + assert by_id["pr-subset"].split == "PR_event_reminder" + assert by_id["pr-subset"].family == "pr" + assert by_id["pr-subset"].task_type == "proactive_reminder" + assert by_id["rtd"].split == "RTD_OCR" + assert by_id["rtd"].family == "rtd" + + +def test_hf_config_layout_stamps_split_name_when_row_has_none(tmp_path, monkeypatch): + # With a proper config-per-split mirror the loader still stamps the Hugging + # Face split name onto rows that do not identify themselves (unchanged). + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + (data_dir / "RTD_OCR-00000-of-00001.parquet").write_bytes(b"PAR1") + (data_dir / "PR_correction-00000-of-00001.parquet").write_bytes(b"PAR1") + + def fake_load_dataset(name, **kwargs): + return { + "RTD_OCR": [{"id": "rtd", "question_text": "Read this."}], + "PR_correction": [{"id": "pr", "question_text": "Correct this."}], + } + + monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset)) + samples = load_samples(hf_dir) + by_id = {sample.id: sample for sample in samples} + assert by_id["rtd"].split == "RTD_OCR" + assert by_id["rtd"].family == "rtd" + assert by_id["pr"].split == "PR_correction" + assert by_id["pr"].family == "pr" + assert by_id["pr"].task_type == "correction" + + +def test_hf_collapse_to_train_without_row_identity_raises_actionable_error(tmp_path, monkeypatch): + # A row that carries no split/family/task_type and lands in a generic + # collapsed ``train`` split cannot be routed; the loader must raise a clear, + # actionable ValueError instead of the bare family_for_split message. + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + (data_dir / "train-00000-of-00021.parquet").write_bytes(b"PAR1") + + def fake_load_dataset(name, **kwargs): + return {"train": [{"id": "x", "question_text": "What is this?"}]} + + monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset)) + with pytest.raises(ValueError, match="Hugging Face splits"): + load_samples(hf_dir) + + +def test_non_manifest_file_raises_clear_error(tmp_path): + # A plain non-JSON file (e.g. a README) must surface a clear ValueError + # listing the accepted inputs instead of a raw JSONDecodeError. + bad = tmp_path / "README.md" + bad.write_text("# readme\nnot a manifest", encoding="utf-8") + with pytest.raises(ValueError, match="JSON/JSONL manifest"): + load_samples(str(bad)) + + +def _write_parquet(path, rows): + pyarrow = pytest.importorskip("pyarrow") + pyarrow.parquet.write_table(pyarrow.Table.from_pylist(rows), str(path)) + + +def test_local_parquet_split_filters_rows_by_identity(tmp_path): + # Regression: a single ``.parquet`` file is exposed by ``load_dataset`` as a + # generic ``train`` physical split, so ``--split RTD_OCR`` used to raise + # ``Unknown split "RTD_OCR"``. The requested split must instead filter the + # rows by their preserved identity. + pytest.importorskip("datasets") + parquet = tmp_path / "samples.parquet" + _write_parquet( + parquet, + [ + {"id": "rtd", "split": "RTD_OCR", "question_text": "Read this."}, + {"id": "pr", "split": "PR_correction", "question_text": "Correct this."}, + ], + ) + assert {sample.id for sample in load_samples(str(parquet))} == {"rtd", "pr"} + rtd = load_samples(str(parquet), split="RTD_OCR") + assert [sample.id for sample in rtd] == ["rtd"] + assert rtd[0].family == "rtd" + pr = load_samples(str(parquet), split="PR_correction") + assert [sample.id for sample in pr] == ["pr"] + assert pr[0].family == "pr" + assert pr[0].task_type == "correction" + + +def test_local_directory_named_splits_resolve_without_row_identity(tmp_path): + # Regression: a single-configuration mirror whose data files are named after + # the benchmark splits must resolve each split for real (no fake loader) and + # without any per-row identity. + pytest.importorskip("datasets") + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + _write_parquet(data_dir / "RTD_OCR-00000-of-00001.parquet", [{"id": "rtd", "question_text": "Read this."}]) + _write_parquet( + data_dir / "PR_correction-00000-of-00001.parquet", + [{"id": "pr", "question_text": "Correct this."}], + ) + by_id = {sample.id: sample for sample in load_samples(hf_dir)} + assert by_id["rtd"].split == "RTD_OCR" + assert by_id["rtd"].family == "rtd" + assert by_id["pr"].split == "PR_correction" + assert by_id["pr"].family == "pr" + assert by_id["pr"].task_type == "correction" + assert [sample.id for sample in load_samples(hf_dir, split="RTD_OCR")] == ["rtd"] + + def test_response_aliases_and_clock_guard(): assert split_text("One. Two!") == ["One.", "Two!"] assert normalize_response_items({"chunks": [{"text": "x", "current_time": 800}]}) == [ @@ -141,3 +312,113 @@ def test_frame_sample_content_passes_frames(tmp_path, monkeypatch): judge_video_mode="frame-sample", ) assert score["content"]["frame_count"] == 2 + + +def test_local_directory_collapsed_train_with_row_identity_and_split_filter(tmp_path): + # Reviewer-pointed gap: a collapsed local mirror (data/train-*.parquet) + # whose rows carry their own split identity must correctly filter by + # --split RTD_OCR. + pytest.importorskip("datasets") + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + _write_parquet( + data_dir / "train-00000-of-00001.parquet", + [ + {"id": "rtd", "split": "RTD_OCR", "question_text": "Read this."}, + {"id": "pr", "split": "PR_correction", "question_text": "Correct this."}, + ], + ) + all_samples = load_samples(hf_dir) + assert {s.id for s in all_samples} == {"rtd", "pr"} + rtd = load_samples(hf_dir, split="RTD_OCR") + assert [s.id for s in rtd] == ["rtd"] + assert rtd[0].family == "rtd" + + +def test_local_directory_mistyped_split_raises_clear_error(tmp_path): + # A named-shard mirror with --split RTD_TYPO must raise ValueError + # listing the observed splits instead of silently returning nothing. + pytest.importorskip("datasets") + hf_dir = tmp_path / "Omni-DuplexEval" + data_dir = hf_dir / "data" + data_dir.mkdir(parents=True) + _write_parquet( + data_dir / "RTD_OCR-00000-of-00001.parquet", + [{"id": "rtd", "question_text": "Read this."}], + ) + _write_parquet( + data_dir / "PR_correction-00000-of-00001.parquet", + [{"id": "pr", "question_text": "Correct this."}], + ) + with pytest.raises(ValueError) as exc_info: + load_samples(hf_dir, split="RTD_TYPO") + message = str(exc_info.value) + assert "matched zero" in message + assert "RTD_OCR" in message or "PR_correction" in message + + +def test_manifest_without_split_identity_split_override_still_works(tmp_path): + # Manifest rows without split/subset/config identity must keep the + # base override behaviour: --split RTD_OCR stamps the split so rows + # are kept instead of rejected (regression test for P2). + manifest = tmp_path / "samples.json" + manifest.write_text( + json.dumps([{"id": "a", "video": "clip.mp4"}, {"id": "b", "video": "clip2.mp4"}]), + encoding="utf-8", + ) + samples = load_samples(manifest, split="RTD_OCR") + assert len(samples) == 2 + assert all(s.split == "RTD_OCR" for s in samples) + assert all(s.family == "rtd" for s in samples) + + +def test_iterable_without_split_identity_split_override_still_works(): + # Iterable rows without split/subset/config identity must keep the + # base override behaviour (regression test for F5/P2). + data = [{"id": "a", "video": "clip.mp4"}, {"id": "b", "video": "clip2.mp4"}] + samples = load_samples(data, split="RTD_OCR") + assert len(samples) == 2 + assert all(s.split == "RTD_OCR" for s in samples) + assert all(s.family == "rtd" for s in samples) + + +def test_audio_feature_decode_false_preserves_bytes_key(tmp_path): + # When a parquet file carries an Audio feature column, the loader + # must recast it with decode=False so rows contain the ``bytes`` key + # instead of ``array``/``sampling_rate``. + pytest.importorskip("datasets") + pytest.importorskip("pyarrow") + import datasets + + audio_bytes = b"\x00\x01\x02" + ds = datasets.Dataset.from_dict( + { + "id": ["a"], + "question_text": ["Hello"], + "task_type": ["pr_correction"], + "audio": [{"bytes": audio_bytes, "path": None}], + }, + features=datasets.Features( + { + "id": datasets.Value("string"), + "question_text": datasets.Value("string"), + "task_type": datasets.Value("string"), + "audio": datasets.Audio(sampling_rate=16000), + } + ), + ) + parquet = tmp_path / "audio_sample.parquet" + ds.to_parquet(str(parquet)) + + samples = load_samples(str(parquet)) + assert len(samples) == 1 + raw = samples[0].raw + assert raw is not None + audio_val = raw.get("audio") + assert audio_val is not None + # The Audio feature was decoded with decode=False, so the struct + # keeps the ``bytes`` key that materialize_media needs. + assert isinstance(audio_val, dict) and "bytes" in audio_val + assert audio_val["bytes"] == audio_bytes + assert "path" in audio_val diff --git a/tests/benchmarks/duplex/test_omni_duplex_eval_media.py b/tests/benchmarks/duplex/test_omni_duplex_eval_media.py new file mode 100644 index 000000000..28a502428 --- /dev/null +++ b/tests/benchmarks/duplex/test_omni_duplex_eval_media.py @@ -0,0 +1,67 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from __future__ import annotations + +import pytest + +from vllm_omni.benchmarks.duplex.omni_duplex_eval_media import materialize_media + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +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 + # does not exist on disk), so downstream ffmpeg saw a missing file. Inline + # bytes must win and be written to a real file whose content equals bytes. + payload = b"\x00\xff\x01RIFFfake-content" + resolved = materialize_media({"bytes": payload, "path": "question_audio.wav"}, tmp_path, "s1_question", ".wav") + assert resolved.is_file() + assert resolved.read_bytes() == payload + assert resolved.parent == tmp_path + assert resolved.name == "s1_question.wav" + + +def test_dict_extension_comes_from_path_safely(tmp_path): + payload = b"flac-ish-bytes" + resolved = materialize_media({"bytes": payload, "path": "audio/clip.FLAC"}, tmp_path, "s1", ".wav") + assert resolved.name == "s1.flac" + assert resolved.read_bytes() == payload + # A path with no usable extension falls back to the caller's suffix. + resolved = materialize_media({"bytes": payload, "path": "audio/clip"}, tmp_path, "s2", ".wav") + assert resolved.name == "s2.wav" + assert resolved.read_bytes() == payload + + +def test_materialization_is_idempotent_and_updated(tmp_path): + payload = b"abc" + first = materialize_media({"bytes": payload, "path": "q.wav"}, tmp_path, "s", ".wav") + second = materialize_media({"bytes": payload, "path": "q.wav"}, tmp_path, "s", ".wav") + assert first == second + assert first.read_bytes() == payload + replacement = b"xyz" + third = materialize_media({"bytes": replacement, "path": "q.wav"}, tmp_path, "s", ".wav") + assert third == first + assert third.read_bytes() == replacement + + +def test_str_path_and_path_only_mapping_returned_as_is(tmp_path): + existing = tmp_path / "clip.mp4" + existing.write_bytes(b"video") + assert materialize_media(str(existing), tmp_path, "s", ".mp4") == existing + assert materialize_media({"path": str(existing)}, tmp_path, "s", ".mp4") == existing + + +def test_raw_bytes_value_is_written(tmp_path): + payload = b"raw-media" + resolved = materialize_media(payload, tmp_path, "r", ".bin") + assert resolved.name == "r.bin" + assert resolved.read_bytes() == payload + + +def test_unresolvable_value_raises_clear_error(tmp_path): + with pytest.raises(ValueError, match="cannot materialize"): + materialize_media(None, tmp_path, "s", ".wav") + with pytest.raises(ValueError, match="cannot materialize"): + materialize_media({"path": None}, tmp_path, "s", ".wav") diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py index 9d6013e80..349d80fc3 100644 --- a/vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py +++ b/vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py @@ -3,15 +3,36 @@ """Dataset and split normalization for Omni-DuplexEval. -The loader accepts a Hugging Face dataset id, a JSON/JSONL manifest, or an +The loader accepts a Hugging Face dataset id, a local Hugging Face dataset +layout (a directory or a ``.parquet`` file), a JSON/JSONL manifest, or an already materialized iterable. Media is deliberately resolved at use time so the benchmark remains usable in air-gapped environments. + +Local directory mirrors are expected to use a *single* Hugging Face +configuration whose data files are named after the benchmark splits +(``data/-00000-of-00001.parquet``, with ```` one of the RTD_*/PR_* +names), so ``datasets.load_dataset`` exposes one physical split per benchmark +split. A single ``.parquet`` file and a mirror that Hugging Face collapses into +a generic ``train`` split are also accepted, but then the requested ``split`` is +applied as a filter on each row's preserved ``split``/``subset``/``config`` +identity instead of as a physical split name. Rows that identify neither their +split nor their family/task raise a clear error that explains the required +layout. + +A requested ``split`` that matches zero rows (for example a mistyped name) is +reported loudly: :func:`load_samples` raises a ``ValueError`` that lists the +split identities actually observed in the loaded rows instead of silently +returning an empty selection. Rows coming from a JSON/JSONL manifest or an +already materialized iterable that carry no split/subset/config identity keep +the base override behavior: ``--split`` is stamped onto them so they are kept +rather than rejected. """ from __future__ import annotations +import importlib import json -from collections.abc import Iterable +from collections.abc import Callable, Iterable from dataclasses import dataclass from pathlib import Path from typing import Any @@ -68,6 +89,23 @@ def family_for_split(split: str, task_type: str | None = None) -> str: raise ValueError(f"cannot infer benchmark family from split {split!r}") +def _identity_error(split: str, sample_id: str, *, family: bool) -> str: + """Build an actionable error for a row whose family/task cannot be derived.""" + subject = "family" if family else "proactive-reminder task type" + label = f" for sample {sample_id!r}" if sample_id else "" + detail = ( + "the split is not an RTD_*/PR_* name and no per-row 'task_type' is present" + if family + else "no per-row 'task_type' is present" + ) + return ( + f"cannot infer the Omni-DuplexEval {subject}{label} from split {split!r}: {detail}. " + "Load a local Omni-DuplexEval mirror whose data files are named after the " + "Hugging Face splits (RTD_*/PR_*) or add per-row " + "'family'/'task_type'/'split' columns." + ) + + def _value(row: dict[str, Any], *keys: str, default: Any = None) -> Any: for key in keys: if key in row and row[key] is not None: @@ -100,14 +138,27 @@ class DuplexSample: raw: dict[str, Any] | None = None @classmethod - def from_row(cls, row: dict[str, Any], *, split: str | None = None, media_root: Path | None = None) -> DuplexSample: - chosen_split = str(split or _value(row, "split", "subset", "config", default="")) + def from_row(cls, row: dict[str, Any], *, media_root: Path | None = None) -> DuplexSample: + chosen_split = str(_value(row, "split", "subset", "config", default="")) + sample_id = str(_value(row, "id", "sample_id", "uid", "name", default="")) task_value = _value(row, "task_type", "task", "type") - family = str(_value(row, "family", default="") or family_for_split(chosen_split, task_value)) + family_value = _value(row, "family", default="") + if family_value: + family = str(family_value) + else: + try: + family = family_for_split(chosen_split, task_value) + except ValueError as exc: + raise ValueError(_identity_error(chosen_split, sample_id, family=True)) from exc task = None if family == "pr": - task = canonical_task_type(str(task_value)) if task_value else task_type_for_split(chosen_split) - sample_id = str(_value(row, "id", "sample_id", "uid", "name", default="")) + if task_value: + task = canonical_task_type(str(task_value)) + else: + try: + task = task_type_for_split(chosen_split) + except ValueError as exc: + raise ValueError(_identity_error(chosen_split, sample_id, family=False)) from exc video = _value(row, "video", "video_path", "video_file") audio = _value(row, "question_audio", "audio", "question_wav") if media_root: @@ -148,6 +199,150 @@ def _read_manifest(path: Path) -> list[dict[str, Any]]: return payload +def _has_split_identity(row: dict[str, Any]) -> bool: + """Whether a row already identifies the benchmark split it belongs to.""" + return any(row.get(key) for key in ("split", "subset", "config")) + + +def _stamp_split(rows: Iterable[dict[str, Any]], name: str) -> list[dict[str, Any]]: + """Stamp a physical Hugging Face split name onto rows that lack identity. + + Hugging Face collapses a config-less local mirror (a bare ``data/`` + directory) or a single ``.parquet`` file into a single generic split + (``train``). A row's own split identity (``split``/``subset``/``config``) + must win so family inference is not poisoned by that generic name; the + physical split name is only stamped when the row does not identify itself. + """ + stamped: list[dict[str, Any]] = [] + for row in rows: + item = dict(row) + if name and not _has_split_identity(item): + item["split"] = name + stamped.append(item) + return stamped + + +def _override_split(rows: list[dict[str, Any]], split: str | None) -> list[dict[str, Any]]: + """Apply the base split-override semantics to manifest/iterable rows. + + Rows that lack split/subset/config identity are stamped with the requested + split so they are kept, matching the pre-override behavior; rows that + already carry identity are left untouched so the requested split stays a + filter. Copies are returned, never mutating the caller's original dicts. + """ + if not split or split == "all": + return rows + return [row if _has_split_identity(row) else {**row, "split": split} for row in rows] + + +def _recast_media_features(loaded: Any) -> Any: + """Recast Audio/Video features to decode=False so rows carry {bytes, path}. + + Hugging Face ``Audio``/``Video`` features default to ``decode=True``, + producing ``{"array", "path", "sampling_rate"}`` dicts without the + ``bytes`` key that :func:`materialize_media` needs. ``decode=False`` keeps + the raw parquet struct instead. Must run before row iteration + (``cast_column`` is lazy; decoding happens at row access). + """ + try: + features_mod = importlib.import_module("datasets.features") + except ImportError: + return loaded + audio_cls = getattr(features_mod, "Audio", None) + video_cls = getattr(features_mod, "Video", None) + if audio_cls is None: + return loaded + + if isinstance(loaded, dict): + # A DatasetDict -- or a plain dict returned by a fake loader whose + # values are lists -- recurses one level and is returned untouched by + # the features guard below. + return {name: _recast_media_features(table) for name, table in loaded.items()} + + feats = getattr(loaded, "features", None) + if feats is None: + return loaded + for column, feature in feats.items(): + if isinstance(feature, audio_cls): + loaded = loaded.cast_column(column, audio_cls(decode=False)) + elif video_cls is not None and isinstance(feature, video_cls): + loaded = loaded.cast_column(column, video_cls(decode=False)) + return loaded + + +def _read_hf(load_dataset: Callable[..., Any], dataset: str, want: str | None) -> Any: + """Call ``datasets.load_dataset`` for a local layout or a remote id. + + The physical split is only requested when one was asked for; ``datasets`` + then returns a single ``Dataset`` (an iterable of rows) instead of a + ``DatasetDict`` (a mapping of split name to such an iterable). The result is + normalized by :func:`_rows_from_hf`. + """ + if Path(dataset).suffix.lower() == ".parquet": + # A single parquet file is always exposed as one generic ``train`` + # split; the logical benchmark split is filtered from row identity + # afterwards. The ``want`` branch was unreachable (the only call site + # passes ``None``) and implied physical split support that does not + # exist, so it was removed. + return load_dataset("parquet", data_files=dataset) + if want: + return load_dataset(dataset, split=want) + return load_dataset(dataset) + + +def _rows_from_hf(dataset: str, *, split: str | None) -> list[dict[str, Any]]: + """Load rows from a Hugging Face dataset id or a local dataset layout. + + ``datasets.load_dataset`` accepts a remote id, a local dataset directory + (``data/-00000-of-00001.parquet``) and, via the ``parquet`` builder, + a single ``.parquet`` file, which keeps the benchmark usable offline with a + locally mirrored copy of the dataset. + + The logical benchmark split (``RTD_OCR``, ``PR_correction``, ...) is not + forwarded blindly as a physical Hugging Face split: a single ``.parquet`` + file and a collapsed local mirror only expose a generic ``train`` split, so + passing the logical name there raises ``Unknown split``. The physical + dataset is loaded instead and the requested split is applied later, on the + preserved per-row identity (see :func:`load_samples`). + """ + try: + from datasets import load_dataset + except ImportError as exc: + raise RuntimeError( + "datasets is required to load a Hugging Face dataset id or a local " + "dataset directory/parquet file; use a local JSON/JSONL manifest instead" + ) from exc + + requested = split if split and split != "all" else None + is_local = Path(dataset).exists() + if Path(dataset).suffix.lower() == ".parquet": + # A single parquet file is always exposed as one generic ``train`` + # split; the logical benchmark split is filtered from row identity + # afterwards rather than forwarded as a physical split name. + loaded = _read_hf(load_dataset, dataset, None) + else: + try: + loaded = _read_hf(load_dataset, dataset, requested) + except ValueError as exc: + # A collapsed *local* mirror (bare ``data/`` directory) does not + # expose the logical split as a physical Hugging Face split; fall + # back to the whole physical dataset and filter by row identity. + # A remote id keeps the original error so a mistyped split still + # fails loudly instead of silently selecting nothing. + if not is_local or requested is None or "Unknown split" not in str(exc): + raise + loaded = _read_hf(load_dataset, dataset, None) + requested = None + + loaded = _recast_media_features(loaded) + if isinstance(loaded, dict): + rows: list[dict[str, Any]] = [] + for name, table in loaded.items(): + rows.extend(_stamp_split((dict(row) for row in table), name)) + return rows + return _stamp_split((dict(row) for row in loaded), requested or "") + + def load_samples( dataset: str | Path | Iterable[dict[str, Any]] = DEFAULT_DATASET, *, @@ -157,37 +352,54 @@ def load_samples( limit: int | None = None, ids: Iterable[str] | None = None, ) -> list[DuplexSample]: - if isinstance(dataset, (str, Path)) and Path(str(dataset)).exists(): - rows = _read_manifest(Path(str(dataset))) - elif not isinstance(dataset, (str, Path)): - rows = list(dataset) - else: - try: - from datasets import load_dataset - except ImportError as exc: - raise RuntimeError("datasets is required for Hugging Face loading; use a local manifest instead") from exc - kwargs: dict[str, Any] = {} - if split and split != "all": - kwargs["split"] = split - loaded = load_dataset(str(dataset), **kwargs) - if isinstance(loaded, dict): - rows = [] - for name, table in loaded.items(): - rows.extend({**dict(row), "split": name} for row in table) + if isinstance(dataset, str | Path): + path = Path(str(dataset)) + if path.exists(): + # A directory or a ``.parquet`` file is a Hugging Face dataset + # layout (e.g. a locally mirrored Omni-DuplexEval snapshot); only + # single JSON/JSONL files are parsed as manifests. Routing by path + # shape instead of mere existence lets local Hugging Face datasets + # load through ``datasets.load_dataset``. + if path.is_dir() or path.suffix.lower() == ".parquet": + rows = _rows_from_hf(str(path), split=split) + else: + try: + rows = _override_split(_read_manifest(path), split) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise ValueError( + f"cannot load dataset from {str(path)!r}: expected a " + "JSON/JSONL manifest file, a local Hugging Face dataset " + "directory, or a single .parquet file" + ) from exc else: - rows = [dict(row) for row in loaded] + rows = _rows_from_hf(str(dataset), split=split) + else: + rows = _override_split(list(dataset), split) wanted = set(str(item) for item in ids) if ids else None root = Path(media_root).expanduser() if media_root else None result = [] + split_matched = 0 for row in rows: - sample = DuplexSample.from_row(row, split=None if split in (None, "all") else split, media_root=root) + # The requested split is a *filter* over each row's own split identity; + # never pass it as an override, which would relabel every row as the + # requested split (turning a collapsed local dataset into a full match + # instead of a filter). + sample = DuplexSample.from_row(row, media_root=root) if split and split != "all" and sample.split != split: continue + split_matched += 1 if family != "all" and sample.family != family: continue if wanted is not None and sample.id not in wanted: continue result.append(sample) + if split and split != "all" and split_matched == 0 and rows: + observed = sorted({str(_value(row, "split", "subset", "config", default="")) or "" for row in rows}) + raise ValueError( + f"--split {split!r} matched zero samples. " + f"Split identities observed in the loaded rows: {observed}. " + "Check the split name for a typo." + ) if limit is not None: result = result[: max(0, limit)] return result diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py index 629d8e877..b45686fb8 100644 --- a/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py +++ b/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py @@ -5,7 +5,9 @@ from __future__ import annotations +import os import subprocess +import tempfile import wave from collections.abc import Iterator from pathlib import Path @@ -20,20 +22,57 @@ def _run_ffmpeg(args: list[str]) -> bytes: def materialize_media(value: Any, output_dir: str | Path, stem: str, suffix: str) -> Path: - """Resolve common Hugging Face media values to a local file.""" + """Resolve common Hugging Face media values to a local file. + + ``str``/``Path`` values are returned as-is. Inline byte payloads (a + ``bytes``/``bytearray`` or a Hugging Face ``Audio``/``Video``/``Image`` + feature mapping ``{"bytes": ..., "path": ...}``) are written to a real file + under ``output_dir`` and that existing path is returned. For such mappings + the ``path`` key is the original artifact name inside the dataset, which is + usually *not* present on the local disk, so it is only used to pick a + content extension and never treated as an existing file. A mapping or + object carrying only a ``path`` (no inline ``bytes``) is returned as an + already-materialized file reference. + """ if isinstance(value, str | Path): return Path(value) path = value.get("path") if isinstance(value, dict) else getattr(value, "path", None) - if path: - return Path(path) - payload = value.get("bytes") if isinstance(value, dict) else None + payload = value.get("bytes") if isinstance(value, dict) else getattr(value, "bytes", None) if payload is None and isinstance(value, bytes | bytearray): payload = value - if payload is None: - raise ValueError(f"cannot materialize media value for {stem!r}") - destination = Path(output_dir) / f"{stem}{suffix}" - destination.parent.mkdir(parents=True, exist_ok=True) - destination.write_bytes(bytes(payload)) + if payload is not None: + return _write_media_bytes(payload, output_dir, stem, _media_extension(path, suffix)) + if path: + return Path(path) + raise ValueError(f"cannot materialize media value for {stem!r}") + + +def _media_extension(path: str | Path | None, fallback: str) -> str: + """Pick a safe lower-case file extension from ``path`` when available.""" + extension = Path(str(path)).suffix.lower() if path else "" + if extension.startswith(".") and len(extension) <= 8 and extension[1:].isalnum(): + return extension + return fallback + + +def _write_media_bytes(payload: bytes | bytearray, output_dir: str | Path, stem: str, extension: str) -> Path: + data = bytes(payload) + root = Path(output_dir) + root.mkdir(parents=True, exist_ok=True) + destination = root / f"{stem}{extension}" + if destination.exists() and destination.stat().st_size == len(data) and destination.read_bytes() == data: + return destination + fd, temporary = tempfile.mkstemp(dir=root, prefix=f".{destination.name}.", suffix=".part") + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + os.replace(temporary, destination) + except OSError: + try: + os.unlink(temporary) + except OSError: + pass + raise return destination diff --git a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py index 748100c5d..c3e55abb7 100644 --- a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py +++ b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py @@ -28,8 +28,28 @@ from vllm_omni.entrypoints.cli.benchmark.base import OmniBenchmarkSubcommandBase def _common(parser: argparse.ArgumentParser) -> None: - parser.add_argument("--dataset", default=DEFAULT_DATASET) - parser.add_argument("--split", default="all") + parser.add_argument( + "--dataset", + default=DEFAULT_DATASET, + help=( + "Hugging Face dataset id, a JSON/JSONL manifest, or a local Hugging Face " + "dataset mirror. A local directory must use a single configuration whose " + "data files are named after the RTD_*/PR_* splits " + "(data/-00000-of-00001.parquet); a single .parquet file is also " + "accepted. --split filters the rows by their preserved split identity, so " + "rows without identity fail loudly." + ), + ) + parser.add_argument( + "--split", + default="all", + help=( + "Restrict to one split (e.g. RTD_OCR) or 'all'. A mistyped split name " + "raises a clear error listing the splits actually observed in the data. " + "Rows from a manifest or an iterable without split/subset/config identity " + "keep the requested split as an override rather than being rejected." + ), + ) parser.add_argument("--family", choices=("all", "rtd", "pr"), default="all") parser.add_argument("--media-root") parser.add_argument("--limit", type=int) @@ -100,6 +120,8 @@ def run(args: argparse.Namespace) -> int: limit=args.limit, ids=args.ids, ) + if not samples: + raise ValueError("no samples selected; check --dataset/--split/--family/--ids/--limit") if args.action == "generate": if args.concurrency < 1: raise ValueError("--concurrency must be at least 1")