Configuration manager

Configuration Manager

Configuration manager is a component of the game coordinator that handles the configuration of the game. It is responsible for loading the configuration from the YAML file and providing it to the game coordinator.

`netsecgame.game.configuration_manager.ConfigurationManager(task_config_file=None, service_host=None, service_port=None)`

Manages the loading and access of game configuration.

Handles fetching configuration from efficient sources (local file or remote service) and provides structured access to configuration data.

Source code in netsecgame/game/configuration_manager.py

def __init__(self, task_config_file: Optional[str] = None, service_host: Optional[str] = None, service_port: Optional[int] = None):
    self.logger = logging.getLogger("ConfigurationManager")
    self._task_config_file = task_config_file
    self._service_host = service_host
    self._service_port = service_port

    self._parser: Optional[ConfigParser] = None
    self._cyst_objects = None
    self._config_file_hash: Optional[str] = None

    # Cache for parsed values
    self._starting_positions: Dict[str, Any] = {}
    self._win_conditions: Dict[str, Any] = {}
    self._goal_descriptions: Dict[str, str] = {}
    self._max_steps: Dict[str, Optional[int]] = {}

`get_all_goal_descriptions()`

Returns goal descriptions for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_goal_descriptions(self) -> Dict[str, str]:
    """Returns goal descriptions for all roles."""
    goal_descriptions = {}
    for agent_role in AgentRole:
        try:
            goal_descriptions[agent_role] = self.get_goal_description(role=agent_role)
        except KeyError:
            goal_descriptions[agent_role] = ""
        self.logger.info(f"Goal description for role '{agent_role}': {goal_descriptions[agent_role]}")
    return goal_descriptions

`get_all_max_steps()`

Returns max steps for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_max_steps(self) -> Dict[str, Optional[int]]:
    """Returns max steps for all roles."""
    # Using self.get_max_steps might raise RuntimeError if checks are there, 
    # but simpler to just call parser directly or the single accessor since we are inside the class.
    # However, the single accessor has the check.
    # But wait, self.get_max_steps(role) does `self._parser.get_max_steps(role)` already.
    # Iterating over AgentRole is correct.
    return {role: self.get_max_steps(role) for role in AgentRole}

`get_all_starting_positions()`

Returns starting positions for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_starting_positions(self) -> Dict[str, Any]:
    """Returns starting positions for all roles."""
    starting_positions = {}
    for agent_role in AgentRole:
        try:
            starting_positions[agent_role] = self.get_starting_position(role=agent_role)
            self.logger.info(f"Starting position for role '{agent_role}': {starting_positions[agent_role]}")
        except KeyError:
            starting_positions[agent_role] = {}
    return starting_positions

`get_all_win_conditions()`

Returns win conditions for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_win_conditions(self) -> Dict[str, Any]:
    """Returns win conditions for all roles."""
    win_conditions = {}
    for agent_role in AgentRole:
        try:
            win_conditions[agent_role] = self.get_win_conditions(role=agent_role)
        except KeyError:
            win_conditions[agent_role] = {}
        self.logger.info(f"Win condition for role '{agent_role}': {win_conditions[agent_role]}")
    return win_conditions

`get_goal_description(role)`

Returns the goal description for a specific role.

Source code in netsecgame/game/configuration_manager.py

def get_goal_description(self, role: str) -> str:
    """Returns the goal description for a specific role."""
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_goal_description(agent_role=role)

`get_max_steps(role)`

Returns the max steps for a specific role.

Source code in netsecgame/game/configuration_manager.py

def get_max_steps(self, role: str) -> Optional[int]:
    """Returns the max steps for a specific role."""
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_max_steps(role)

`get_rewards(reward_names=['step', 'success', 'fail', 'false_positive'], default_value=0)`

Returns the rewards configuration.

Source code in netsecgame/game/configuration_manager.py

def get_rewards(self, reward_names: List[str] = ["step", "success", "fail", "false_positive"], default_value: int = 0) -> dict:
    """Returns the rewards configuration."""
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_rewards(reward_names, default_value)

`get_starting_position(role)`

Returns the starting position configuration for a specific role.

Source code in netsecgame/game/configuration_manager.py

def get_starting_position(self, role: str) -> dict:
    """Returns the starting position configuration for a specific role."""
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_start_position(agent_role=role)

`get_win_conditions(role)`

Returns the win conditions for a specific role.

Source code in netsecgame/game/configuration_manager.py

def get_win_conditions(self, role: str) -> dict:
    """Returns the win conditions for a specific role."""
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_win_conditions(agent_role=role)

`load()` `async`

Determines the source and loads the configuration. Prioritizes remote service if configured, otherwise falls back to local file.

Source code in netsecgame/game/configuration_manager.py

async def load(self) -> None:
    """
    Determines the source and loads the configuration.
    Prioritizes remote service if configured, otherwise falls back to local file.
    """
    if self._service_host and self._service_port:
        self.logger.info(f"Fetching task configuration from {self._service_host}:{self._service_port}")
        await self._fetch_remote_configuration()
    elif self._task_config_file:
        self.logger.info(f"Loading task configuration from file: {self._task_config_file}")
        self._load_local_configuration()
    else:
        raise ValueError("Task configuration source not specified (neither file nor service)")

ConfigParser

ConfigParser is a class that is responsible for parsing the YAML configuration file and providing it to the game coordinator.

`netsecgame.game.config_parser.ConfigParser(task_config_file=None, config_dict=None)`

Class to deal with the configuration file of NetSecGame Coordinator Args: task_config_file (str|None): Path to the configuration file config_dict (dict|None): Dictionary with configuration data

Initializes the configuration parser. Required either path to a confgiuration file or a dict with configuraitons.