from __future__ import annotations import hashlib import json from copy import deepcopy from pathlib import Path from typing import Any import yaml def load_config(path: str | Path) -> dict[str, Any]: with Path(path).open("r", encoding="utf-8") as handle: data = yaml.safe_load(handle) if not isinstance(data, dict): raise ValueError(f"Config at {path} must deserialize into a mapping.") return data def deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: merged = deepcopy(base) for key, value in override.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = deep_merge(merged[key], value) else: merged[key] = value return merged def ensure_dir(path: str | Path) -> Path: resolved = Path(path) resolved.mkdir(parents=True, exist_ok=True) return resolved def save_json(path: str | Path, payload: dict[str, Any]) -> None: target = Path(path) target.parent.mkdir(parents=True, exist_ok=True) with target.open("w", encoding="utf-8") as handle: json.dump(payload, handle, indent=2, ensure_ascii=False) def flatten_for_wandb(payload: dict[str, Any], prefix: str = "") -> dict[str, Any]: flat: dict[str, Any] = {} for key, value in payload.items(): full_key = f"{prefix}.{key}" if prefix else key if isinstance(value, dict): flat.update(flatten_for_wandb(value, prefix=full_key)) else: flat[full_key] = value return flat def fingerprint_payload(payload: dict[str, Any]) -> str: serialized = json.dumps(payload, sort_keys=True, ensure_ascii=True, separators=(",", ":")) return hashlib.sha256(serialized.encode("utf-8")).hexdigest()[:16]