"""Resolve explicit capabilities without benchmark-specific fallbacks.""" from typing import Any from hydra.utils import instantiate from omegaconf import DictConfig, OmegaConf def identity(cfg: DictConfig) -> dict[str, Any]: spec = cfg.get("instantiation") if spec is None or not spec.get("name") or spec.get("version") != 1: raise ValueError("An explicit instantiation name and supported version (1) are required") return {"name": str(spec.name), "version": int(spec.version)} def provider(cfg: DictConfig, capability: str) -> Any: identity(cfg) spec = OmegaConf.select(cfg, f"instantiation.providers.{capability}") if not isinstance(spec, DictConfig) or not spec.get("_target_"): raise ValueError(f"Instantiation {cfg.instantiation.name!r} lacks {capability!r} provider") instance = instantiate(spec) required = ("prepare", "build") if capability == "model" else ( ("build",) if capability in ("dataset", "metrics") else ("run",) ) for method in required: if not callable(getattr(instance, method, None)): raise ValueError(f"Provider {capability!r} must implement {method}()") return instance def evaluate(cfg: DictConfig) -> dict: OmegaConf.resolve(cfg) return provider(cfg, "evaluation").run(cfg) def predict(cfg: DictConfig) -> dict: OmegaConf.resolve(cfg) return provider(cfg, "prediction").run(cfg) def train_from_scratch(cfg: DictConfig): OmegaConf.resolve(cfg) return provider(cfg, "workflow").run(cfg)