| 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] |
|
|
