"""Explicit-path evaluation wrapper for the paper benchmark suite.""" from __future__ import annotations import argparse import json import os import re import shlex import subprocess from datetime import datetime, timezone from importlib.metadata import PackageNotFoundError, distribution from importlib.util import find_spec from pathlib import Path, PurePosixPath from typing import Any, Mapping, Sequence import yaml from orarl.evaluation import ( EvaluationSchemaError, ManifestError, StagingError, evaluation_asset_paths, load_asset_manifest, load_dataset_manifest, load_evaluation_jsonl, validate_asset_manifest, validate_dataset_manifest, ) VIDEO_QA_TASKS = ( "videomme", "videommev2", "mvbench", "mmvu", "videoholmes", "longvideobench", "mlvu", ) PAPER_TASKS = ( *VIDEO_QA_TASKS, "vsi", "mmsi", "mindcube", "revsi", "spatial_grounding", "tracking", "stvg", "temporal_grounding", "segmentation", ) _TASK_PREFIX = { **{task: task.upper() for task in VIDEO_QA_TASKS}, "vsi": "VSI", "mmsi": "MMSI", "mindcube": "MINDCUBE", "revsi": "REVSI", "spatial_grounding": "SPATIAL_GROUNDING", "tracking": "TRACKING", "stvg": "STVG", "temporal_grounding": "TIMELENS", "segmentation": "SEGMENTATION", } _REQUIRED_INPUTS = { "videomme": {"VIDEOMME_DATA_FILE", "VIDEOMME_VIDEO_BASE"}, "videommev2": {"VIDEOMMEV2_DATA_FILE", "VIDEOMMEV2_VIDEO_BASE"}, "mvbench": {"MVBENCH_DATA_FILE", "MVBENCH_VIDEO_ROOT"}, "mmvu": {"MMVU_DATA_FILE", "MMVU_VIDEO_ROOT"}, "videoholmes": {"VIDEOHOLMES_DATA_FILE", "VIDEOHOLMES_VIDEO_ROOT"}, "longvideobench": {"LONGVIDEOBENCH_DATA_FILE", "LONGVIDEOBENCH_VIDEO_ROOT"}, "mlvu": {"MLVU_DATA_FILE", "MLVU_VIDEO_ROOT"}, "vsi": {"VSI_DATA_FILE", "VSI_PREPROCESSED_VIDEO_DIR"}, "mmsi": {"MMSI_DATA_FILE"}, "mindcube": {"MINDCUBE_DATA_FILE"}, "revsi": {"REVSI_DATA_FILE", "REVSI_VIDEO_ROOT"}, "spatial_grounding": { "SPATIAL_GROUNDING_BENCH_DIR", "SPATIAL_GROUNDING_IMAGE_ROOT", }, "tracking": {"TRACKING_BENCH_DIR", "TRACKING_BASE_PREFIX"}, "stvg": {"STVG_BENCH_DIR", "STVG_BASE_PREFIX"}, "temporal_grounding": {"TIMELENS_BENCH_DIR"}, "segmentation": {"SEGMENTATION_BENCH_DIR", "SEGMENTATION_DATA_ROOT"}, } _PATH_ENV_SUFFIXES = ( "_FILE", "_DIR", "_ROOT", "_PATH", "_CKPT", "_CFG", "_BASE", "_PREFIX", ) _STRUCTURED_TASKS = frozenset( { "spatial_grounding", "tracking", "stvg", "temporal_grounding", "segmentation", } ) _SPATIAL_GROUNDING_SPLIT_NAMES = { "refcoco_val": "refcoco-val", "refcoco_test_a": "refcoco-testA", "refcoco_test_b": "refcoco-testB", "refcocop_val": "refcoco+-val", "refcocop_test_a": "refcoco+-testA", "refcocop_test_b": "refcoco+-testB", "refcocog_val": "refcocog-val", "refcocog_test": "refcocog-test", } _TRACKING_SPLIT_NAMES = { "got10k": "eval_got10k", } _STVG_SPLIT_NAMES = { "stvg": "eval_stvg", } def _evaluator_split_name(task: str, split: object) -> str: value = str(split) if task == "spatial_grounding": return _SPATIAL_GROUNDING_SPLIT_NAMES.get(value, value) if task == "tracking": return _TRACKING_SPLIT_NAMES.get(value, value) if task == "stvg": return _STVG_SPLIT_NAMES.get(value, value) return value _MANIFEST_FILES = ("datasets.jsonl", "assets.jsonl") _INDEX_MANIFEST_FILES = ("datasets.jsonl",) _SENSITIVE_NAME = re.compile( r"(?:^|_)(?:api_?key|credential|password|private_?key|secret|token)(?:$|_)", flags=re.IGNORECASE, ) _URI_USERINFO = re.compile(r"://[^/\s:@]+:[^@\s/]+@") class CliError(ValueError): """Raised for invalid evaluation requests.""" def create_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="orarl-eval", description=( "Run the repository evaluator with explicit benchmark inputs and " "write one aggregate JSON summary. The default is a dry run." ), ) parser.add_argument("--model", required=True, help="Local model or checkpoint path.") parser.add_argument( "--tasks", required=True, help=( "Comma-separated paper tasks. Use video_qa for the documented Video QA " "group or paper for the full suite." ), ) parser.add_argument( "--dataset", help=( "Canonical local staging root or Hugging Face dataset repository id. " "This is the preferred evaluation-data input." ), ) parser.add_argument( "--revision", default="main", help="Hugging Face dataset revision (default: main).", ) parser.add_argument( "--splits", help="Optional comma-separated canonical split ids to evaluate.", ) parser.add_argument( "--asset-root", help=( "Optional local root containing canonical media/ and artifacts/ " "when JSONL metadata is stored separately." ), ) parser.add_argument("--cache-dir", help="Optional Hugging Face snapshot cache directory.") parser.add_argument( "--local-files-only", action="store_true", help="Resolve a Hugging Face dataset only from the local cache.", ) parser.add_argument( "--data-root", help=( "Legacy root containing one directory per task. Cannot be combined " "with --dataset." ), ) parser.add_argument( "--task-config", action="append", default=[], metavar="TASK=FILE", help="YAML/JSON environment mapping for one task; repeat as needed.", ) parser.add_argument( "--evaluator", help="Path to eval/task/eval.sh (normally discovered from the source checkout).", ) parser.add_argument("--gpus", default="0", help="Comma-separated local GPU ids.") parser.add_argument("--tp-size", type=int, default=1, help="Tensor parallel size.") parser.add_argument( "--batch-size", type=int, help="Optional per-task batch-size override for this launch.", ) parser.add_argument( "--max-samples", type=int, help=( "Optional sample limit for supported task smoke tests." ), ) parser.add_argument("--force-merge", action="store_true") parser.add_argument("--skip-merge", action="store_true") parser.add_argument("--merged-model") parser.add_argument("--base-model") parser.add_argument("--segmentation-run-sam2", action="store_true") parser.add_argument( "--summary", default="orarl-eval-summary.json", help="Aggregate JSON output path.", ) parser.add_argument( "--results-root", help="Directory to scan for evaluator summary.json files.", ) mode = parser.add_mutually_exclusive_group() mode.add_argument("--dry-run", dest="dry_run", action="store_true") mode.add_argument("--run", dest="dry_run", action="store_false") parser.set_defaults(dry_run=True) return parser def _existing(value: str, label: str, *, directory: bool | None = None) -> Path: path = Path(value).expanduser().resolve() if not path.exists(): raise CliError(f"{label} does not exist: {path}") if directory is True and not path.is_dir(): raise CliError(f"{label} must be a directory: {path}") if directory is False and not path.is_file(): raise CliError(f"{label} must be a file: {path}") return path def _parse_tasks(value: str) -> list[str]: requested = [item.strip().casefold() for item in value.split(",") if item.strip()] if not requested: raise CliError("--tasks cannot be empty") expanded: list[str] = [] for task in requested: values = ( PAPER_TASKS if task == "paper" else VIDEO_QA_TASKS if task == "video_qa" else (task,) ) for value_item in values: if value_item not in PAPER_TASKS: raise CliError( f"unsupported task {value_item!r}; choose from {', '.join(PAPER_TASKS)}" ) if value_item not in expanded: expanded.append(value_item) return expanded def _installed_evaluator() -> Path | None: try: package_distribution = distribution("orarl") except PackageNotFoundError: return None suffix = ("share", "orarl", "eval", "task", "eval.sh") for entry in package_distribution.files or (): parts = PurePosixPath(str(entry)).parts if len(parts) < len(suffix) or tuple(parts[-len(suffix) :]) != suffix: continue candidate = Path(package_distribution.locate_file(entry)).resolve() if candidate.is_file(): return candidate return None def _discover_evaluator(explicit: str | None) -> Path: candidates: list[Path] = [] if explicit: candidate = Path(explicit).expanduser() if not candidate.is_file(): raise CliError(f"evaluator does not exist or is not a file: {candidate}") return candidate.resolve() configured = os.environ.get("ORARL_EVALUATOR") if configured: candidate = Path(configured).expanduser() if not candidate.is_file(): raise CliError(f"ORARL_EVALUATOR does not name an evaluator file: {candidate}") return candidate.resolve() runtime_root = os.environ.get("ORARL_RUNTIME_ROOT") if runtime_root: candidates.append(Path(runtime_root).expanduser() / "eval" / "task" / "eval.sh") try: runtime_spec = find_spec("verl") except (ImportError, ModuleNotFoundError, ValueError): runtime_spec = None if runtime_spec is not None: if runtime_spec.origin: candidates.append( Path(runtime_spec.origin).resolve().parent.parent / "eval" / "task" / "eval.sh" ) for location in runtime_spec.submodule_search_locations or (): candidates.append(Path(location).resolve().parent / "eval" / "task" / "eval.sh") candidates.extend( ( Path.cwd() / "eval" / "task" / "eval.sh", Path(__file__).resolve().parents[2] / "eval" / "task" / "eval.sh", Path(__file__).resolve().parents[3] / "eval" / "task" / "eval.sh", ) ) installed_evaluator = _installed_evaluator() if installed_evaluator is not None: candidates.append(installed_evaluator) for candidate in candidates: if candidate.is_file(): return candidate.resolve() raise CliError( "eval/task/eval.sh was not found; pass --evaluator, set " "ORARL_EVALUATOR, or set ORARL_RUNTIME_ROOT to a checkout that ships it" ) def _root_environment(root: Path, tasks: Sequence[str]) -> dict[str, str]: environment: dict[str, str] = {} for task in tasks: task_root = root / task if task == "videomme": environment.update( { "VIDEOMME_VIDEO_BASE": str(task_root), "VIDEOMME_VIDEO_DIR": str(task_root / "videos"), "VIDEOMME_DATA_FILE": str(task_root / "annotations.jsonl"), "VIDEOMME_PREPROCESSED_VIDEO_DIR": str(task_root / "preprocessed_videos"), } ) elif task == "videommev2": environment.update( { "VIDEOMMEV2_VIDEO_BASE": str(task_root), "VIDEOMMEV2_VIDEO_DIR": str(task_root / "videos"), "VIDEOMMEV2_DATA_FILE": str(task_root / "annotations.jsonl"), "VIDEOMMEV2_PREPROCESSED_VIDEO_DIR": str(task_root / "preprocessed_videos"), } ) elif task in { "mvbench", "mmvu", "videoholmes", "longvideobench", "mlvu", }: prefix = _TASK_PREFIX[task] environment[f"{prefix}_DATA_FILE"] = str(task_root / "annotations.jsonl") environment[f"{prefix}_VIDEO_ROOT"] = str(task_root) elif task == "vsi": environment.update( { "VSI_DATA_FILE": str(task_root / "annotations.jsonl"), "VSI_PREPROCESSED_VIDEO_DIR": str(task_root / "preprocessed_videos"), } ) elif task == "mmsi": environment["MMSI_DATA_FILE"] = str(task_root / "MMSI_bench.tsv") elif task == "mindcube": environment.update( { "MINDCUBE_DATA_FILE": str(task_root / "combined-00000-of-00001.parquet"), "MINDCUBE_EXPECTED_SAMPLES": "1050", } ) elif task == "revsi": environment.update( { "REVSI_DATA_FILE": str( task_root / "all_frame" / "test-00000-of-00001.parquet" ), "REVSI_VIDEO_ROOT": str(task_root / "all_frame"), } ) elif task == "spatial_grounding": environment.update( { "SPATIAL_GROUNDING_BENCH_DIR": str(task_root), "SPATIAL_GROUNDING_IMAGE_ROOT": str(task_root / "images"), } ) elif task == "tracking": environment.update( { "TRACKING_BENCH_DIR": str(task_root), "TRACKING_BASE_PREFIX": str(task_root), } ) elif task == "stvg": environment.update( { "STVG_BENCH_DIR": str(task_root), "STVG_BASE_PREFIX": str(task_root), } ) elif task == "temporal_grounding": environment["TIMELENS_BENCH_DIR"] = str(task_root) elif task == "segmentation": environment.update( { "SEGMENTATION_BENCH_DIR": str(task_root), "SEGMENTATION_DATA_ROOT": str(task_root), } ) return environment def _stringify_env(value: Any) -> str: if isinstance(value, bool): return "true" if value else "false" if isinstance(value, list): return ",".join(str(item) for item in value) if value is None: return "" return str(value) def _download_dataset_snapshot( repo_id: str, namespace: argparse.Namespace, allow_patterns: Sequence[str], ) -> Path: try: from huggingface_hub import snapshot_download except ImportError as error: raise CliError( "resolving a Hugging Face --dataset requires huggingface_hub" ) from error cache_dir = ( str(Path(namespace.cache_dir).expanduser()) if namespace.cache_dir else None ) try: snapshot = snapshot_download( repo_id=repo_id, repo_type="dataset", revision=namespace.revision, cache_dir=cache_dir, local_files_only=namespace.local_files_only, allow_patterns=list(allow_patterns), ) except Exception as error: raise CliError(f"cannot resolve Hugging Face dataset {repo_id!r}: {error}") from error return _existing(snapshot, "downloaded dataset root", directory=True) def _load_canonical_manifests( root: Path, asset_root: Path | None = None, ) -> tuple[list[Mapping[str, Any]], list[Mapping[str, Any]]]: assets_root = asset_root or root required = { "datasets.jsonl": root / "datasets.jsonl", "assets.jsonl": assets_root / "assets.jsonl", } missing = [name for name, path in required.items() if not path.is_file()] if missing: raise CliError( f"canonical dataset root {root} is missing: {', '.join(missing)}" ) try: datasets = load_dataset_manifest(root) assets = load_asset_manifest(assets_root) except (ManifestError, StagingError, OSError) as error: raise CliError(f"invalid canonical dataset root {root}: {error}") from error if not datasets: raise CliError(f"canonical dataset manifest is empty: {root / 'datasets.jsonl'}") return list(datasets), list(assets) def _select_dataset_records( records: Sequence[Mapping[str, Any]], tasks: Sequence[str], splits: set[str] | None = None, ) -> list[Mapping[str, Any]]: if splits is not None: available_splits = { str(record["split"]) for record in records if str(record["task"]) in tasks } missing_splits = sorted(splits - available_splits) if missing_splits: raise CliError( "canonical dataset has no requested split(s): " + ", ".join(missing_splits) ) selected = [ record for record in records if str(record["task"]) in tasks and (splits is None or str(record["split"]) in splits) ] covered = {str(record["task"]) for record in selected} missing = sorted(set(tasks) - covered) if missing: raise CliError( "canonical dataset has no configuration record for: " + ", ".join(missing) ) return selected def _selected_asset_records( assets: Sequence[Mapping[str, Any]], records: Sequence[Mapping[str, Any]], ) -> list[Mapping[str, Any]]: annotation_paths = {str(record["annotation_path"]) for record in records} declared_roots = { str(path).rstrip("/") for record in records for field in ("media_paths", "artifact_paths") for path in record.get(field, ()) } def selected(path: str) -> bool: return path in annotation_paths or any( path == root or path.startswith(f"{root}/") for root in declared_roots ) return [ asset for asset in assets if selected(str(asset["path"])) ] def _verify_runtime_assets( root: Path, records: Sequence[Mapping[str, Any]], assets: Sequence[Mapping[str, Any]], asset_root: Path | None = None, ) -> None: selected_assets = _selected_asset_records(assets, records) manifested = {str(asset["path"]) for asset in selected_assets} annotations = {str(record["annotation_path"]) for record in records} missing_annotations = sorted(annotations - manifested) if missing_annotations: raise CliError( "canonical annotations are missing from assets.jsonl: " + ", ".join(missing_annotations) ) try: if asset_root is None: validate_asset_manifest( selected_assets, repository_root=root, context=str(root / "assets.jsonl"), ) validate_dataset_manifest( records, repository_root=root, context=str(root / "datasets.jsonl"), ) else: annotation_assets = [ asset for asset in selected_assets if str(asset["path"]).startswith("annotations/") ] media_assets = [ asset for asset in selected_assets if not str(asset["path"]).startswith("annotations/") ] validate_asset_manifest( annotation_assets, repository_root=root, context=str(root / "assets.jsonl"), ) validate_asset_manifest( media_assets, repository_root=asset_root, context=str(root / "assets.jsonl"), ) validate_dataset_manifest( records, context=str(root / "datasets.jsonl"), ) for record in records: annotation = root / str(record["annotation_path"]) rows = load_evaluation_jsonl( annotation, benchmark=str(record["benchmark"]), split=str(record["split"]), eval_task=str(record["task"]), repository_root=root if asset_root is None else None, ) expected = int(record["expected_count"]) if len(rows) != expected: raise CliError( f"{record['benchmark']}/{record['split']}: expected " f"{expected} rows, found {len(rows)}" ) unmanifested = sorted( { path for row in rows for path in evaluation_asset_paths(row) if path not in manifested } ) if unmanifested: raise CliError( f"{record['benchmark']}/{record['split']} references " "unmanifested assets: " + ", ".join(unmanifested[:5]) ) except (EvaluationSchemaError, ManifestError, StagingError, OSError) as error: raise CliError(f"canonical dataset validation failed: {error}") from error def _resolve_canonical_dataset( namespace: argparse.Namespace, tasks: Sequence[str], asset_root: Path | None = None, ) -> tuple[Path, list[Mapping[str, Any]]]: raw_dataset = str(namespace.dataset).strip() if not raw_dataset: raise CliError("--dataset cannot be empty") local = Path(raw_dataset).expanduser() remote = not local.exists() if remote and ( local.is_absolute() or raw_dataset.startswith((".", "~")) or raw_dataset.endswith(("/", os.sep)) ): raise CliError(f"local canonical dataset root does not exist: {local}") root = ( _download_dataset_snapshot( raw_dataset, namespace, _INDEX_MANIFEST_FILES if asset_root is not None else _MANIFEST_FILES, ) if remote else _existing(raw_dataset, "canonical dataset root", directory=True) ) requested_splits = ( { value.strip().casefold().replace("-", "_") for value in str(namespace.splits).split(",") if value.strip() } if namespace.splits else None ) if requested_splits == set(): raise CliError("--splits cannot be empty") all_records, all_assets = _load_canonical_manifests(root, asset_root) selected = _select_dataset_records(all_records, tasks, requested_splits) if remote and not namespace.dry_run: selected_assets = _selected_asset_records(all_assets, selected) if asset_root is not None: selected_assets = [ asset for asset in selected_assets if str(asset["path"]).startswith("annotations/") ] allow_patterns = [ *(_INDEX_MANIFEST_FILES if asset_root is not None else _MANIFEST_FILES), *(str(asset["path"]) for asset in selected_assets), ] root = _download_dataset_snapshot( raw_dataset, namespace, tuple(dict.fromkeys(allow_patterns)), ) all_records, all_assets = _load_canonical_manifests(root, asset_root) selected = _select_dataset_records(all_records, tasks, requested_splits) if not namespace.dry_run: _verify_runtime_assets(root, selected, all_assets, asset_root) return root, selected def _resolve_profile_path( root: Path, value: str, name: str, asset_root: Path | None = None, ) -> str: relative = Path(value) selected_root = ( asset_root if asset_root is not None and relative.parts and relative.parts[0].casefold() in {"artifacts", "media"} else root ) path = (selected_root / relative).resolve() try: path.relative_to(selected_root) except ValueError as error: raise CliError( f"canonical profile path escapes dataset root: {name}={value}" ) from error return str(path) def _merge_profile_value( environment: dict[str, str], *, task: str, name: str, value: str, ) -> None: previous = environment.get(name) if previous is not None and previous != value: raise CliError( f"conflicting canonical profiles for {task}: " f"{name} is both {previous!r} and {value!r}" ) environment[name] = value def _validate_task_profiles( task: str, records: Sequence[Mapping[str, Any]], ) -> None: splits: set[str] = set() baseline: dict[str, Any] | None = None for record in records: split = str(record["split"]) if split in splits: raise CliError( f"duplicate canonical profile split for {task}: {split}" ) splits.add(split) profile = { "family": record.get("family"), "evaluation": record.get("evaluation"), } if task != "temporal_grounding": profile["preprocessing"] = record.get("preprocessing") if baseline is None: baseline = profile elif profile != baseline: raise CliError( f"conflicting canonical evaluation profiles for task {task}" ) if task not in _STRUCTURED_TASKS and len(records) > 1: raise CliError( f"duplicate canonical profiles for non-structured task {task}" ) def _canonical_task_defaults( root: Path, task: str, records: Sequence[Mapping[str, Any]], asset_root: Path | None = None, ) -> dict[str, str]: annotations = [ (root / str(record["annotation_path"])).resolve() for record in records ] prefix = _TASK_PREFIX[task] defaults: dict[str, str] = {} runtime_asset_root = asset_root or root if task in _STRUCTURED_TASKS: parents = {path.parent for path in annotations} if len(parents) != 1: raise CliError( f"{task} canonical annotations must share one benchmark directory; " "use benchmark equal to eval_task and splits for sub-benchmarks" ) bench_dir = str(next(iter(parents))) defaults[f"{prefix}_BENCH_DIR"] = bench_dir defaults[f"{prefix}_DATASETS"] = ",".join( _evaluator_split_name(task, record["split"]) for record in records ) if task == "spatial_grounding": defaults["SPATIAL_GROUNDING_IMAGE_ROOT"] = str(runtime_asset_root) elif task == "tracking": defaults["TRACKING_BASE_PREFIX"] = str(runtime_asset_root) elif task == "stvg": defaults["STVG_BASE_PREFIX"] = str(runtime_asset_root) elif task == "segmentation": defaults["SEGMENTATION_DATA_ROOT"] = str(runtime_asset_root) return defaults annotation = str(annotations[0]) defaults[f"{prefix}_DATA_FILE"] = annotation if task in {"videomme", "videommev2"}: defaults[f"{prefix}_VIDEO_BASE"] = str(runtime_asset_root) defaults[f"{prefix}_VIDEO_DIR"] = str(runtime_asset_root) defaults[f"{prefix}_PREPROCESSED_VIDEO_DIR"] = str(runtime_asset_root) elif task in { "mvbench", "mmvu", "videoholmes", "longvideobench", "mlvu", }: defaults[f"{prefix}_VIDEO_ROOT"] = str(runtime_asset_root) if task == "longvideobench": defaults["LONGVIDEOBENCH_SUBTITLE_ROOT"] = str(runtime_asset_root) elif task == "vsi": defaults["VSI_PREPROCESSED_VIDEO_DIR"] = str(runtime_asset_root) defaults["VSI_EXPECTED_SAMPLES"] = str(records[0]["expected_count"]) elif task == "mmsi": defaults["MMSI_EXPECTED_SAMPLES"] = str(records[0]["expected_count"]) elif task == "mindcube": defaults["MINDCUBE_EXPECTED_SAMPLES"] = str(records[0]["expected_count"]) elif task == "revsi": defaults["REVSI_VIDEO_ROOT"] = str(runtime_asset_root) defaults["REVSI_EXPECTED_SAMPLES"] = str(records[0]["expected_count"]) return defaults def _canonical_environment( root: Path, records: Sequence[Mapping[str, Any]], tasks: Sequence[str], asset_root: Path | None = None, ) -> dict[str, str]: environment = { "ORARL_EVAL_DATA_ROOT": str(root), "ORARL_EVAL_DATASETS_JSONL": str(root / "datasets.jsonl"), } if asset_root is not None: environment["ORARL_EVAL_ASSET_ROOT"] = str(asset_root) for task in tasks: task_records = [record for record in records if record["task"] == task] _validate_task_profiles(task, task_records) prefix = _TASK_PREFIX[task] + "_" for record in task_records: values = record.get("legacy_environment", {}) if not isinstance(values, Mapping): continue for raw_name, raw_value in values.items(): name = str(raw_name).strip().upper() if not name.startswith(prefix) and name != "EVAL_CONDA_ENV": raise CliError( f"unrelated canonical profile setting for {task}: {name}" ) if _SENSITIVE_NAME.search(name): raise CliError("canonical profiles cannot contain sensitive settings") value = _stringify_env(raw_value) if _URI_USERINFO.search(value): raise CliError("canonical profiles cannot contain credential-bearing URLs") if name.endswith(_PATH_ENV_SUFFIXES) and value: value = _resolve_profile_path(root, value, name, asset_root) _merge_profile_value( environment, task=task, name=name, value=value, ) for name, value in _canonical_task_defaults( root, task, task_records, asset_root, ).items(): if name.endswith("_DATASETS"): selected = value.split(",") previous = environment.get(name, "").split(",") ordered = [ item for prior in previous for item in selected if prior.strip().casefold().replace("-", "_") == item.casefold().replace("-", "_") ] environment[name] = ",".join( dict.fromkeys([*ordered, *selected]) ) else: environment.setdefault(name, value) if task == "temporal_grounding": for field, name in { "fps": "TIMELENS_FPS", "min_tokens": "TIMELENS_MIN_TOKENS", "max_frames": "TIMELENS_MAX_FRAMES", "max_pixels": "TIMELENS_MAX_PIXELS", "total_tokens": "TIMELENS_TOTAL_TOKENS", }.items(): values = { str(preprocessing[field]) for record in task_records if isinstance( preprocessing := record.get("preprocessing"), Mapping, ) and field in preprocessing } if len(values) == 1: environment[name] = values.pop() return environment def _task_configs( specifications: Sequence[str], tasks: Sequence[str], data_root: Path | None, ) -> tuple[dict[str, str], set[str]]: environment: dict[str, str] = {} configured_tasks: set[str] = set() for specification in specifications: task, separator, raw_path = specification.partition("=") task = task.strip().casefold() if not separator or task not in PAPER_TASKS: raise CliError(f"--task-config expects TASK=FILE for a paper task: {specification!r}") if task not in tasks: raise CliError(f"task config supplied for unrequested task: {task}") config_path = _existing(raw_path, f"{task} config", directory=False) try: payload = yaml.safe_load(config_path.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError) as error: raise CliError(f"cannot load {config_path}: {error}") from error if not isinstance(payload, Mapping): raise CliError(f"task config must be a mapping: {config_path}") declared_task = payload.get("task") if declared_task is not None and str(declared_task).casefold() != task: raise CliError(f"{config_path} declares task {declared_task!r}, expected {task!r}") values = payload.get("environment", payload.get("env", payload)) if not isinstance(values, Mapping): raise CliError(f"{config_path}: environment must be a mapping") prefix = _TASK_PREFIX[task] + "_" for raw_name, raw_value in values.items(): name = str(raw_name).strip().upper() if name == "TASK": continue if not name.startswith(prefix) and name != "EVAL_CONDA_ENV": raise CliError(f"{config_path}: {name} is not allowlisted for {task}") if _SENSITIVE_NAME.search(name): raise CliError(f"{config_path}: sensitive environment names are not accepted") value = _stringify_env(raw_value) if _URI_USERINFO.search(value): raise CliError(f"{config_path}: credential-bearing URLs are not accepted") if name.endswith(_PATH_ENV_SUFFIXES) and value and not Path(value).is_absolute(): base = data_root if data_root is not None else config_path.parent value = str((base / value).resolve()) environment[name] = value configured_tasks.add(task) return environment, configured_tasks def build_launch( namespace: argparse.Namespace, ) -> tuple[list[str], dict[str, str], list[str], Path, Path]: model = _existing(namespace.model, "model", directory=True) tasks = _parse_tasks(namespace.tasks) evaluator = _discover_evaluator(namespace.evaluator) if not re.fullmatch(r"\d+(?:,\d+)*", namespace.gpus): raise CliError("--gpus must be a comma-separated list of non-negative ids") if namespace.tp_size <= 0: raise CliError("--tp-size must be positive") if namespace.batch_size is not None and namespace.batch_size <= 0: raise CliError("--batch-size must be positive") if namespace.max_samples is not None and namespace.max_samples <= 0: raise CliError("--max-samples must be positive") if namespace.force_merge and namespace.skip_merge: raise CliError("--force-merge and --skip-merge are mutually exclusive") if namespace.dataset and namespace.data_root: raise CliError("--dataset and legacy --data-root are mutually exclusive") if namespace.asset_root and not namespace.dataset: raise CliError("--asset-root requires --dataset") if namespace.splits and not namespace.dataset: raise CliError("--splits requires --dataset") canonical_root: Path | None = None canonical_records: list[Mapping[str, Any]] = [] asset_root = ( _existing(namespace.asset_root, "asset root", directory=True) if namespace.asset_root else None ) if namespace.dataset: canonical_root, canonical_records = _resolve_canonical_dataset( namespace, tasks, asset_root, ) environment = _canonical_environment( canonical_root, canonical_records, tasks, asset_root, ) else: environment = {} data_root = None if namespace.data_root: data_root = _existing(namespace.data_root, "data root", directory=True) environment.update(_root_environment(data_root, tasks)) config_root = canonical_root if canonical_root is not None else data_root configured_environment, configured_tasks = _task_configs( namespace.task_config, tasks, config_root ) environment.update(configured_environment) if canonical_root is not None: configured_tasks.update(tasks) if namespace.batch_size is not None: for task in tasks: environment[f"{_TASK_PREFIX[task]}_BATCH_SIZE"] = str(namespace.batch_size) if namespace.max_samples is not None: unsupported = set(tasks) - { *VIDEO_QA_TASKS, "temporal_grounding", "spatial_grounding", "tracking", "stvg", "revsi", "segmentation", } if unsupported: raise CliError( "--max-samples only supports video QA, temporal_grounding, " "spatial_grounding, tracking, stvg, revsi, or segmentation" ) if "temporal_grounding" in tasks: environment["TIMELENS_MAX_SAMPLES"] = str(namespace.max_samples) if "spatial_grounding" in tasks: environment["SPATIAL_GROUNDING_MAX_SAMPLES"] = str( namespace.max_samples ) if "tracking" in tasks: environment["TRACKING_MAX_SAMPLES"] = str(namespace.max_samples) if "stvg" in tasks: environment["STVG_MAX_SAMPLES"] = str(namespace.max_samples) if "revsi" in tasks: environment["REVSI_MAX_SAMPLES"] = str(namespace.max_samples) if "segmentation" in tasks: environment["SEGMENTATION_MAX_SAMPLES"] = str(namespace.max_samples) for task in tasks: if task in VIDEO_QA_TASKS: environment[f"{_TASK_PREFIX[task]}_MAX_SAMPLES"] = str( namespace.max_samples ) eval_conda_env = environment.pop("EVAL_CONDA_ENV", "").strip() if not namespace.dry_run: missing_configs = ( sorted(set(tasks) - configured_tasks) if data_root is None and canonical_root is None else [] ) if missing_configs: raise CliError( "--run requires --dataset, --data-root, or one --task-config per task; " f"missing: {', '.join(missing_configs)}" ) for task in tasks: missing_inputs = sorted(_REQUIRED_INPUTS[task] - environment.keys()) if missing_inputs: raise CliError( f"{task} is missing explicit input settings: " + ", ".join(missing_inputs) ) for name in sorted(_REQUIRED_INPUTS[task]): path = Path(environment[name]) expected_file = name.endswith("_FILE") if not path.exists(): raise CliError(f"{task} input does not exist: {name}={path}") if expected_file and not path.is_file(): raise CliError(f"{task} input must be a file: {name}={path}") if not expected_file and not path.is_dir(): raise CliError(f"{task} input must be a directory: {name}={path}") if namespace.segmentation_run_sam2: postprocessor_inputs = { "SEGMENTATION_POSTPROCESSOR_PATH", "SEGMENTATION_SAM2_CFG", "SEGMENTATION_SAM2_CKPT", } missing_postprocessor = sorted(postprocessor_inputs - environment.keys()) if missing_postprocessor: raise CliError( "segmentation post-processing requires explicit settings: " + ", ".join(missing_postprocessor) ) for name in sorted(postprocessor_inputs): path = Path(environment[name]) if not path.is_file(): raise CliError( f"segmentation post-processing input must be a file: {name}={path}" ) command = [ "bash", str(evaluator), "--model", str(model), "--tasks", ",".join(tasks), "--gpus", namespace.gpus, "--tp-size", str(namespace.tp_size), ] if namespace.force_merge: command.append("--force-merge") if namespace.skip_merge: command.append("--skip-merge") if namespace.merged_model: command.extend(("--merged-model", str(Path(namespace.merged_model).expanduser().resolve()))) if namespace.base_model: command.extend(("--base-model", str(_existing(namespace.base_model, "base model")))) if namespace.segmentation_run_sam2: command.append("--segmentation-run-sam2") if eval_conda_env: command.extend(("--env", eval_conda_env)) project_root = evaluator.parents[2] results_root = ( Path(namespace.results_root).expanduser().resolve() if namespace.results_root else project_root / "outputs" ) summary_path = Path(namespace.summary).expanduser().resolve() return command, environment, tasks, results_root, summary_path def _execution_environment(configured: Mapping[str, str]) -> dict[str, str]: """Build an evaluator environment without inherited task-path state.""" environment = os.environ.copy() task_prefixes = tuple(prefix + "_" for prefix in set(_TASK_PREFIX.values())) for name in tuple(environment): if ( name.startswith(task_prefixes) or name == "EVAL_CONDA_ENV" or name.startswith("ORARL_EVAL_") ): environment.pop(name, None) environment.update(configured) return environment def _summary_state(root: Path) -> dict[Path, int]: if not root.is_dir(): return {} return {path: path.stat().st_mtime_ns for path in root.rglob("summary.json") if path.is_file()} def _task_for_summary(path: Path, tasks: Sequence[str]) -> str | None: lowered_parts = {part.casefold() for part in path.parts} for task in tasks: if task in lowered_parts: return task return None def _write_aggregate( destination: Path, *, model: str, tasks: Sequence[str], results_root: Path, before: Mapping[Path, int], returncode: int, ) -> None: after = _summary_state(results_root) changed = [ path for path, modified in after.items() if path not in before or before[path] != modified ] records: list[dict[str, Any]] = [] observed: set[str] = set() for path in sorted(changed): task = _task_for_summary(path, tasks) if task is not None: observed.add(task) try: metrics: Any = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: metrics = {"error": str(error)} try: display_path = str(path.relative_to(results_root)) except ValueError: display_path = str(path) records.append({"task": task, "summary": display_path, "metrics": metrics}) payload = { "schema_version": 1, "created_at": datetime.now(timezone.utc).isoformat(), "model": model, "requested_tasks": list(tasks), "completed_tasks": sorted(observed), "missing_tasks": sorted(set(tasks) - observed), "evaluator_returncode": returncode, "results": records, } destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def main(argv: Sequence[str] | None = None) -> int: parser = create_parser() namespace = parser.parse_args(argv) try: command, configured_env, tasks, results_root, summary_path = build_launch(namespace) except CliError as error: parser.error(str(error)) print(f"mode={'dry-run' if namespace.dry_run else 'run'} tasks={','.join(tasks)}") if configured_env: print("configured environment: " + ", ".join(sorted(configured_env))) print(shlex.join(command)) if namespace.dry_run: print(f"aggregate summary would be written to {summary_path}") return 0 before = _summary_state(results_root) environment = _execution_environment(configured_env) completed = subprocess.run(command, env=environment, check=False) _write_aggregate( summary_path, model=str(Path(namespace.model).expanduser().resolve()), tasks=tasks, results_root=results_root, before=before, returncode=completed.returncode, ) print(f"aggregate summary: {summary_path}") return completed.returncode if __name__ == "__main__": raise SystemExit(main())