Spaces:
Paused
Paused
Download src/config/loader.py from ThomasHeisig/Brain-5D-Space: direct link, hf CLI and curl.
- Browser
- Download file 28.2 kB
-
https://huggingface.co/spaces/ThomasHeisig/Brain-5D-Space/resolve/main/src/config/loader.py
- Command line
-
hf download hf://spaces/ThomasHeisig/Brain-5D-Space/src/config/loader.py
-
curl -L -o loader.py https://huggingface.co/spaces/ThomasHeisig/Brain-5D-Space/resolve/main/src/config/loader.py
28.2 kB
| """ | |
| Configuration loader and validator for Brain‑5D experiments. | |
| This module provides robust loading and validation of YAML configuration files, | |
| ensuring all parameters meet the constraints required by the Brain-5D core. | |
| Features: | |
| - Full YAML configuration loading with comprehensive validation | |
| - Support for all Brain-5D configuration sections (simulation, topology, network, neuron, energy, STDP) | |
| - Clear error messages with context | |
| - Default values for optional parameters | |
| - Type-safe configuration return with TypedDict | |
| - Backward compatibility with existing configuration files | |
| Example: | |
| >>> from src.config import load_config | |
| >>> config = load_config("configs/poc_config.yaml") | |
| >>> print(config["dimensions"]) | |
| (50, 50, 50, 50, 50) | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from pathlib import Path | |
| from typing import Any, TypedDict, TypeGuard, cast | |
| import yaml | |
| from .modes import validate_modes | |
| logger = logging.getLogger(__name__) | |
| # ============================================================================ | |
| # Type Definitions | |
| # ============================================================================ | |
| class SimulationConfig(TypedDict, total=False): | |
| """Configuration for simulation parameters.""" | |
| dt_ms: float | |
| max_delay: int | |
| debug_invariants: bool | |
| class TopologyConfig(TypedDict, total=False): | |
| """Configuration for network topology.""" | |
| allow_self_connections: bool | |
| allow_parallel_connections: bool | |
| class NetworkConfig(TypedDict, total=False): | |
| """Configuration for network parameters.""" | |
| weight_min: float | |
| weight_max: float | |
| initial_connections_per_neuron: int | |
| neighbour_radius: float | |
| class NeuronConfig(TypedDict, total=False): | |
| """Configuration for neuron parameters.""" | |
| a: float | |
| b: float | |
| c: float | |
| d: float | |
| class EnergyConfig(TypedDict, total=False): | |
| """Configuration for energy parameters.""" | |
| initial: float | |
| spike_cost: float | |
| class STDPConfig(TypedDict, total=False): | |
| """Configuration for STDP parameters.""" | |
| a_plus: float | |
| a_minus: float | |
| tau_plus: float | |
| tau_minus: float | |
| enabled: bool | |
| min_weight: float | |
| max_weight: float | |
| enable_triplet: bool | |
| enable_metaplasticity: bool | |
| class RewardConfig(TypedDict, total=False): | |
| """Configuration for reward-modulated plasticity.""" | |
| reward_source: str | |
| output_spike_value: float | |
| enabled: bool | |
| learning_rate: float | |
| delay_ticks: int | |
| clamp_weights: bool | |
| reset_trace_after_reward: bool | |
| trace_epsilon: float | |
| class VisualizationConfig(TypedDict, total=False): | |
| """Configuration for visualization.""" | |
| enabled: bool | |
| refresh_interval_ticks: int | |
| class TelemetryConfig(TypedDict, total=False): | |
| """Configuration for telemetry.""" | |
| history_ticks: int | |
| spike_history_ticks: int | |
| class LoggingConfig(TypedDict, total=False): | |
| """Configuration for logging.""" | |
| interval_ticks: int | |
| class ConfigDict(TypedDict, total=False): | |
| """Complete Brain-5D configuration dictionary. | |
| All fields are optional with defaults applied during validation. | |
| """ | |
| # Required | |
| dimensions: tuple[int, int, int, int, int] | |
| initial_neurons: int | |
| # Optional sections | |
| seed: int | |
| simulation: SimulationConfig | |
| topology: TopologyConfig | |
| network: NetworkConfig | |
| neuron: NeuronConfig | |
| energy: EnergyConfig | |
| stdp: STDPConfig | |
| reward: RewardConfig | |
| visualization: VisualizationConfig | |
| telemetry: TelemetryConfig | |
| logging: LoggingConfig | |
| diagnostics: dict[str, Any] | |
| topology_input: dict[str, Any] # Backward compatibility | |
| homeostasis: dict[str, Any] | |
| self_organization: dict[str, Any] | |
| eligibility: dict[str, Any] | |
| storage: dict[str, Any] | |
| dashboard: dict[str, Any] | |
| state_mode: str | |
| observability: str | |
| # ============================================================================ | |
| # Default Configuration | |
| # ============================================================================ | |
| DEFAULT_CONFIG: ConfigDict = { | |
| "state_mode": "operator", | |
| "observability": "minimal", | |
| "simulation": { | |
| "dt_ms": 1.0, | |
| "max_delay": 5, | |
| "debug_invariants": False, | |
| }, | |
| "topology": { | |
| "allow_self_connections": False, | |
| "allow_parallel_connections": False, | |
| }, | |
| "network": { | |
| "weight_min": 0.0, | |
| "weight_max": 0.5, | |
| "initial_connections_per_neuron": 10, | |
| "neighbour_radius": 5.0, | |
| }, | |
| "neuron": { | |
| "a": 0.02, | |
| "b": 0.2, | |
| "c": -65.0, | |
| "d": 8.0, | |
| }, | |
| "energy": { | |
| "initial": 1.0, | |
| "spike_cost": 0.001, | |
| }, | |
| "stdp": { | |
| "enabled": False, | |
| "a_plus": 0.1, | |
| "a_minus": 0.12, | |
| "tau_plus": 20.0, | |
| "tau_minus": 20.0, | |
| "enable_triplet": False, | |
| "enable_metaplasticity": False, | |
| "min_weight": 0.0, | |
| "max_weight": 1.0, | |
| }, | |
| "reward": { | |
| "enabled": False, | |
| "reward_source": "external", | |
| "output_spike_value": 1.0, | |
| "learning_rate": 0.01, | |
| "delay_ticks": 0, | |
| "clamp_weights": True, | |
| "reset_trace_after_reward": False, | |
| "trace_epsilon": 1.0e-12, | |
| }, | |
| "visualization": { | |
| "enabled": False, | |
| "refresh_interval_ticks": 100, | |
| }, | |
| "telemetry": { | |
| "history_ticks": 10000, | |
| "spike_history_ticks": 1000, | |
| }, | |
| "logging": { | |
| "interval_ticks": 100, | |
| }, | |
| "seed": 42, | |
| } | |
| # ============================================================================ | |
| # Validation Functions | |
| # ============================================================================ | |
| def _is_dimension_sequence( | |
| value: object, | |
| ) -> TypeGuard[list[object] | tuple[object, ...]]: | |
| return isinstance(value, (list, tuple)) | |
| def _validate_dimensions(value: Any) -> tuple[int, int, int, int, int]: | |
| """Validate dimensions parameter.""" | |
| if not _is_dimension_sequence(value): | |
| raise ValueError("dimensions must be a list or tuple") | |
| if len(value) != 5: | |
| raise ValueError(f"dimensions must have exactly 5 values, got {len(value)}") | |
| dims: list[int] = [] | |
| for i, d in enumerate(value): | |
| if not isinstance(d, (int, float)): | |
| raise ValueError(f"dimension {i} must be numeric, got {type(d).__name__}") | |
| dim = int(d) | |
| if dim < 1 or dim > 256: | |
| raise ValueError(f"dimension {i} must be in 1..256, got {dim}") | |
| dims.append(dim) | |
| return (dims[0], dims[1], dims[2], dims[3], dims[4]) | |
| def _validate_initial_neurons( | |
| value: Any, | |
| dims: tuple[int, int, int, int, int], | |
| ) -> int: | |
| """Validate initial_neurons parameter.""" | |
| if not isinstance(value, (int, float)): | |
| raise ValueError(f"initial_neurons must be numeric, got {type(value).__name__}") | |
| initial = int(value) | |
| if initial <= 0: | |
| raise ValueError(f"initial_neurons must be positive, got {initial}") | |
| total = 1 | |
| for d in dims: | |
| total *= d | |
| if initial > total: | |
| raise ValueError( | |
| f"initial_neurons ({initial}) exceeds total positions ({total})" | |
| ) | |
| return initial | |
| def _validate_seed(value: Any) -> int: | |
| """Validate seed parameter.""" | |
| if value is None: | |
| return 42 | |
| if not isinstance(value, (int, float)): | |
| raise ValueError(f"seed must be numeric, got {type(value).__name__}") | |
| return int(value) | |
| def _validate_simulation_config( | |
| raw: dict[str, Any], | |
| defaults: SimulationConfig, | |
| ) -> SimulationConfig: | |
| """Validate and merge simulation configuration.""" | |
| result: SimulationConfig = {} | |
| # dt_ms | |
| dt_raw = raw.get("dt_ms", defaults.get("dt_ms", 1.0)) | |
| if not isinstance(dt_raw, (int, float)): | |
| raise ValueError(f"dt_ms must be numeric, got {type(dt_raw).__name__}") | |
| dt = float(dt_raw) | |
| if dt != 1.0: | |
| raise ValueError(f"Sprint 1 reference core requires dt_ms=1.0, got {dt}") | |
| result["dt_ms"] = dt | |
| # max_delay | |
| delay_raw = raw.get("max_delay", defaults.get("max_delay", 5)) | |
| if not isinstance(delay_raw, (int, float)): | |
| raise ValueError(f"max_delay must be numeric, got {type(delay_raw).__name__}") | |
| delay = int(delay_raw) | |
| if delay < 1: | |
| raise ValueError(f"max_delay must be >= 1, got {delay}") | |
| result["max_delay"] = delay | |
| # debug_invariants | |
| debug_raw = raw.get("debug_invariants", defaults.get("debug_invariants", False)) | |
| result["debug_invariants"] = bool(debug_raw) | |
| return result | |
| def _validate_topology_config( | |
| raw: dict[str, Any], | |
| defaults: TopologyConfig, | |
| ) -> TopologyConfig: | |
| """Validate and merge topology configuration.""" | |
| result: TopologyConfig = {} | |
| result["allow_self_connections"] = bool( | |
| raw.get("allow_self_connections", defaults.get("allow_self_connections", False)) | |
| ) | |
| result["allow_parallel_connections"] = bool( | |
| raw.get( | |
| "allow_parallel_connections", | |
| defaults.get("allow_parallel_connections", False), | |
| ) | |
| ) | |
| return result | |
| def _validate_network_config( | |
| raw: dict[str, Any], | |
| defaults: NetworkConfig, | |
| ) -> NetworkConfig: | |
| """Validate and merge network configuration.""" | |
| result: NetworkConfig = {} | |
| # weight_min | |
| wmin_raw = raw.get("weight_min", defaults.get("weight_min", 0.0)) | |
| if not isinstance(wmin_raw, (int, float)): | |
| raise ValueError(f"weight_min must be numeric, got {type(wmin_raw).__name__}") | |
| result["weight_min"] = float(wmin_raw) | |
| # weight_max | |
| wmax_raw = raw.get("weight_max", defaults.get("weight_max", 0.5)) | |
| if not isinstance(wmax_raw, (int, float)): | |
| raise ValueError(f"weight_max must be numeric, got {type(wmax_raw).__name__}") | |
| result["weight_max"] = float(wmax_raw) | |
| if result["weight_min"] > result["weight_max"]: | |
| raise ValueError( | |
| f"weight_min ({result['weight_min']}) > weight_max ({result['weight_max']})" | |
| ) | |
| # initial_connections_per_neuron | |
| conn_raw = raw.get( | |
| "initial_connections_per_neuron", | |
| defaults.get("initial_connections_per_neuron", 10), | |
| ) | |
| if not isinstance(conn_raw, (int, float)): | |
| raise ValueError( | |
| f"initial_connections_per_neuron must be numeric, got {type(conn_raw).__name__}" | |
| ) | |
| conn = int(conn_raw) | |
| if conn < 0: | |
| raise ValueError(f"initial_connections_per_neuron must be >= 0, got {conn}") | |
| result["initial_connections_per_neuron"] = conn | |
| # neighbour_radius | |
| radius_raw = raw.get("neighbour_radius", defaults.get("neighbour_radius", 5.0)) | |
| if not isinstance(radius_raw, (int, float)): | |
| raise ValueError( | |
| f"neighbour_radius must be numeric, got {type(radius_raw).__name__}" | |
| ) | |
| radius = float(radius_raw) | |
| if radius < 0: | |
| raise ValueError(f"neighbour_radius must be >= 0, got {radius}") | |
| result["neighbour_radius"] = radius | |
| return result | |
| def _validate_neuron_config( | |
| raw: dict[str, Any], | |
| defaults: NeuronConfig, | |
| ) -> NeuronConfig: | |
| """Validate and merge neuron configuration.""" | |
| result: NeuronConfig = {} | |
| for key in ["a", "b", "c", "d"]: | |
| value = raw.get(key, defaults.get(key)) | |
| if value is None: | |
| continue | |
| if not isinstance(value, (int, float)): | |
| raise ValueError( | |
| f"neuron.{key} must be numeric, got {type(value).__name__}" | |
| ) | |
| result[key] = float(value) # type: ignore[literal-required] | |
| return result | |
| def _validate_energy_config( | |
| raw: dict[str, Any], | |
| defaults: EnergyConfig, | |
| ) -> EnergyConfig: | |
| """Validate and merge energy configuration.""" | |
| result: EnergyConfig = {} | |
| # initial | |
| init_raw = raw.get("initial", defaults.get("initial", 1.0)) | |
| if not isinstance(init_raw, (int, float)): | |
| raise ValueError( | |
| f"energy.initial must be numeric, got {type(init_raw).__name__}" | |
| ) | |
| result["initial"] = float(init_raw) | |
| # spike_cost | |
| cost_raw = raw.get("spike_cost", defaults.get("spike_cost", 0.001)) | |
| if not isinstance(cost_raw, (int, float)): | |
| raise ValueError( | |
| f"energy.spike_cost must be numeric, got {type(cost_raw).__name__}" | |
| ) | |
| result["spike_cost"] = float(cost_raw) | |
| return result | |
| def _validate_stdp_config( | |
| raw: dict[str, Any], | |
| defaults: STDPConfig, | |
| ) -> STDPConfig: | |
| """Validate and merge STDP configuration.""" | |
| result: STDPConfig = {} | |
| for key in [ | |
| "a_plus", | |
| "a_minus", | |
| "tau_plus", | |
| "tau_minus", | |
| "min_weight", | |
| "max_weight", | |
| ]: | |
| value = raw.get(key, defaults.get(key)) | |
| if value is None: | |
| continue | |
| if not isinstance(value, (int, float)): | |
| raise ValueError(f"stdp.{key} must be numeric, got {type(value).__name__}") | |
| result[key] = float(value) # type: ignore[literal-required] | |
| # Booleans | |
| result["enabled"] = bool(raw.get("enabled", defaults.get("enabled", False))) | |
| result["enable_triplet"] = bool( | |
| raw.get("enable_triplet", defaults.get("enable_triplet", False)) | |
| ) | |
| result["enable_metaplasticity"] = bool( | |
| raw.get("enable_metaplasticity", defaults.get("enable_metaplasticity", False)) | |
| ) | |
| return result | |
| def _validate_reward_config( | |
| raw: dict[str, Any], | |
| defaults: RewardConfig, | |
| ) -> RewardConfig: | |
| """Validate and merge reward configuration.""" | |
| result: RewardConfig = {} | |
| result["enabled"] = bool(raw.get("enabled", defaults.get("enabled", False))) | |
| # reward_source | |
| source = raw.get("reward_source", defaults.get("reward_source", "external")) | |
| if not isinstance(source, str): | |
| raise ValueError(f"reward_source must be a string, got {type(source).__name__}") | |
| if source not in {"external", "output_spike"}: | |
| raise ValueError( | |
| f"reward_source must be 'external' or 'output_spike', got {source}" | |
| ) | |
| result["reward_source"] = source | |
| # output_spike_value | |
| val_raw = raw.get("output_spike_value", defaults.get("output_spike_value", 1.0)) | |
| if not isinstance(val_raw, (int, float)): | |
| raise ValueError( | |
| f"output_spike_value must be numeric, got {type(val_raw).__name__}" | |
| ) | |
| result["output_spike_value"] = float(val_raw) | |
| rate_raw = raw.get("learning_rate", defaults.get("learning_rate", 0.01)) | |
| if not isinstance(rate_raw, (int, float)): | |
| raise ValueError( | |
| f"reward.learning_rate must be numeric, got {type(rate_raw).__name__}" | |
| ) | |
| result["learning_rate"] = float(rate_raw) | |
| delay_raw = raw.get("delay_ticks", defaults.get("delay_ticks", 0)) | |
| if not isinstance(delay_raw, (int, float)): | |
| raise ValueError( | |
| f"reward.delay_ticks must be numeric, got {type(delay_raw).__name__}" | |
| ) | |
| result["delay_ticks"] = int(delay_raw) | |
| if result["delay_ticks"] < 0: | |
| raise ValueError("reward.delay_ticks must be >= 0") | |
| result["clamp_weights"] = bool( | |
| raw.get("clamp_weights", defaults.get("clamp_weights", True)) | |
| ) | |
| result["reset_trace_after_reward"] = bool( | |
| raw.get( | |
| "reset_trace_after_reward", defaults.get("reset_trace_after_reward", False) | |
| ) | |
| ) | |
| epsilon_raw = raw.get("trace_epsilon", defaults.get("trace_epsilon", 1.0e-12)) | |
| if not isinstance(epsilon_raw, (int, float)): | |
| raise ValueError( | |
| f"reward.trace_epsilon must be numeric, got {type(epsilon_raw).__name__}" | |
| ) | |
| result["trace_epsilon"] = float(epsilon_raw) | |
| return result | |
| def _validate_visualization_config( | |
| raw: dict[str, Any], | |
| defaults: VisualizationConfig, | |
| ) -> VisualizationConfig: | |
| """Validate and merge visualization configuration.""" | |
| result: VisualizationConfig = {} | |
| result["enabled"] = bool(raw.get("enabled", defaults.get("enabled", False))) | |
| refresh_raw = raw.get( | |
| "refresh_interval_ticks", defaults.get("refresh_interval_ticks", 100) | |
| ) | |
| if not isinstance(refresh_raw, (int, float)): | |
| raise ValueError( | |
| f"refresh_interval_ticks must be numeric, got {type(refresh_raw).__name__}" | |
| ) | |
| refresh = int(refresh_raw) | |
| if refresh < 1: | |
| raise ValueError(f"refresh_interval_ticks must be >= 1, got {refresh}") | |
| result["refresh_interval_ticks"] = refresh | |
| return result | |
| def _validate_telemetry_config( | |
| raw: dict[str, Any], | |
| defaults: TelemetryConfig, | |
| ) -> TelemetryConfig: | |
| """Validate and merge telemetry configuration.""" | |
| result: TelemetryConfig = {} | |
| for key in ["history_ticks", "spike_history_ticks"]: | |
| value = raw.get(key, defaults.get(key)) | |
| if value is None: | |
| continue | |
| if not isinstance(value, (int, float)): | |
| raise ValueError( | |
| f"telemetry.{key} must be numeric, got {type(value).__name__}" | |
| ) | |
| val = int(value) | |
| if val < 1: | |
| raise ValueError(f"telemetry.{key} must be >= 1, got {val}") | |
| result[key] = val # type: ignore[literal-required] | |
| return result | |
| def _validate_logging_config( | |
| raw: dict[str, Any], | |
| defaults: LoggingConfig, | |
| ) -> LoggingConfig: | |
| """Validate and merge logging configuration.""" | |
| result: LoggingConfig = {} | |
| interval_raw = raw.get("interval_ticks", defaults.get("interval_ticks", 100)) | |
| if not isinstance(interval_raw, (int, float)): | |
| raise ValueError( | |
| f"interval_ticks must be numeric, got {type(interval_raw).__name__}" | |
| ) | |
| interval = int(interval_raw) | |
| if interval < 1: | |
| raise ValueError(f"interval_ticks must be >= 1, got {interval}") | |
| result["interval_ticks"] = interval | |
| return result | |
| # ============================================================================ | |
| # Main Loader | |
| # ============================================================================ | |
| def load_config( | |
| path: str | Path, | |
| _apply_defaults: bool = True, | |
| ) -> ConfigDict: | |
| """ | |
| Load and validate the experiment configuration from a YAML file. | |
| Args: | |
| path: Path to the YAML configuration file. | |
| apply_defaults: Whether to fill missing values with defaults. | |
| Returns: | |
| ConfigDict: Fully validated and merged configuration. | |
| Raises: | |
| ValueError: On validation failure with detailed error message. | |
| FileNotFoundError: If the file does not exist. | |
| yaml.YAMLError: On YAML parse errors. | |
| Example: | |
| >>> config = load_config("configs/poc_config.yaml") | |
| >>> print(config["dimensions"]) | |
| (50, 50, 50, 50, 50) | |
| """ | |
| path = Path(path) | |
| if not path.exists(): | |
| raise FileNotFoundError(f"Configuration file not found: {path}") | |
| with path.open("r", encoding="utf-8") as f: | |
| try: | |
| raw = yaml.safe_load(f) | |
| except yaml.YAMLError as e: | |
| raise yaml.YAMLError(f"Failed to parse YAML file {path}: {e}") from e | |
| if raw is None: | |
| raw = {} | |
| elif not isinstance(raw, dict): | |
| raise ValueError("Configuration file must contain a YAML dictionary") | |
| raw_dict = cast("dict[str, Any]", raw) | |
| # ------------------------------------------------------------------------ | |
| # Required fields | |
| # ------------------------------------------------------------------------ | |
| dims = _validate_dimensions(raw_dict.get("dimensions")) | |
| initial_neurons = _validate_initial_neurons(raw_dict.get("initial_neurons"), dims) | |
| result: ConfigDict = { | |
| "dimensions": dims, | |
| "initial_neurons": initial_neurons, | |
| } | |
| # ------------------------------------------------------------------------ | |
| # Optional fields with defaults | |
| # ------------------------------------------------------------------------ | |
| defaults = cast("dict[str, Any]", DEFAULT_CONFIG) | |
| # Seed | |
| result["seed"] = _validate_seed(raw_dict.get("seed", defaults.get("seed", 42))) | |
| # Orthogonal runtime axes | |
| state_mode = cast( | |
| "str", raw_dict.get("state_mode", defaults.get("state_mode", "operator")) | |
| ) | |
| observability = cast( | |
| "str", | |
| raw_dict.get("observability", defaults.get("observability", "minimal")), | |
| ) | |
| validate_modes(state_mode, observability) | |
| result["state_mode"] = state_mode | |
| result["observability"] = observability | |
| # Simulation | |
| sim_raw = raw_dict.get("simulation", {}) | |
| if not isinstance(sim_raw, dict): | |
| raise ValueError("simulation section must be a dictionary") | |
| result["simulation"] = _validate_simulation_config( | |
| cast("dict[str, Any]", sim_raw), | |
| cast("SimulationConfig", defaults["simulation"]), | |
| ) | |
| # Topology | |
| topo_raw = raw_dict.get("topology", {}) | |
| if not isinstance(topo_raw, dict): | |
| raise ValueError("topology section must be a dictionary") | |
| result["topology"] = _validate_topology_config( | |
| cast("dict[str, Any]", topo_raw), cast("TopologyConfig", defaults["topology"]) | |
| ) | |
| # Network | |
| net_raw = raw_dict.get("network", {}) | |
| if not isinstance(net_raw, dict): | |
| raise ValueError("network section must be a dictionary") | |
| result["network"] = _validate_network_config( | |
| cast("dict[str, Any]", net_raw), cast("NetworkConfig", defaults["network"]) | |
| ) | |
| # Neuron | |
| neuron_raw = raw_dict.get("neuron", {}) | |
| if not isinstance(neuron_raw, dict): | |
| raise ValueError("neuron section must be a dictionary") | |
| result["neuron"] = _validate_neuron_config( | |
| cast("dict[str, Any]", neuron_raw), cast("NeuronConfig", defaults["neuron"]) | |
| ) | |
| # Energy | |
| energy_raw = raw_dict.get("energy", {}) | |
| if not isinstance(energy_raw, dict): | |
| raise ValueError("energy section must be a dictionary") | |
| result["energy"] = _validate_energy_config( | |
| cast("dict[str, Any]", energy_raw), cast("EnergyConfig", defaults["energy"]) | |
| ) | |
| # STDP | |
| stdp_raw = raw_dict.get("stdp", {}) | |
| if not isinstance(stdp_raw, dict): | |
| raise ValueError("stdp section must be a dictionary") | |
| result["stdp"] = _validate_stdp_config( | |
| cast("dict[str, Any]", stdp_raw), cast("STDPConfig", defaults["stdp"]) | |
| ) | |
| # Reward | |
| reward_raw = raw_dict.get("reward", {}) | |
| if not isinstance(reward_raw, dict): | |
| raise ValueError("reward section must be a dictionary") | |
| result["reward"] = _validate_reward_config( | |
| cast("dict[str, Any]", reward_raw), cast("RewardConfig", defaults["reward"]) | |
| ) | |
| # Visualization | |
| vis_raw = raw_dict.get("visualization", {}) | |
| if not isinstance(vis_raw, dict): | |
| raise ValueError("visualization section must be a dictionary") | |
| result["visualization"] = _validate_visualization_config( | |
| cast("dict[str, Any]", vis_raw), | |
| cast("VisualizationConfig", defaults["visualization"]), | |
| ) | |
| # Telemetry | |
| tele_raw = raw_dict.get("telemetry", {}) | |
| if not isinstance(tele_raw, dict): | |
| raise ValueError("telemetry section must be a dictionary") | |
| result["telemetry"] = _validate_telemetry_config( | |
| cast("dict[str, Any]", tele_raw), cast("TelemetryConfig", defaults["telemetry"]) | |
| ) | |
| # Logging | |
| log_raw = raw_dict.get("logging", {}) | |
| if not isinstance(log_raw, dict): | |
| raise ValueError("logging section must be a dictionary") | |
| result["logging"] = _validate_logging_config( | |
| cast("dict[str, Any]", log_raw), cast("LoggingConfig", defaults["logging"]) | |
| ) | |
| # Diagnostics (passthrough, optional) | |
| if "diagnostics" in raw_dict: | |
| if not isinstance(raw_dict["diagnostics"], dict): | |
| raise ValueError("diagnostics section must be a dictionary") | |
| result["diagnostics"] = raw_dict["diagnostics"] | |
| # Homeostasis (passthrough, optional) | |
| if "homeostasis" in raw_dict: | |
| if not isinstance(raw_dict["homeostasis"], dict): | |
| raise ValueError("homeostasis section must be a dictionary") | |
| result["homeostasis"] = raw_dict["homeostasis"] | |
| # Self-organization (passthrough, optional) | |
| if "self_organization" in raw_dict: | |
| if not isinstance(raw_dict["self_organization"], dict): | |
| raise ValueError("self_organization section must be a dictionary") | |
| result["self_organization"] = raw_dict["self_organization"] | |
| # Eligibility (passthrough, optional) | |
| if "eligibility" in raw_dict: | |
| if not isinstance(raw_dict["eligibility"], dict): | |
| raise ValueError("eligibility section must be a dictionary") | |
| result["eligibility"] = raw_dict["eligibility"] | |
| # Storage (passthrough, optional) | |
| if "storage" in raw_dict: | |
| if not isinstance(raw_dict["storage"], dict): | |
| raise ValueError("storage section must be a dictionary") | |
| result["storage"] = raw_dict["storage"] | |
| # Dashboard (passthrough, optional). This is runtime configuration: | |
| # telemetry capture and state publication must match the loaded profile. | |
| if "dashboard" in raw_dict: | |
| if not isinstance(raw_dict["dashboard"], dict): | |
| raise ValueError("dashboard section must be a dictionary") | |
| result["dashboard"] = raw_dict["dashboard"] | |
| # Topology input (backward compatibility) | |
| if "topology" in raw_dict and "input" in raw_dict["topology"]: | |
| result["topology_input"] = raw_dict["topology"]["input"] | |
| logger.info(f"Loaded configuration from {path}") | |
| logger.debug(f"Configuration: {result}") | |
| return result | |
| def validate_config(config: ConfigDict) -> None: | |
| """ | |
| Validate a configuration dictionary without loading from file. | |
| Useful for testing or for validating programmatically generated configs. | |
| Args: | |
| config: Configuration dictionary to validate. | |
| Raises: | |
| ValueError: On validation failure. | |
| """ | |
| # Re-validate dimensions | |
| dims = _validate_dimensions(config.get("dimensions")) | |
| _validate_initial_neurons(config.get("initial_neurons"), dims) | |
| state_mode = config.get("state_mode", "operator") | |
| observability = config.get("observability", "minimal") | |
| validate_modes(state_mode, observability) | |
| # Validate each section (will raise on errors) | |
| defaults = cast("dict[str, Any]", DEFAULT_CONFIG) | |
| sim = config.get("simulation") | |
| if sim: | |
| _validate_simulation_config( | |
| cast("dict[str, Any]", sim), | |
| cast("SimulationConfig", defaults["simulation"]), | |
| ) | |
| topo = config.get("topology") | |
| if topo: | |
| _validate_topology_config( | |
| cast("dict[str, Any]", topo), cast("TopologyConfig", defaults["topology"]) | |
| ) | |
| net = config.get("network") | |
| if net: | |
| _validate_network_config( | |
| cast("dict[str, Any]", net), cast("NetworkConfig", defaults["network"]) | |
| ) | |
| # ============================================================================ | |
| # Helper Functions | |
| # ============================================================================ | |
| def config_to_dict(config: ConfigDict) -> dict[str, Any]: | |
| """Convert ConfigDict to a plain dictionary (for serialization).""" | |
| return {k: v for k, v in config.items() if v is not None} | |
| def save_config(config: ConfigDict, path: str | Path) -> None: | |
| """ | |
| Save a configuration to a YAML file. | |
| Args: | |
| config: Configuration dictionary to save. | |
| path: Path where to save the configuration. | |
| """ | |
| path = Path(path) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| # Convert tuples to lists for YAML compatibility | |
| data = config_to_dict(config) | |
| if "dimensions" in data: | |
| data["dimensions"] = list(data["dimensions"]) | |
| with path.open("w", encoding="utf-8") as f: | |
| yaml.dump(data, f, default_flow_style=False, sort_keys=False) | |
| logger.info(f"Saved configuration to {path}") | |
| # ============================================================================ | |
| # Module Exports | |
| # ============================================================================ | |
| __all__ = [ | |
| "DEFAULT_CONFIG", | |
| "ConfigDict", | |
| "EnergyConfig", | |
| "LoggingConfig", | |
| "NetworkConfig", | |
| "NeuronConfig", | |
| "RewardConfig", | |
| "STDPConfig", | |
| "SimulationConfig", | |
| "TelemetryConfig", | |
| "TopologyConfig", | |
| "VisualizationConfig", | |
| "config_to_dict", | |
| "load_config", | |
| "save_config", | |
| "validate_config", | |
| ] | |