csle_agents.agents.particle_swarm package

Submodules

csle_agents.agents.particle_swarm.particle_swarm_agent module

class csle_agents.agents.particle_swarm.particle_swarm_agent.ParticleSwarmAgent(simulation_env_config: SimulationEnvConfig, emulation_env_config: Union[None, EmulationEnvConfig], experiment_config: ExperimentConfig, env: Optional[BaseEnv] = None, training_job: Optional[TrainingJobConfig] = None, save_to_metastore: bool = True)[source]

Bases: BaseAgent

Particle Swarm Search Agent

static compute_avg_metrics(metrics: Dict[str, List[Union[float, int]]]) → Dict[str, Union[float, int]][source]

Computes the average metrics of a dict with aggregated metrics

Parameters

metrics – the dict with the aggregated metrics

Returns

the average metrics

eval_theta(policy: Union[MultiThresholdStoppingPolicy, LinearThresholdStoppingPolicy], max_steps: int = 200) → Dict[str, Union[float, int]][source]

Evaluates a given threshold policy by running monte-carlo simulations

Parameters

policy – the policy to evaluate

Returns

the average metrics of the evaluation

get_policy(theta: List[float], L: int) → Union[MultiThresholdStoppingPolicy, LinearThresholdStoppingPolicy][source]

Gets the policy of a given parameter vector

Parameters

  • theta – the parameter vector

  • L – the number of parameters

Returns

the policy

hparam_names() → List[str][source]
Returns

a list with the hyperparameter names

static initial_theta(L: int, S: int, b_lo: Union[int, float], b_up: Union[int, float]) → Tuple[ndarray[Any, dtype[Any]], ndarray[Any, dtype[Any]]][source]

Initializes particle positions (thetas) randomly

Parameters

  • L – the dimension of theta

  • S – the number of particles in the swarm

  • b_lo – lower boundary of randomization

  • b_up – upper boundary of randomization

Returns

the initialized theta vector

static initial_velocity(L: int, S: int, b_lo: Union[int, float], b_up: Union[int, float]) → ndarray[Any, dtype[Any]][source]

Initializes the voleicities amongst each particle in the swarm :param L: the dimension :param S: the number of particles in the swarm :param b_lo: lower boundary od randomization :param b_up: upper boundary of randomization

particle_swarm(exp_result: ExperimentResult, seed: int, random_seeds: List[int], training_job: TrainingJobConfig)[source]

Runs the particle swarm algorithm

Parameters

  • exp_result – the experiment result object to store the result

  • seed – the seed

  • training_job – the training job config

  • random_seeds – list of seeds

Returns

the updated experiment result and the trained policy

random_position(L: int, S, b_lo: float, b_up: float) → ndarray[Any, dtype[Any]][source]

Utility function to get a random position

Parameters

  • L – the number of parameters

  • S – number of points

  • b_lo – lower bound

  • b_up – upper bound

Returns

an array with the random coordinates

static round_vec(vec) → List[float][source]

Rounds a vector to 3 decimals

Parameters

vec – the vector to round

Returns

the rounded vector

train() → ExperimentExecution[source]

Performs the policy training for the given random seeds using particle swarm

Returns

the training metrics and the trained policies

static update_metrics(metrics: Dict[str, List[Union[float, int]]], info: Dict[str, Union[float, int]]) → Dict[str, List[Union[float, int]]][source]

Update a dict with aggregated metrics using new information from the environment

Parameters

  • metrics – the dict with the aggregated metrics

  • info – the new information

Returns

the updated dict

Module contents