File size: 2,523 Bytes
7da2ecb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """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",
"mature_cloud_dir",
"prediction_dir",
"all_targets_json",
"model_available_json",
"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
|