| """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) |
|
|
