""" Common support for configurable CORE objects. """ import logging from collections import OrderedDict from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Optional, Union from core.emane.nodes import EmaneNet from core.emulator.enumerations import ConfigDataTypes from core.errors import CoreConfigError from core.nodes.network import WlanNode logger = logging.getLogger(__name__) if TYPE_CHECKING: from core.location.mobility import WirelessModel WirelessModelType = type[WirelessModel] _BOOL_OPTIONS: set[str] = {"0", "1"} @dataclass class ConfigGroup: """ Defines configuration group tabs used for display by ConfigurationOptions. """ name: str start: int stop: int @dataclass class Configuration: """ Represents a configuration option. """ id: str type: ConfigDataTypes label: str = None default: str = "" options: list[str] = field(default_factory=list) group: str = "Configuration" def __post_init__(self) -> None: self.label = self.label if self.label else self.id if self.type == ConfigDataTypes.BOOL: if self.default and self.default not in _BOOL_OPTIONS: raise CoreConfigError( f"{self.id} bool value must be one of: {_BOOL_OPTIONS}: " f"{self.default}" ) elif self.type == ConfigDataTypes.FLOAT: if self.default: try: float(self.default) except ValueError: raise CoreConfigError( f"{self.id} is not a valid float: {self.default}" ) elif self.type != ConfigDataTypes.STRING: if self.default: try: int(self.default) except ValueError: raise CoreConfigError( f"{self.id} is not a valid int: {self.default}" ) @dataclass class ConfigBool(Configuration): """ Represents a boolean configuration option. """ type: ConfigDataTypes = ConfigDataTypes.BOOL value: bool = False @dataclass class ConfigFloat(Configuration): """ Represents a float configuration option. """ type: ConfigDataTypes = ConfigDataTypes.FLOAT value: float = 0.0 @dataclass class ConfigInt(Configuration): """ Represents an integer configuration option. """ type: ConfigDataTypes = ConfigDataTypes.INT32 value: int = 0 @dataclass class ConfigString(Configuration): """ Represents a string configuration option. """ type: ConfigDataTypes = ConfigDataTypes.STRING value: str = "" class ConfigurableOptions: """ Provides a base for defining configuration options within CORE. """ name: Optional[str] = None options: list[Configuration] = [] @classmethod def configurations(cls) -> list[Configuration]: """ Provides the configurations for this class. :return: configurations """ return cls.options @classmethod def config_groups(cls) -> list[ConfigGroup]: """ Defines how configurations are grouped. :return: configuration group definition """ return [ConfigGroup("Options", 1, len(cls.configurations()))] @classmethod def default_values(cls) -> dict[str, str]: """ Provides an ordered mapping of configuration keys to default values. :return: ordered configuration mapping default values """ return OrderedDict( [(config.id, config.default) for config in cls.configurations()] ) class ConfigurableManager: """ Provides convenience methods for storing and retrieving configuration options for nodes. """ _default_node: int = -1 _default_type: int = _default_node def __init__(self) -> None: """ Creates a ConfigurableManager object. """ self.node_configurations = {} def nodes(self) -> list[int]: """ Retrieves the ids of all node configurations known by this manager. :return: list of node ids """ return [x for x in self.node_configurations if x != self._default_node] def config_reset(self, node_id: int = None) -> None: """ Clears all configurations or configuration for a specific node. :param node_id: node id to clear configurations for, default is None and clears all configurations :return: nothing """ if not node_id: self.node_configurations.clear() elif node_id in self.node_configurations: self.node_configurations.pop(node_id) def set_config( self, _id: str, value: str, node_id: int = _default_node, config_type: str = _default_type, ) -> None: """ Set a specific configuration value for a node and configuration type. :param _id: configuration key :param value: configuration value :param node_id: node id to store configuration for :param config_type: configuration type to store configuration for :return: nothing """ node_configs = self.node_configurations.setdefault(node_id, OrderedDict()) node_type_configs = node_configs.setdefault(config_type, OrderedDict()) node_type_configs[_id] = value def set_configs( self, config: dict[str, str], node_id: int = _default_node, config_type: str = _default_type, ) -> None: """ Set configurations for a node and configuration type. :param config: configurations to set :param node_id: node id to store configuration for :param config_type: configuration type to store configuration for :return: nothing """ logger.debug( "setting config for node(%s) type(%s): %s", node_id, config_type, config ) node_configs = self.node_configurations.setdefault(node_id, OrderedDict()) node_configs[config_type] = config def get_config( self, _id: str, node_id: int = _default_node, config_type: str = _default_type, default: str = None, ) -> str: """ Retrieves a specific configuration for a node and configuration type. :param _id: specific configuration to retrieve :param node_id: node id to store configuration for :param config_type: configuration type to store configuration for :param default: default value to return when value is not found :return: configuration value """ result = default node_type_configs = self.get_configs(node_id, config_type) if node_type_configs: result = node_type_configs.get(_id, default) return result def get_configs( self, node_id: int = _default_node, config_type: str = _default_type ) -> Optional[dict[str, str]]: """ Retrieve configurations for a node and configuration type. :param node_id: node id to store configuration for :param config_type: configuration type to store configuration for :return: configurations """ result = None node_configs = self.node_configurations.get(node_id) if node_configs: result = node_configs.get(config_type) return result def get_all_configs(self, node_id: int = _default_node) -> dict[str, Any]: """ Retrieve all current configuration types for a node. :param node_id: node id to retrieve configurations for :return: all configuration types for a node """ return self.node_configurations.get(node_id) class ModelManager(ConfigurableManager): """ Helps handle setting models for nodes and managing their model configurations. """ def __init__(self) -> None: """ Creates a ModelManager object. """ super().__init__() self.models: dict[str, Any] = {} self.node_models: dict[int, str] = {} def set_model_config( self, node_id: int, model_name: str, config: dict[str, str] = None ) -> None: """ Set configuration data for a model. :param node_id: node id to set model configuration for :param model_name: model to set configuration for :param config: configuration data to set for model :return: nothing """ # get model class to configure model_class = self.models.get(model_name) if not model_class: raise ValueError(f"{model_name} is an invalid model") # retrieve default values model_config = self.get_model_config(node_id, model_name) if not config: config = {} for key in config: value = config[key] model_config[key] = value # set as node model for startup self.node_models[node_id] = model_name # set configuration self.set_configs(model_config, node_id=node_id, config_type=model_name) def get_model_config(self, node_id: int, model_name: str) -> dict[str, str]: """ Retrieve configuration data for a model. :param node_id: node id to set model configuration for :param model_name: model to set configuration for :return: current model configuration for node """ # get model class to configure model_class = self.models.get(model_name) if not model_class: raise ValueError(f"{model_name} is an invalid model") config = self.get_configs(node_id=node_id, config_type=model_name) if not config: # set default values, when not already set config = model_class.default_values() self.set_configs(config, node_id=node_id, config_type=model_name) return config def set_model( self, node: Union[WlanNode, EmaneNet], model_class: "WirelessModelType", config: dict[str, str] = None, ) -> None: """ Set model and model configuration for node. :param node: node to set model for :param model_class: model class to set for node :param config: model configuration, None for default configuration :return: nothing """ logger.debug( "setting model(%s) for node(%s): %s", model_class.name, node.id, config ) self.set_model_config(node.id, model_class.name, config) config = self.get_model_config(node.id, model_class.name) node.setmodel(model_class, config) def get_models( self, node: Union[WlanNode, EmaneNet] ) -> list[tuple[type, dict[str, str]]]: """ Return a list of model classes and values for a net if one has been configured. This is invoked when exporting a session to XML. :param node: network node to get models for :return: list of model and values tuples for the network node """ all_configs = self.get_all_configs(node.id) if not all_configs: all_configs = {} models = [] for model_name in all_configs: config = all_configs[model_name] if model_name == ModelManager._default_node: continue model_class = self.models[model_name] models.append((model_class, config)) logger.debug("models for node(%s): %s", node.id, models) return models