| """Typed local source manifests for deterministic evaluation-data builds.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import re |
| from collections.abc import Mapping |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Any, TypedDict, Union |
| from urllib.parse import urlsplit |
|
|
| from .layout import LayoutError, validate_dataset_id |
|
|
| PathLike = Union[str, os.PathLike[str]] |
|
|
| GENERIC_ADAPTERS = frozenset( |
| { |
| "generic", |
| "json", |
| "video_qa", |
| "one_thinker", |
| "onethinker", |
| "spatial_grounding", |
| } |
| ) |
| SOURCE_ADAPTERS = frozenset( |
| {*GENERIC_ADAPTERS, "mmsi", "mindcube", "revsi", "timelens"} |
| ) |
| _MEDIA_ROOT_KEYS = frozenset({"images", "videos", "subtitles", "default"}) |
| _URI_SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://") |
| _ENVIRONMENT_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]*$") |
|
|
|
|
| class _RequiredEvaluationSourceRecord(TypedDict): |
| benchmark: str |
| eval_task: str |
| split: str |
| family: str |
| adapter: str |
| annotation_input: str |
| expected_count: int | None |
| license: str |
| source_url: str |
| redistribution_authorized: bool |
| evaluation: dict[str, Any] |
|
|
|
|
| class EvaluationSourceRecord(_RequiredEvaluationSourceRecord, total=False): |
| """One JSONL source record for a benchmark/split conversion.""" |
|
|
| annotation: str |
| annotation_path: str |
| input: str |
| media_root: str |
| media_roots: dict[str, str] |
| image_root: str |
| video_root: str |
| subtitle_root: str |
| artifact_root: str |
| preprocessed_root: str |
| preprocessed_roots: dict[str, str] |
| preprocessing: dict[str, Any] |
| legacy_environment: dict[str, Any] |
| metadata: dict[str, Any] |
|
|
|
|
| class SourceManifestError(ValueError): |
| """Raised when a local evaluation source manifest is invalid.""" |
|
|
|
|
| @dataclass(frozen=True) |
| class EvaluationSource: |
| """Resolved local source information that is never serialized to outputs.""" |
|
|
| benchmark: str |
| eval_task: str |
| split: str |
| family: str |
| adapter: str |
| annotation_input: Path |
| media_roots: Mapping[str, Path] |
| preprocessed_roots: Mapping[str, Path] |
| expected_count: int | None |
| license: str |
| source_url: str |
| redistribution_authorized: bool |
| evaluation: Mapping[str, Any] |
| preprocessing: Mapping[str, Any] |
| legacy_environment: Mapping[str, Any] |
| metadata: Mapping[str, Any] |
| manifest_index: int |
|
|
| @property |
| def annotation(self) -> Path: |
| """Return the resolved annotation input path.""" |
|
|
| return self.annotation_input |
|
|
| def media_root(self, kind: str) -> Path: |
| """Return the most specific source root for a media kind.""" |
|
|
| return self.media_roots.get(kind, self.media_roots.get("default", self.annotation.parent)) |
|
|
| def preprocessed_root(self, key: str | None = None) -> Path: |
| """Return a source artifact root selected by key, then by default.""" |
|
|
| if key: |
| folded = key.casefold() |
| for name, root in self.preprocessed_roots.items(): |
| if name != "default" and name.casefold() in folded: |
| return root |
| if "default" in self.preprocessed_roots: |
| return self.preprocessed_roots["default"] |
| if self.preprocessed_roots: |
| return self.preprocessed_roots[sorted(self.preprocessed_roots)[0]] |
| return self.annotation.parent |
|
|
|
|
| def _local_path(base: Path, value: object, *, context: str) -> Path: |
| if not isinstance(value, str) or not value.strip() or value != value.strip(): |
| raise SourceManifestError(f"{context} must be a nonempty local path string") |
| if _URI_SCHEME_RE.match(value): |
| raise SourceManifestError(f"{context} must be local, not a URL") |
| candidate = Path(os.path.expanduser(value)) |
| if not candidate.is_absolute(): |
| candidate = base / candidate |
| return Path(os.path.abspath(os.path.normpath(str(candidate)))) |
|
|
|
|
| def _required_text(record: Mapping[str, Any], field: str, *, context: str) -> str: |
| value = record.get(field) |
| if not isinstance(value, str) or not value.strip() or value != value.strip(): |
| raise SourceManifestError(f"{context}.{field} must be a nonempty string") |
| return value |
|
|
|
|
| def _mapping(record: Mapping[str, Any], field: str, *, context: str) -> dict[str, Any]: |
| value = record.get(field) |
| if value is None: |
| return {} |
| if not isinstance(value, Mapping): |
| raise SourceManifestError(f"{context}.{field} must be a JSON object") |
| return {str(key): item for key, item in value.items()} |
|
|
|
|
| def _source_url(record: Mapping[str, Any], *, context: str) -> str: |
| value = _required_text(record, "source_url", context=context) |
| parsed = urlsplit(value) |
| if ( |
| parsed.scheme not in {"http", "https"} |
| or not parsed.netloc |
| or parsed.username is not None |
| or parsed.password is not None |
| or any(character.isspace() for character in value) |
| ): |
| raise SourceManifestError( |
| f"{context}.source_url must be an HTTP(S) URL without embedded credentials" |
| ) |
| return value |
|
|
|
|
| def _annotation_value(record: Mapping[str, Any], *, context: str) -> object: |
| supplied = [ |
| (field, record[field]) |
| for field in ("annotation_input", "annotation", "annotation_path", "input") |
| if record.get(field) is not None |
| ] |
| if not supplied: |
| raise SourceManifestError(f"{context}.annotation_input is required") |
| spellings = {str(value) for _field, value in supplied} |
| if len(spellings) != 1: |
| fields = ", ".join(field for field, _value in supplied) |
| raise SourceManifestError(f"{context} has conflicting annotation fields: {fields}") |
| return supplied[0][1] |
|
|
|
|
| def _root_mapping( |
| record: Mapping[str, Any], |
| base: Path, |
| *, |
| context: str, |
| ) -> tuple[dict[str, Path], dict[str, Path]]: |
| media: dict[str, Path] = {} |
| media_root = record.get("media_root") |
| if media_root is not None: |
| media["default"] = _local_path(base, media_root, context=f"{context}.media_root") |
|
|
| raw_media = record.get("media_roots") |
| if raw_media is not None: |
| if not isinstance(raw_media, Mapping): |
| raise SourceManifestError(f"{context}.media_roots must be a JSON object") |
| for raw_kind, raw_path in raw_media.items(): |
| kind = str(raw_kind).casefold().removesuffix("_root") |
| if kind in {"image", "video", "subtitle"}: |
| kind += "s" |
| if kind not in _MEDIA_ROOT_KEYS: |
| choices = ", ".join(sorted(_MEDIA_ROOT_KEYS)) |
| raise SourceManifestError( |
| f"{context}.media_roots keys must be one of {{{choices}}}" |
| ) |
| media[kind] = _local_path( |
| base, |
| raw_path, |
| context=f"{context}.media_roots[{raw_kind!r}]", |
| ) |
|
|
| for field, kind in ( |
| ("image_root", "images"), |
| ("video_root", "videos"), |
| ("subtitle_root", "subtitles"), |
| ): |
| if record.get(field) is not None: |
| media[kind] = _local_path(base, record[field], context=f"{context}.{field}") |
|
|
| preprocessed: dict[str, Path] = {} |
| raw_preprocessed_root = record.get( |
| "preprocessed_root", |
| record.get("artifact_root"), |
| ) |
| if raw_preprocessed_root is not None: |
| preprocessed["default"] = _local_path( |
| base, |
| raw_preprocessed_root, |
| context=f"{context}.preprocessed_root", |
| ) |
| raw_preprocessed = record.get("preprocessed_roots") |
| if raw_preprocessed is not None: |
| if not isinstance(raw_preprocessed, Mapping): |
| raise SourceManifestError(f"{context}.preprocessed_roots must be a JSON object") |
| for raw_name, raw_path in raw_preprocessed.items(): |
| name = str(raw_name).strip().casefold() |
| if not name: |
| raise SourceManifestError(f"{context}.preprocessed_roots keys must be nonempty") |
| preprocessed[name] = _local_path( |
| base, |
| raw_path, |
| context=f"{context}.preprocessed_roots[{raw_name!r}]", |
| ) |
| return media, preprocessed |
|
|
|
|
| def _source_from_record( |
| record: Mapping[str, Any], |
| manifest_path: Path, |
| index: int, |
| ) -> EvaluationSource: |
| context = f"{manifest_path}:{index}" |
| try: |
| benchmark = validate_dataset_id(record.get("benchmark"), context="benchmark") |
| eval_task = validate_dataset_id(record.get("eval_task"), context="eval_task") |
| split = validate_dataset_id(record.get("split"), context="split") |
| family = validate_dataset_id(record.get("family"), context="family") |
| except LayoutError as error: |
| raise SourceManifestError(f"{context}: {error}") from error |
|
|
| adapter = _required_text(record, "adapter", context=context).casefold().replace("-", "_") |
| if adapter not in SOURCE_ADAPTERS: |
| choices = ", ".join(sorted(SOURCE_ADAPTERS)) |
| raise SourceManifestError(f"{context}.adapter must be one of {{{choices}}}") |
|
|
| annotation = _local_path( |
| manifest_path.parent, |
| _annotation_value(record, context=context), |
| context=f"{context}.annotation_input", |
| ) |
| if not annotation.is_file(): |
| raise SourceManifestError(f"{context}.annotation_input does not exist: {annotation}") |
| if annotation.suffix.casefold() in {".yaml", ".yml"}: |
| raise SourceManifestError(f"{context}.annotation_input must not be YAML") |
|
|
| expected_count = record.get("expected_count") |
| if expected_count is not None and ( |
| isinstance(expected_count, bool) |
| or not isinstance(expected_count, int) |
| or expected_count < 0 |
| ): |
| raise SourceManifestError(f"{context}.expected_count must be null or a nonnegative integer") |
|
|
| redistribution = record.get("redistribution_authorized") |
| if not isinstance(redistribution, bool): |
| raise SourceManifestError( |
| f"{context}.redistribution_authorized must be an explicit boolean" |
| ) |
|
|
| evaluation = _mapping(record, "evaluation", context=context) |
| if not evaluation: |
| raise SourceManifestError(f"{context}.evaluation must be a nonempty JSON object") |
| preprocessing = _mapping(record, "preprocessing", context=context) |
| legacy_environment = _mapping(record, "legacy_environment", context=context) |
| for name in legacy_environment: |
| if not _ENVIRONMENT_NAME_RE.fullmatch(name): |
| raise SourceManifestError( |
| f"{context}.legacy_environment names must use uppercase letters, " |
| "digits, and underscores" |
| ) |
| metadata = _mapping(record, "metadata", context=context) |
| media_roots, preprocessed_roots = _root_mapping( |
| record, |
| manifest_path.parent, |
| context=context, |
| ) |
|
|
| try: |
| json.dumps( |
| { |
| "evaluation": evaluation, |
| "preprocessing": preprocessing, |
| "legacy_environment": legacy_environment, |
| "metadata": metadata, |
| }, |
| ensure_ascii=False, |
| allow_nan=False, |
| ) |
| except (TypeError, ValueError) as error: |
| raise SourceManifestError(f"{context}: metadata must be strict JSON: {error}") from error |
|
|
| license_name = _required_text(record, "license", context=context) |
| if license_name.startswith(("/", "~/", "file://")) or re.match( |
| r"^[A-Za-z]:[\\/]", license_name |
| ): |
| raise SourceManifestError(f"{context}.license must not contain a local path") |
|
|
| return EvaluationSource( |
| benchmark=benchmark, |
| eval_task=eval_task, |
| split=split, |
| family=family, |
| adapter=adapter, |
| annotation_input=annotation, |
| media_roots=media_roots, |
| preprocessed_roots=preprocessed_roots, |
| expected_count=expected_count, |
| license=license_name, |
| source_url=_source_url(record, context=context), |
| redistribution_authorized=redistribution, |
| evaluation=evaluation, |
| preprocessing=preprocessing, |
| legacy_environment=legacy_environment, |
| metadata=metadata, |
| manifest_index=index, |
| ) |
|
|
|
|
| def load_source_manifest(path: PathLike) -> list[EvaluationSource]: |
| """Load a local JSONL source manifest, resolving only local input paths.""" |
|
|
| manifest_path = Path(path).expanduser() |
| if manifest_path.suffix.casefold() not in {".jsonl", ".ndjson"}: |
| raise SourceManifestError("evaluation source manifest must be JSONL, not YAML or JSON") |
| manifest_path = manifest_path.resolve() |
| if not manifest_path.is_file(): |
| raise SourceManifestError(f"source manifest does not exist: {manifest_path}") |
|
|
| sources: list[EvaluationSource] = [] |
| datasets: dict[tuple[str, str], int] = {} |
| with manifest_path.open("r", encoding="utf-8") as handle: |
| for line_number, line in enumerate(handle, start=1): |
| if not line.strip(): |
| continue |
| try: |
| record = json.loads(line) |
| except json.JSONDecodeError as error: |
| raise SourceManifestError( |
| f"{manifest_path}:{line_number}: invalid JSON: {error}" |
| ) from error |
| if not isinstance(record, Mapping): |
| raise SourceManifestError( |
| f"{manifest_path}:{line_number}: source record must be a JSON object" |
| ) |
| source = _source_from_record(record, manifest_path, line_number) |
| key = (source.benchmark, source.split) |
| previous = datasets.get(key) |
| if previous is not None: |
| raise SourceManifestError( |
| f"{manifest_path}:{line_number}: duplicate source {key[0]}/{key[1]} " |
| f"(first seen at line {previous})" |
| ) |
| datasets[key] = line_number |
| sources.append(source) |
| if not sources: |
| raise SourceManifestError(f"{manifest_path}: source manifest is empty") |
| return sources |
|
|
|
|
| load_evaluation_source_manifest = load_source_manifest |
| load_evaluation_sources = load_source_manifest |
| EvaluationSourceSpec = EvaluationSource |
| SourceManifestRecord = EvaluationSourceRecord |
|
|
|
|
| __all__ = [ |
| "EvaluationSource", |
| "EvaluationSourceRecord", |
| "EvaluationSourceSpec", |
| "GENERIC_ADAPTERS", |
| "SOURCE_ADAPTERS", |
| "SourceManifestError", |
| "SourceManifestRecord", |
| "load_evaluation_source_manifest", |
| "load_evaluation_sources", |
| "load_source_manifest", |
| ] |
|
|
