"""Configuration utilities for loading YAML configs and merging with CLI args.""" import os import sys from typing import Dict, Any, Type, TypeVar import yaml from dataclasses import fields, is_dataclass from experiment_logger.unified_logger import get_logger T = TypeVar("T") def load_yaml_config(config_path: str) -> Dict[str, Any]: """Load configuration from YAML file.""" if not os.path.exists(config_path): raise FileNotFoundError(f"Config file not found: {config_path}") with open(config_path, "r") as f: config = yaml.safe_load(f) if config is None: return {} get_logger().info(f"Loaded configuration from: {config_path}") return config def save_yaml_config(config: Dict[str, Any], config_path: str): """Save configuration to YAML file.""" os.makedirs(os.path.dirname(config_path), exist_ok=True) with open(config_path, "w") as f: yaml.dump(config, f, default_flow_style=False, indent=2, sort_keys=False) get_logger().info(f"Saved configuration to: {config_path}") def dataclass_from_dict(cls: Type[T], config_dict: Dict[str, Any]) -> T: """Create dataclass instance from dictionary, handling type conversions.""" if not is_dataclass(cls): raise ValueError(f"{cls} is not a dataclass") # Get field names and types field_map = {f.name: f for f in fields(cls)} # type: ignore # Filter config to only include valid fields filtered_config: Dict[str, Any] = {} for key, value in config_dict.items(): if key in field_map: field = field_map[key] # Handle type conversion if needed try: # Handle None values and optional types if value is None: filtered_config[key] = None elif hasattr(field.type, "__origin__") and field.type.__origin__ is type(None): # Optional type (Union[X, None]) filtered_config[key] = value else: # Try to convert to the expected type if field.type is bool and isinstance(value, str): filtered_config[key] = value.lower() in ("true", "1", "yes", "on") else: filtered_config[key] = field.type(value) if value is not None else None # type: ignore except (ValueError, TypeError) as e: get_logger().warning(f"Could not convert {key}={value} to {field.type}: {e}") filtered_config[key] = value else: get_logger().warning(f"Unknown configuration parameter: {key}") return cls(**filtered_config) def print_config(config: Any, title: str = "Configuration"): """Pretty print configuration.""" get_logger().info(f"{title}:") if is_dataclass(config): for field in fields(config): value = getattr(config, field.name) get_logger().info(f" {field.name}: {value}") else: for key, value in vars(config).items(): get_logger().info(f" {key}: {value}")