Merge branch 'dev' into feature/2706-Terminal_Sim_Component

src/primaite/game/agent/actions.py

@@ -49,7 +49,7 @@ class AbstractAction(ABC):
         objects."""
 
     @abstractmethod
-    def form_request(self) -> List[str]:
+    def form_request(self) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         return []
 
@@ -67,7 +67,7 @@ class DoNothingAction(AbstractAction):
         # i.e. a choice between one option. To make enumerating this action easier, we are adding a 'dummy' paramter
         # with one option. This just aids the Action Manager to enumerate all possibilities.
 
-    def form_request(self, **kwargs) -> List[str]:
+    def form_request(self, **kwargs) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         return ["do_nothing"]
 
@@ -86,7 +86,7 @@ class NodeServiceAbstractAction(AbstractAction):
         self.shape: Dict[str, int] = {"node_id": num_nodes, "service_id": num_services}
         self.verb: str  # define but don't initialise: defends against children classes not defining this
 
-    def form_request(self, node_id: int, service_id: int) -> List[str]:
+    def form_request(self, node_id: int, service_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         service_name = self.manager.get_service_name_by_idx(node_id, service_id)
@@ -181,7 +181,7 @@ class NodeApplicationAbstractAction(AbstractAction):
         self.shape: Dict[str, int] = {"node_id": num_nodes, "application_id": num_applications}
         self.verb: str  # define but don't initialise: defends against children classes not defining this
 
-    def form_request(self, node_id: int, application_id: int) -> List[str]:
+    def form_request(self, node_id: int, application_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         application_name = self.manager.get_application_name_by_idx(node_id, application_id)
@@ -229,7 +229,7 @@ class NodeApplicationInstallAction(AbstractAction):
         super().__init__(manager=manager)
         self.shape: Dict[str, int] = {"node_id": num_nodes}
 
-    def form_request(self, node_id: int, application_name: str) -> List[str]:
+    def form_request(self, node_id: int, application_name: str) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         if node_name is None:
@@ -324,7 +324,7 @@ class NodeApplicationRemoveAction(AbstractAction):
         super().__init__(manager=manager)
         self.shape: Dict[str, int] = {"node_id": num_nodes}
 
-    def form_request(self, node_id: int, application_name: str) -> List[str]:
+    def form_request(self, node_id: int, application_name: str) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         if node_name is None:
@@ -346,7 +346,7 @@ class NodeFolderAbstractAction(AbstractAction):
         self.shape: Dict[str, int] = {"node_id": num_nodes, "folder_id": num_folders}
         self.verb: str  # define but don't initialise: defends against children classes not defining this
 
-    def form_request(self, node_id: int, folder_id: int) -> List[str]:
+    def form_request(self, node_id: int, folder_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         folder_name = self.manager.get_folder_name_by_idx(node_idx=node_id, folder_idx=folder_id)
@@ -394,7 +394,9 @@ class NodeFileCreateAction(AbstractAction):
         super().__init__(manager, num_nodes=num_nodes, num_folders=num_folders, **kwargs)
         self.verb: str = "create"
 
-    def form_request(self, node_id: int, folder_name: str, file_name: str, force: Optional[bool] = False) -> List[str]:
+    def form_request(
+        self, node_id: int, folder_name: str, file_name: str, force: Optional[bool] = False
+    ) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         if node_name is None or folder_name is None or file_name is None:
@@ -409,7 +411,7 @@ class NodeFolderCreateAction(AbstractAction):
         super().__init__(manager, num_nodes=num_nodes, num_folders=num_folders, **kwargs)
         self.verb: str = "create"
 
-    def form_request(self, node_id: int, folder_name: str) -> List[str]:
+    def form_request(self, node_id: int, folder_name: str) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         if node_name is None or folder_name is None:
@@ -430,7 +432,7 @@ class NodeFileAbstractAction(AbstractAction):
         self.shape: Dict[str, int] = {"node_id": num_nodes, "folder_id": num_folders, "file_id": num_files}
         self.verb: str  # define but don't initialise: defends against children classes not defining this
 
-    def form_request(self, node_id: int, folder_id: int, file_id: int) -> List[str]:
+    def form_request(self, node_id: int, folder_id: int, file_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         folder_name = self.manager.get_folder_name_by_idx(node_idx=node_id, folder_idx=folder_id)
@@ -463,7 +465,7 @@ class NodeFileDeleteAction(NodeFileAbstractAction):
         super().__init__(manager, num_nodes=num_nodes, num_folders=num_folders, num_files=num_files, **kwargs)
         self.verb: str = "delete"
 
-    def form_request(self, node_id: int, folder_id: int, file_id: int) -> List[str]:
+    def form_request(self, node_id: int, folder_id: int, file_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         folder_name = self.manager.get_folder_name_by_idx(node_idx=node_id, folder_idx=folder_id)
@@ -504,7 +506,7 @@ class NodeFileAccessAction(AbstractAction):
         super().__init__(manager, num_nodes=num_nodes, num_folders=num_folders, **kwargs)
         self.verb: str = "access"
 
-    def form_request(self, node_id: int, folder_name: str, file_name: str) -> List[str]:
+    def form_request(self, node_id: int, folder_name: str, file_name: str) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         if node_name is None or folder_name is None or file_name is None:
@@ -525,7 +527,7 @@ class NodeAbstractAction(AbstractAction):
         self.shape: Dict[str, int] = {"node_id": num_nodes}
         self.verb: str  # define but don't initialise: defends against children classes not defining this
 
-    def form_request(self, node_id: int) -> List[str]:
+    def form_request(self, node_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_id)
         return ["network", "node", node_name, self.verb]
@@ -740,7 +742,7 @@ class RouterACLRemoveRuleAction(AbstractAction):
         super().__init__(manager=manager)
         self.shape: Dict[str, int] = {"position": max_acl_rules}
 
-    def form_request(self, target_router: str, position: int) -> List[str]:
+    def form_request(self, target_router: str, position: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         return ["network", "node", target_router, "acl", "remove_rule", position]
 
@@ -923,7 +925,7 @@ class HostNICAbstractAction(AbstractAction):
         self.shape: Dict[str, int] = {"node_id": num_nodes, "nic_id": max_nics_per_node}
         self.verb: str  # define but don't initialise: defends against children classes not defining this
 
-    def form_request(self, node_id: int, nic_id: int) -> List[str]:
+    def form_request(self, node_id: int, nic_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         node_name = self.manager.get_node_name_by_idx(node_idx=node_id)
         nic_num = self.manager.get_nic_num_by_idx(node_idx=node_id, nic_idx=nic_id)
@@ -960,7 +962,7 @@ class NetworkPortEnableAction(AbstractAction):
         super().__init__(manager=manager)
         self.shape: Dict[str, int] = {"port_id": max_nics_per_node}
 
-    def form_request(self, target_nodename: str, port_id: int) -> List[str]:
+    def form_request(self, target_nodename: str, port_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         if target_nodename is None or port_id is None:
             return ["do_nothing"]
@@ -979,7 +981,7 @@ class NetworkPortDisableAction(AbstractAction):
         super().__init__(manager=manager)
         self.shape: Dict[str, int] = {"port_id": max_nics_per_node}
 
-    def form_request(self, target_nodename: str, port_id: int) -> List[str]:
+    def form_request(self, target_nodename: str, port_id: int) -> RequestFormat:
         """Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
         if target_nodename is None or port_id is None:
             return ["do_nothing"]
@@ -1315,7 +1317,7 @@ class ActionManager:
         act_identifier, act_options = self.action_map[action]
         return act_identifier, act_options
 
-    def form_request(self, action_identifier: str, action_options: Dict) -> List[str]:
+    def form_request(self, action_identifier: str, action_options: Dict) -> RequestFormat:
         """Take action in CAOS format and use the execution definition to change it into PrimAITE request format."""
         act_obj = self.actions[action_identifier]
         return act_obj.form_request(**action_options)

src/primaite/game/agent/agent_log.py

@@ -0,0 +1,188 @@
+# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
+import logging
+from pathlib import Path
+
+from prettytable import MARKDOWN, PrettyTable
+
+from primaite.simulator import LogLevel, SIM_OUTPUT
+
+
+class _NotJSONFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        """
+        Determines if a log message does not start and end with '{' and '}' (i.e., it is not a JSON-like message).
+
+        :param record: LogRecord object containing all the information pertinent to the event being logged.
+        :return: True if log message is not JSON-like, False otherwise.
+        """
+        return not record.getMessage().startswith("{") and not record.getMessage().endswith("}")
+
+
+class AgentLog:
+    """
+    A Agent Log class is a simple logger dedicated to managing and writing logging updates and information for an agent.
+
+    Each log message is written to a file located at: <simulation output directory>/agent_name/agent_name.log
+    """
+
+    def __init__(self, agent_name: str):
+        """
+        Constructs a Agent Log instance for a given hostname.
+
+        :param hostname: The hostname associated with the system logs being recorded.
+        """
+        self.agent_name = agent_name
+        self.current_episode: int = 1
+        self.current_timestep: int = 0
+        self.setup_logger()
+
+    @property
+    def timestep(self) -> int:
+        """Returns the current timestep. Used for log indexing.
+
+        :return: The current timestep as an Int.
+        """
+        return self.current_timestep
+
+    def update_timestep(self, new_timestep: int):
+        """
+        Updates the self.current_timestep attribute with the given parameter.
+
+        This method is called within .step() to ensure that all instances of Agent Logs
+        are in sync with one another.
+
+        :param new_timestep: The new timestep.
+        """
+        self.current_timestep = new_timestep
+
+    def setup_logger(self):
+        """
+        Configures the logger for this Agent Log instance.
+
+        The logger is set to the DEBUG level, and is equipped with a handler that writes to a file and filters out
+        JSON-like messages.
+        """
+        if not SIM_OUTPUT.save_agent_logs:
+            return
+
+        log_path = self._get_log_path()
+        file_handler = logging.FileHandler(filename=log_path)
+        file_handler.setLevel(logging.DEBUG)
+
+        log_format = "%(timestep)s::%(levelname)s::%(message)s"
+        file_handler.setFormatter(logging.Formatter(log_format))
+
+        self.logger = logging.getLogger(f"{self.agent_name}_log")
+        for handler in self.logger.handlers:
+            self.logger.removeHandler(handler)
+        self.logger.setLevel(logging.DEBUG)
+        self.logger.addHandler(file_handler)
+
+    def _get_log_path(self) -> Path:
+        """
+        Constructs the path for the log file based on the agent name.
+
+        :return: Path object representing the location of the log file.
+        """
+        root = SIM_OUTPUT.agent_behaviour_path / f"episode_{self.current_episode}" / self.agent_name
+        root.mkdir(exist_ok=True, parents=True)
+        return root / f"{self.agent_name}.log"
+
+    def _write_to_terminal(self, msg: str, level: str, to_terminal: bool = False):
+        if to_terminal or SIM_OUTPUT.write_agent_log_to_terminal:
+            print(f"{self.agent_name}: ({ self.timestep}) ({level}) {msg}")
+
+    def debug(self, msg: str, to_terminal: bool = False):
+        """
+        Logs a message with the DEBUG level.
+
+        :param msg: The message to be logged.
+        :param to_terminal: If True, prints to the terminal too.
+        """
+        if SIM_OUTPUT.agent_log_level > LogLevel.DEBUG:
+            return
+
+        if SIM_OUTPUT.save_agent_logs:
+            self.logger.debug(msg, extra={"timestep": self.timestep})
+        self._write_to_terminal(msg, "DEBUG", to_terminal)
+
+    def info(self, msg: str, to_terminal: bool = False):
+        """
+        Logs a message with the INFO level.
+
+        :param msg: The message to be logged.
+        :param timestep: The current timestep.
+        :param to_terminal: If True, prints to the terminal too.
+        """
+        if SIM_OUTPUT.agent_log_level > LogLevel.INFO:
+            return
+
+        if SIM_OUTPUT.save_agent_logs:
+            self.logger.info(msg, extra={"timestep": self.timestep})
+        self._write_to_terminal(msg, "INFO", to_terminal)
+
+    def warning(self, msg: str, to_terminal: bool = False):
+        """
+        Logs a message with the WARNING level.
+
+        :param msg: The message to be logged.
+        :param timestep: The current timestep.
+        :param to_terminal: If True, prints to the terminal too.
+        """
+        if SIM_OUTPUT.agent_log_level > LogLevel.WARNING:
+            return
+
+        if SIM_OUTPUT.save_agent_logs:
+            self.logger.warning(msg, extra={"timestep": self.timestep})
+        self._write_to_terminal(msg, "WARNING", to_terminal)
+
+    def error(self, msg: str, to_terminal: bool = False):
+        """
+        Logs a message with the ERROR level.
+
+        :param msg: The message to be logged.
+        :param timestep: The current timestep.
+        :param to_terminal: If True, prints to the terminal too.
+        """
+        if SIM_OUTPUT.agent_log_level > LogLevel.ERROR:
+            return
+
+        if SIM_OUTPUT.save_agent_logs:
+            self.logger.error(msg, extra={"timestep": self.timestep})
+        self._write_to_terminal(msg, "ERROR", to_terminal)
+
+    def critical(self, msg: str, to_terminal: bool = False):
+        """
+        Logs a message with the CRITICAL level.
+
+        :param msg: The message to be logged.
+        :param timestep: The current timestep.
+        :param to_terminal: If True, prints to the terminal too.
+        """
+        if LogLevel.CRITICAL < SIM_OUTPUT.agent_log_level:
+            return
+
+        if SIM_OUTPUT.save_agent_logs:
+            self.logger.critical(msg, extra={"timestep": self.timestep})
+        self._write_to_terminal(msg, "CRITICAL", to_terminal)
+
+    def show(self, last_n: int = 10, markdown: bool = False):
+        """
+        Print an Agents Log as a table.
+
+        Generate and print PrettyTable instance that shows the agents behaviour log, with columns Time step,
+        Level and Message.
+
+        :param markdown: Use Markdown style in table output. Defaults to False.
+        """
+        table = PrettyTable(["Time Step", "Level", "Message"])
+        if markdown:
+            table.set_style(MARKDOWN)
+        table.align = "l"
+        table.title = f"{self.agent_name} Behaviour Log"
+        if self._get_log_path().exists():
+            with open(self._get_log_path()) as file:
+                lines = file.readlines()
+            for line in lines[-last_n:]:
+                table.add_row(line.strip().split("::"))
+        print(table)

src/primaite/game/agent/interface.py

@@ -7,6 +7,7 @@ from gymnasium.core import ActType, ObsType
 from pydantic import BaseModel, model_validator
 
 from primaite.game.agent.actions import ActionManager
+from primaite.game.agent.agent_log import AgentLog
 from primaite.game.agent.observations.observation_manager import ObservationManager
 from primaite.game.agent.rewards import RewardFunction
 from primaite.interface.request import RequestFormat, RequestResponse
@@ -69,6 +70,8 @@ class AgentSettings(BaseModel):
     "Configuration for when an agent begins performing it's actions"
     flatten_obs: bool = True
     "Whether to flatten the observation space before passing it to the agent. True by default."
+    action_masking: bool = False
+    "Whether to return action masks at each step."
 
     @classmethod
     def from_config(cls, config: Optional[Dict]) -> "AgentSettings":
@@ -116,6 +119,7 @@ class AbstractAgent(ABC):
         self.reward_function: Optional[RewardFunction] = reward_function
         self.agent_settings = agent_settings or AgentSettings()
         self.history: List[AgentHistoryItem] = []
+        self.logger = AgentLog(agent_name)
 
     def update_observation(self, state: Dict) -> ObsType:
         """
@@ -205,6 +209,7 @@ class ProxyAgent(AbstractAgent):
         )
         self.most_recent_action: ActType
         self.flatten_obs: bool = agent_settings.flatten_obs if agent_settings else False
+        self.action_masking: bool = agent_settings.action_masking if agent_settings else False
 
     def get_action(self, obs: ObsType, timestep: int = 0) -> Tuple[str, Dict]:
         """

src/primaite/game/agent/rewards.py

@@ -8,6 +8,7 @@ weighed sum of the components.
 
 The reward function is typically specified using a config yaml file or a config dictionary. The following example shows
 the structure:
+
 ```yaml
     reward_function:
         reward_components:

src/primaite/game/agent/scripted_agents/data_manipulation_bot.py

@@ -38,10 +38,11 @@ class DataManipulationAgent(AbstractScriptedAgent):
         :rtype: Tuple[str, Dict]
         """
         if timestep < self.next_execution_timestep:
+            self.logger.debug(msg="Performing do NOTHING")
             return "DONOTHING", {}
 
         self._set_next_execution_timestep(timestep + self.agent_settings.start_settings.frequency)
-
+        self.logger.info(msg="Performing a data manipulation attack!")
         return "NODE_APPLICATION_EXECUTE", {"node_id": self.starting_node_idx, "application_id": 0}
 
     def setup_agent(self) -> None:
@@ -54,3 +55,4 @@ class DataManipulationAgent(AbstractScriptedAgent):
         # we are assuming that every node in the node manager has a data manipulation application at idx 0
         num_nodes = len(self.action_manager.node_names)
         self.starting_node_idx = random.randint(0, num_nodes - 1)
+        self.logger.debug(msg=f"Select Start Node ID: {self.starting_node_idx}")

src/primaite/game/agent/scripted_agents/probabilistic_agent.py

@@ -85,4 +85,5 @@ class ProbabilisticAgent(AbstractScriptedAgent):
         :rtype: Tuple[str, Dict]
         """
         choice = self.rng.choice(len(self.action_manager.action_map), p=self.probabilities)
+        self.logger.info(f"Performing Action: {choice}")
         return self.action_manager.get_action(choice)

src/primaite/game/game.py

@@ -3,6 +3,7 @@
 from ipaddress import IPv4Address
 from typing import Dict, List, Optional
 
+import numpy as np
 from pydantic import BaseModel, ConfigDict
 
 from primaite import DEFAULT_BANDWIDTH, getLogger
@@ -15,6 +16,8 @@ from primaite.game.agent.scripted_agents.probabilistic_agent import Probabilisti
 from primaite.game.agent.scripted_agents.random_agent import PeriodicAgent
 from primaite.game.agent.scripted_agents.tap001 import TAP001
 from primaite.game.science import graph_has_cycle, topological_sort
+from primaite.simulator import SIM_OUTPUT
+from primaite.simulator.network.airspace import AirSpaceFrequency
 from primaite.simulator.network.hardware.base import NodeOperatingState
 from primaite.simulator.network.hardware.nodes.host.computer import Computer
 from primaite.simulator.network.hardware.nodes.host.host_node import NIC
@@ -166,6 +169,8 @@ class PrimaiteGame:
         for _, agent in self.agents.items():
             obs = agent.observation_manager.current_observation
             action_choice, parameters = agent.get_action(obs, timestep=self.step_counter)
+            if SIM_OUTPUT.save_agent_logs:
+                agent.logger.debug(f"Chosen Action: {action_choice}")
             request = agent.format_request(action_choice, parameters)
             response = self.simulation.apply_request(request)
             agent.process_action_response(
@@ -184,8 +189,14 @@ class PrimaiteGame:
         """Advance timestep."""
         self.step_counter += 1
         _LOGGER.debug(f"Advancing timestep to {self.step_counter} ")
+        self.update_agent_loggers()
         self.simulation.apply_timestep(self.step_counter)
 
+    def update_agent_loggers(self) -> None:
+        """Updates Agent Loggers with new timestep."""
+        for agent in self.agents.values():
+            agent.logger.update_timestep(self.step_counter)
+
     def calculate_truncated(self) -> bool:
         """Calculate whether the episode is truncated."""
         current_step = self.step_counter
@@ -194,6 +205,23 @@ class PrimaiteGame:
             return True
         return False
 
+    def action_mask(self, agent_name: str) -> np.ndarray:
+        """
+        Return the action mask for the agent.
+
+        This is a boolean list corresponding to the agent's action space. A False entry means this action cannot be
+        performed during this step.
+
+        :return: Action mask
+        :rtype: List[bool]
+        """
+        agent = self.agents[agent_name]
+        mask = [True] * len(agent.action_manager.action_map)
+        for i, action in agent.action_manager.action_map.items():
+            request = agent.action_manager.form_request(action_identifier=action[0], action_options=action[1])
+            mask[i] = self.simulation._request_manager.check_valid(request, {})
+        return np.asarray(mask, dtype=np.int8)
+
     def close(self) -> None:
         """Close the game, this will close the simulation."""
         return NotImplemented
@@ -229,6 +257,12 @@ class PrimaiteGame:
 
         simulation_config = cfg.get("simulation", {})
         network_config = simulation_config.get("network", {})
+        airspace_cfg = network_config.get("airspace", {})
+        frequency_max_capacity_mbps_cfg = airspace_cfg.get("frequency_max_capacity_mbps", {})
+
+        frequency_max_capacity_mbps_cfg = {AirSpaceFrequency[k]: v for k, v in frequency_max_capacity_mbps_cfg.items()}
+
+        net.airspace.frequency_max_capacity_mbps_ = frequency_max_capacity_mbps_cfg
 
         nodes_cfg = network_config.get("nodes", [])
         links_cfg = network_config.get("links", [])