Source code for csle_tolerance.dao.intrusion_recovery_pomdp_config

from typing import List, Dict, Any
import numpy as np
import logging
import io
import json
from csle_common.dao.simulation_config.simulation_env_input_config import SimulationEnvInputConfig


[docs]class IntrusionRecoveryPomdpConfig(SimulationEnvInputConfig): """ DTO containing the configuration of an intrusion recovery POMDP """ def __init__(self, eta: float, p_a: float, p_c_1: float, p_c_2: float, p_u: float, BTR: int, negate_costs: bool, seed: int, discount_factor: float, states: List[int], actions: List[int], observations: List[int], cost_tensor: List[List[float]], observation_tensor: List[List[float]], transition_tensor: List[List[List[float]]], b1: List[float], T: int, simulation_env_name: str, gym_env_name: str, max_horizon: float = np.inf) -> None: """ Initializes the DTO :param eta: the scaling factor for the cost function :param p_a: the intrusion probability :param p_c_1: the crash probability in the healthy state :param p_c_2: the crash probability in the compromised state :param p_u: the software upgrade probability :param BTR: the periodic recovery interval :param negate_costs: boolean flag indicating whether costs should be negated or not :param seed: the random seed :param discount_factor: the discount factor :param states: the list of states :param actions: the list of actions :param observations: the list of observations :param cost_tensor: the cost tensor :param observation_tensor: the observation tensor :param transition_tensor: the transition tensor :param b1: the initial belief :param T: the time horizon :param simulation_env_name: name of the simulation environment :param gym_env_name: name of the gym environment :param max_horizon: the maximum horizon to avoid infinie simulations """ self.eta = eta self.p_a = p_a self.p_c_1 = p_c_1 self.p_c_2 = p_c_2 self.p_u = p_u self.BTR = BTR self.negate_costs = negate_costs self.seed = seed self.discount_factor = discount_factor self.states = states self.actions = actions self.observations = observations self.cost_tensor = cost_tensor self.observation_tensor = observation_tensor self.transition_tensor = transition_tensor self.b1 = b1 self.T = T self.simulation_env_name = simulation_env_name self.gym_env_name = gym_env_name self.max_horizon = max_horizon def __str__(self) -> str: """ :return: a string representation of the DTO """ return (f"eta: {self.eta}, p_a: {self.p_a}, p_c_1: {self.p_c_1}, p_c_2: {self.p_c_2}, p_u: {self.p_u}, " f"BTR: {self.BTR}, negate_costs: {self.negate_costs}, seed: {self.seed}, " f"discount_factor: {self.discount_factor}, states: {self.states}, actions: {self.actions}, " f"observations: {self.observation_tensor}, cost_tensor: {self.cost_tensor}, " f"observation_tensor: {self.observation_tensor}, transition_tensor: {self.transition_tensor}, " f"b1:{self.b1}, T: {self.T}, simulation_env_name: {self.simulation_env_name}, " f"gym_env_name: {self.gym_env_name}, max_horizon: {self.max_horizon}")
[docs] @staticmethod def from_dict(d: Dict[str, Any]) -> "IntrusionRecoveryPomdpConfig": """ Converts a dict representation to an instance :param d: the dict to convert :return: the created instance """ dto = IntrusionRecoveryPomdpConfig( eta=d["eta"], p_a=d["p_a"], p_c_1=d["p_c_1"], p_c_2=d["p_c_2"], p_u=d["p_u"], BTR=d["BTR"], negate_costs=d["negate_costs"], seed=d["seed"], discount_factor=d["discount_factor"], states=d["states"], actions=d["actions"], observations=d["observations"], cost_tensor=d["cost_tensor"], observation_tensor=d["observation_tensor"], transition_tensor=d["transition_tensor"], b1=d["b1"], T=d["T"], simulation_env_name=d["simulation_env_name"], gym_env_name=d["gym_env_name"]) return dto
[docs] def to_dict(self) -> Dict[str, Any]: """ Gets a dict representation of the object :return: A dict representation of the object """ d: Dict[str, Any] = {} d["eta"] = self.eta d["p_a"] = self.p_a d["p_c_1"] = self.p_c_1 d["p_c_2"] = self.p_c_2 d["p_u"] = self.p_u d["BTR"] = self.BTR d["negate_costs"] = self.negate_costs d["seed"] = self.seed d["discount_factor"] = self.discount_factor d["states"] = self.states d["actions"] = self.actions d["observations"] = self.observations d["cost_tensor"] = self.cost_tensor d["observation_tensor"] = self.observation_tensor d["transition_tensor"] = self.transition_tensor d["b1"] = self.b1 d["T"] = self.T d["simulation_env_name"] = self.simulation_env_name d["gym_env_name"] = self.gym_env_name return d
[docs] @staticmethod def from_json_file(json_file_path: str) -> "IntrusionRecoveryPomdpConfig": """ Reads a json file and converts it to a DTO :param json_file_path: the json file path :return: the converted DTO """ logging.info(msg="msg") with io.open(json_file_path, 'r') as f: json_str = f.read() print(json_str) return IntrusionRecoveryPomdpConfig.from_dict(json.loads(json_str))