File indexing completed on 2026-08-12 08:24:56
0001 """Configuration loader wrappers for AID2E.
0002
0003 These helpers provide section-level loading APIs on top of the existing
0004 ``load_config`` flow so callers can load canonical config sections directly.
0005 """
0006
0007 from __future__ import annotations
0008
0009 import json
0010 from pathlib import Path
0011 from typing import Any, Dict, Optional
0012
0013 import yaml
0014
0015 from .full_config import _normalize_full_config_data
0016 from .optimizer_config import OptimizerConfiguration
0017 from .problem_config import ProblemConfiguration
0018 from .scheduler_config import SchedulerConfigLoader, SchedulerConfiguration
0019 from .workflow_config import WorkflowDefinition, WorkflowsConfiguration
0020
0021
0022 def load_raw_config(config_file: str) -> Dict[str, Any]:
0023 """Load raw YAML/JSON configuration content from disk.
0024
0025 Args:
0026 config_file: Path to YAML or JSON config.
0027
0028 Returns:
0029 Raw top-level configuration dictionary.
0030
0031 Raises:
0032 FileNotFoundError: If the config file does not exist.
0033 ValueError: If the extension is unsupported.
0034 """
0035 path = Path(config_file)
0036 if not path.exists():
0037 raise FileNotFoundError(f"Config file not found: {config_file}")
0038
0039 text = path.read_text(encoding="utf-8")
0040 suffix = path.suffix.lower()
0041 if suffix in {".yaml", ".yml"}:
0042 return yaml.safe_load(text) or {}
0043 if suffix == ".json":
0044 return json.loads(text)
0045
0046 raise ValueError(
0047 f"Unsupported config extension '{suffix}'. "
0048 "Use .yaml, .yml, or .json."
0049 )
0050
0051
0052 def _normalize_sections(config_file: str) -> Dict[str, Any]:
0053 """Normalize full config content into canonical sections."""
0054 path = Path(config_file)
0055 raw = load_raw_config(config_file)
0056 return _normalize_full_config_data(raw, path)
0057
0058
0059 def load_problem_config(config_file: str) -> ProblemConfiguration:
0060 """Load only the problem section from a full config file."""
0061 normalized = _normalize_sections(config_file)
0062 return normalized["problem"]
0063
0064
0065 def load_optimizer_config(config_file: str) -> OptimizerConfiguration:
0066 """Load only the optimizer section from a full config file."""
0067 normalized = _normalize_sections(config_file)
0068 return normalized["optimizer"]
0069
0070
0071 def load_scheduler_config(config_file: str) -> Optional[SchedulerConfiguration]:
0072 """Load only the scheduler section from a full config file.
0073
0074 Accepts only canonical scheduler payloads with ``parameters``.
0075 """
0076 path = Path(config_file)
0077 raw = load_raw_config(config_file)
0078 scheduler_raw = raw.get("scheduler")
0079 if not scheduler_raw:
0080 normalized = _normalize_full_config_data(raw, path)
0081 return normalized.get("scheduler")
0082
0083 if not isinstance(scheduler_raw, dict):
0084 raise ValueError("'scheduler' section must be a mapping")
0085
0086 return SchedulerConfigLoader.from_dict(scheduler_raw, base_dir=str(path.parent))
0087
0088
0089 def load_workflow_config(config_file: str) -> Optional[WorkflowsConfiguration]:
0090 """Load workflow configuration from full config if present.
0091
0092 Accepted layout:
0093 - ``workflows: {workflows: [...]}``
0094 """
0095 normalized = _normalize_sections(config_file)
0096 return normalized.get("workflows")
