# ictonyx/config.py
from typing import Any, Dict, ItemsView, KeysView, List, Optional, Union, ValuesView
import numpy as np
[docs]
class ModelConfig:
"""Training configuration container with validation and factory methods.
Stores hyperparameters as a flat dictionary with typed property
accessors for common parameters (epochs, batch_size, learning_rate).
Supports dictionary-style access, method chaining, deep copying, and
parameter validation.
Use the factory classmethods to create pre-configured defaults for
common model types.
Args:
params: Dictionary of hyperparameters. If ``None``, an empty
config is created.
Example::
config = ModelConfig({'epochs': 20, 'batch_size': 64})
config.learning_rate = 0.001
config.set('dropout', 0.3).set('optimizer', 'adam')
See Also:
:meth:`from_defaults`, :meth:`for_cnn`, :meth:`for_xgboost`,
:meth:`for_variability_study`
"""
[docs]
def __init__(self, params: Optional[Dict[str, Any]] = None):
"""Initialize the config with a dictionary of parameters."""
self.params = params if params is not None else {}
self._frozen = False
def __repr__(self) -> str:
"""Provides a clean string representation."""
return f"ModelConfig({self.params})"
def __getitem__(self, key: str) -> Any:
"""Allow dictionary-style access like `config['epochs']`."""
if key not in self.params:
raise KeyError(
f"Parameter '{key}' not found in config. Available: {list(self.params.keys())}"
)
return self.params[key]
def __setitem__(self, key: str, value: Any):
"""Allow dictionary-style setting like `config['epochs'] = 10`."""
if self._frozen:
raise RuntimeError(
"ModelConfig is frozen and cannot be modified. "
"Create a copy with config.copy() before making changes."
)
self.params[key] = value
def __contains__(self, key: str) -> bool:
"""Support 'in' operator: `'epochs' in config`."""
return key in self.params
def __iter__(self):
"""Iterate over parameter keys.
Enables ``dict(config)`` and ``for key in config:``.
"""
return iter(self.params)
def __len__(self) -> int:
"""Return the number of parameters. Enables ``len(config)``."""
return len(self.params)
def __eq__(self, other: object) -> bool:
"""Compare by parameter contents.
Equal to another ``ModelConfig`` or a plain ``dict`` if their
parameter dictionaries are identical.
"""
if isinstance(other, ModelConfig):
return self.params == other.params
if isinstance(other, dict):
return self.params == other
return NotImplemented
__hash__ = None # type: ignore[assignment]
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Return a shallow copy of the parameters dictionary."""
return dict(self.params)
[docs]
def get(self, key: str, default: Any = None) -> Any:
"""Get parameter with optional default value."""
return self.params.get(key, default)
[docs]
def set(self, key: str, value: Any) -> "ModelConfig":
"""Set parameter and return self for chaining."""
if self._frozen:
raise RuntimeError(
"ModelConfig is frozen and cannot be modified. "
"Create a copy with config.copy() before making changes."
)
self.params[key] = value
return self
[docs]
def update(self, other_params: Dict[str, Any]) -> "ModelConfig":
"""Update multiple parameters at once and return self for chaining."""
if self._frozen:
raise RuntimeError(
"ModelConfig is frozen and cannot be modified. "
"Create a copy with config.copy() before making changes."
)
self.params.update(other_params)
return self
[docs]
def merge(self, other_params: Dict[str, Any]) -> "ModelConfig":
"""Alias for :meth:`update`.
.. deprecated:: 0.3.9
``merge()`` is identical to ``update()`` and will be removed
in v0.5.0. Use ``update()`` instead.
"""
import warnings
warnings.warn(
"ModelConfig.merge() is deprecated and will be removed in "
"v0.5.0. Use ModelConfig.update() instead.",
UserWarning,
stacklevel=2,
)
return self.update(other_params)
[docs]
def has(self, key: str) -> bool:
"""Check if parameter exists.
.. deprecated:: 0.3.9
Use the ``in`` operator instead: ``'key' in config``.
Will be removed in v0.5.0.
"""
import warnings
warnings.warn(
"ModelConfig.has() is deprecated and will be removed in "
"v0.5.0. Use the 'in' operator instead: 'key' in config.",
UserWarning,
stacklevel=2,
)
return key in self.params
[docs]
def keys(self) -> KeysView[str]:
"""Get all parameter keys."""
return self.params.keys()
[docs]
def values(self) -> ValuesView[Any]:
"""Get all parameter values."""
return self.params.values()
[docs]
def items(self) -> ItemsView[str, Any]:
"""Get all parameter key-value pairs."""
return self.params.items()
[docs]
def copy(self) -> "ModelConfig":
"""Create a deep copy of the configuration.
The frozen state is preserved: a frozen config produces a frozen copy.
"""
import copy
new = ModelConfig(copy.deepcopy(self.params))
if self._frozen:
new.freeze()
return new
[docs]
def freeze(self) -> "ModelConfig":
"""Make this config read-only to prevent mutation after construction.
Once frozen, any attempt to set or update parameters raises
``RuntimeError``. Returns ``self`` for method chaining.
Example::
config = ModelConfig({"epochs": 10}).freeze()
config.set("epochs", 20) # raises RuntimeError
"""
self._frozen = True
return self
[docs]
def validate_required(self, required_params: List[str]) -> List[str]:
"""
Check if required parameters are present.
Args:
required_params: List of parameter names that must be present
Returns:
List of missing parameter names (empty if all present)
"""
missing = [param for param in required_params if param not in self.params]
return missing
[docs]
def validate_types(self, type_specs: Dict[str, type]) -> List[str]:
"""
Validate parameter types.
Args:
type_specs: Dictionary mapping parameter names to expected types
Returns:
List of validation error messages (empty if all valid)
"""
errors = []
for param_name, expected_type in type_specs.items():
if param_name in self.params:
value = self.params[param_name]
if not isinstance(value, expected_type):
errors.append(
f"Parameter '{param_name}' should be {expected_type.__name__}, got {type(value).__name__}"
)
return errors
# Common parameter properties with validation
@property
def epochs(self) -> Optional[int]:
"""Get epochs parameter."""
return self.params.get("epochs")
@epochs.setter
def epochs(self, value: int):
"""Set epochs parameter with validation."""
if not isinstance(value, (int, np.integer)) or value <= 0:
raise ValueError(f"epochs must be a positive integer, got {value}")
self.params["epochs"] = value
@property
def batch_size(self) -> Optional[int]:
"""Get batch_size parameter."""
return self.params.get("batch_size")
@batch_size.setter
def batch_size(self, value: int):
"""Set batch_size parameter with validation."""
if not isinstance(value, (int, np.integer)) or value <= 0:
raise ValueError(f"batch_size must be a positive integer, got {value}")
self.params["batch_size"] = value
@property
def learning_rate(self) -> Optional[float]:
"""Get learning_rate parameter."""
return self.params.get("learning_rate")
@learning_rate.setter
def learning_rate(self, value: Union[float, int]):
"""Set learning_rate parameter with validation."""
if not isinstance(value, (float, int, np.integer)) or value <= 0:
raise ValueError(f"learning_rate must be a positive number, got {value}")
self.params["learning_rate"] = float(value)
@property
def verbose(self) -> Optional[int]:
"""Get verbose parameter."""
return self.params.get("verbose")
@verbose.setter
def verbose(self, value: int):
"""Set verbose parameter with validation."""
if not isinstance(value, (int, np.integer)) or value < 0:
raise ValueError(f"verbose must be a non-negative integer, got {value}")
self.params["verbose"] = value
# NEW: Experiment-specific properties for runners.py
@property
def num_runs(self) -> Optional[int]:
"""Get num_runs parameter for variability studies."""
return self.params.get("num_runs")
@num_runs.setter
def num_runs(self, value: int):
"""Set num_runs parameter with validation."""
if not isinstance(value, (int, np.integer)) or value <= 0:
raise ValueError(f"num_runs must be a positive integer, got {value}")
self.params["num_runs"] = value
@property
def epochs_per_run(self) -> Optional[int]:
"""Get epochs_per_run parameter for variability studies."""
return self.params.get("epochs_per_run")
@epochs_per_run.setter
def epochs_per_run(self, value: int):
"""Set epochs_per_run parameter with validation."""
if not isinstance(value, (int, np.integer)) or value <= 0:
raise ValueError(f"epochs_per_run must be a positive integer, got {value}")
self.params["epochs_per_run"] = value
# Factory Methods for Smart Defaults
[docs]
@classmethod
def from_defaults(cls) -> "ModelConfig":
"""Create a config with sensible defaults for a generic training run.
Returns:
A :class:`ModelConfig` with ``epochs=10``, ``batch_size=32``,
``learning_rate=0.001``, ``verbose=0``, and
``cleanup_threshold=0.8``.
"""
return cls(
{
"epochs": 10,
"batch_size": 32,
"learning_rate": 0.001,
"verbose": 0,
"cleanup_threshold": 0.8,
}
)
[docs]
@classmethod
def for_cnn(cls, input_shape: tuple = (128, 128, 3), num_classes: int = 10) -> "ModelConfig":
"""Create a config with defaults suitable for a CNN model.
Extends :meth:`from_defaults` with CNN-specific parameters and a
more aggressive cleanup threshold (0.75) for memory-intensive models.
Args:
input_shape: Spatial dimensions and channels, e.g. ``(64, 64, 3)``.
num_classes: Number of output classes.
Returns:
A :class:`ModelConfig` with CNN-appropriate defaults.
"""
base_config = cls.from_defaults()
base_config.update(
{
"input_shape": input_shape,
"num_classes": num_classes,
"loss": "categorical_crossentropy",
"metrics": ["accuracy"],
"cleanup_threshold": 0.75, # More aggressive for memory-intensive CNNs
}
)
return base_config
[docs]
@classmethod
def for_xgboost(cls, num_classes: int = 10) -> "ModelConfig":
"""Create a config with defaults suitable for an XGBoost model.
Extends :meth:`from_defaults` with ``n_estimators=180``,
``max_depth=7``, and the appropriate objective function
(``binary:logistic`` for 2 classes, ``multi:softprob`` otherwise).
Args:
num_classes: Number of output classes.
Returns:
A :class:`ModelConfig` with XGBoost-appropriate defaults.
"""
base_config = cls.from_defaults()
base_config.update(
{
"n_estimators": 180,
"max_depth": 7,
"objective": "multi:softprob" if num_classes > 2 else "binary:logistic",
}
)
return base_config
[docs]
@classmethod
def for_variability_study(cls, base_config: "ModelConfig", num_runs: int = 20) -> "ModelConfig":
"""Create a config for a variability study from an existing base config.
Copies the base config and adds study-specific parameters
(``num_runs``, ``epochs_per_run``).
Args:
base_config: An existing :class:`ModelConfig` to use as a template.
num_runs: Number of independent training runs. Default 20.
Returns:
A new :class:`ModelConfig` with study parameters added.
"""
study_config = base_config.copy()
study_config.set("num_runs", num_runs)
study_config.set("epochs_per_run", base_config.get("epochs", 10))
return study_config
@property
def cleanup_threshold(self) -> Optional[float]:
"""Get cleanup_threshold parameter."""
return self.params.get("cleanup_threshold")
@cleanup_threshold.setter
def cleanup_threshold(self, value: float):
"""Set cleanup threshold with validation."""
if not isinstance(value, (float, int, np.integer)) or not 0.1 <= value <= 1.0:
raise ValueError("cleanup_threshold must be between 0.1 and 1.0")
self.params["cleanup_threshold"] = float(value)