| """Configuration loading with paths resolved relative to the config file.""" |
|
|
| from __future__ import annotations |
|
|
| from pathlib import Path |
| from typing import Any |
|
|
| import yaml |
|
|
|
|
| PATH_KEYS = { |
| "Calibration_table_path", |
| "save_dir", |
| "radar_path_cappi", |
| "radar_path_hsr", |
| "radar_path_hsp", |
| "radar_latlon_file", |
| "gk2a_ea020_latlon_file", |
| "gk2a_base_dir", |
| "gk2a_l2_base_dir", |
| "input_root", |
| "region_root", |
| "bt_root", |
| "hsr_root", |
| "label_root", |
| "links_root", |
| "log_file", |
| "log_path", |
| "root", |
| "output_root", |
| "output_dir", |
| "catalog_path", |
| "checkpoint_path", |
| "config", |
| "dataset_config", |
| "stats_path", |
| "physics_formula_path", |
| "validation_json_template", |
| "temporal_overlapping_dir", |
| "mask_root", |
| } |
|
|
|
|
| def _resolve(value: Any, base: Path) -> Any: |
| if isinstance(value, str): |
| path = Path(value).expanduser() |
| return str(path if path.is_absolute() else (base / path).resolve()) |
| if isinstance(value, list): |
| return [_resolve(item, base) for item in value] |
| return value |
|
|
|
|
| def resolve_paths(config: dict[str, Any], config_path: str | Path) -> dict[str, Any]: |
| """Return a copy with known filesystem fields made absolute.""" |
| base = Path(config_path).resolve().parent |
|
|
| def walk(value: Any) -> Any: |
| if isinstance(value, dict): |
| out: dict[str, Any] = {} |
| for key, item in value.items(): |
| if key in PATH_KEYS and item not in (None, ""): |
| out[key] = _resolve(item, base) |
| elif key == "model_output_dirs" and isinstance(item, dict): |
| out[key] = { |
| provider: _resolve(provider_path, base) |
| for provider, provider_path in item.items() |
| } |
| else: |
| out[key] = walk(item) |
| return out |
| if isinstance(value, list): |
| return [walk(item) for item in value] |
| return value |
|
|
| return walk(config) |
|
|
|
|
| def load_config(path: str | Path) -> dict[str, Any]: |
| path = Path(path).resolve() |
| with path.open("r", encoding="utf-8") as stream: |
| config = yaml.safe_load(stream) |
| if not isinstance(config, dict): |
| raise ValueError(f"config must be a mapping: {path}") |
| config = resolve_paths(config, path) |
| config["_config_path"] = str(path) |
| return config |
|
|
