csle_common.dao.system_identification package

Submodules

csle_common.dao.system_identification.empirical_conditional module

class csle_common.dao.system_identification.empirical_conditional.EmpiricalConditional(conditional_name: str, metric_name: str, sample_space: List[int], probabilities: List[float])[source]

Bases: csle_base.json_serializable.JSONSerializable

A DTO representing an empirical conditional distribution

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.empirical_conditional.EmpiricalConditional[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.empirical_conditional.EmpiricalConditional[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.empirical_system_model module

class csle_common.dao.system_identification.empirical_system_model.EmpiricalSystemModel(emulation_env_name: str, emulation_statistic_id: int, conditional_metric_distributions: List[List[csle_common.dao.system_identification.empirical_conditional.EmpiricalConditional]], descr: str)[source]

Bases: csle_common.dao.system_identification.system_model.SystemModel

A system model (list of conditional distributions) made up of empirical distributions

compute_kl_divergences() → None[source]

Computes the KL-divergences betwen different conditional distributions

Returns

None

copy() → csle_common.dao.system_identification.empirical_system_model.EmpiricalSystemModel[source]
Returns

a copy of the DTO

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.empirical_system_model.EmpiricalSystemModel[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.empirical_system_model.EmpiricalSystemModel[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.emulation_statistics module

class csle_common.dao.system_identification.emulation_statistics.EmulationStatistics(emulation_name: str, descr: str = '')[source]

Bases: csle_base.json_serializable.JSONSerializable

DTO representing delta-statistics measured from teh emulation

compute_descriptive_statistics_and_distributions() → None[source]

Computes descriptive statistics and empirical probability distributions based on the counters.

Returns

None

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.emulation_statistics.EmulationStatistics[source]

Converts a dict representation of the object to a DTO

Parameters

d – the dict to convert

Returns

the created instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.emulation_statistics.EmulationStatistics[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

get_number_of_samples() → int[source]

Counts the number of samples

Returns

the number of samples

static initialize_counters(d: Dict[str, Dict[int, int]], agg_labels: List[str]) → Dict[str, Dict[int, int]][source]

Initializes counters for a given dict

Parameters

  • d – the dict to initialize

  • agg_labels – the labels

Returns

the updated dict

static initialize_machine_counters(d: Dict[str, Dict[int, int]], s: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState, labels: List[str]) → Dict[str, Dict[int, int]][source]

Initializes counters for a given dict

Parameters

  • d – the dict to initialize

  • s – the state with the list of machines

  • labels – the labels to initialize

Returns

the initialized dict

initialize_machines(s: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState) → None[source]

Initializes counters for a given dict

Parameters

  • d – the dict to initialize

  • s – the state with the list of machines

Returns

the initialized dict

merge(second_statistic: csle_common.dao.system_identification.emulation_statistics.EmulationStatistics) → None[source]

Merges the statistic with another statistic by adding the counts

Parameters

second_statistic – the statistic to merge with

Returns

None

to_dict() → Dict[str, Any][source]

Converts the object to a dict representation

Returns

a dict representation of the object

update_counters(d: Dict[str, Any], s: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState, s_prime: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState) → None[source]

Updates the delta counters for a specific dict based on a state transition s->s’

Parameters

  • d – the dict to update

  • s – the current state

  • s_prime – the new state

Returns

None

update_delta_statistics(s: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState, s_prime: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState, a1: csle_common.dao.emulation_action.defender.emulation_defender_action.EmulationDefenderAction, a2: csle_common.dao.emulation_action.attacker.emulation_attacker_action.EmulationAttackerAction) → None[source]

Updates the emulation statistics (delta counters) with a given transition (s, a1, a2) -> (s’)

Parameters

  • s – the previous state

  • s_prime – the new state

  • a1 – the defender action

  • a2 – the attacker action

Returns

None

update_initial_statistics(s: csle_common.dao.emulation_config.emulation_env_state.EmulationEnvState) → None[source]

Updates the emulation statistics for the initial state

Parameters

s – the initial state

Returns

None

csle_common.dao.system_identification.gaussian_mixture_conditional module

class csle_common.dao.system_identification.gaussian_mixture_conditional.GaussianMixtureConditional(conditional_name: str, metric_name: str, num_mixture_components: int, dim: int, mixtures_means: List[List[float]], mixtures_covariance_matrix: List[List[List[float]]], mixture_weights: List[float], sample_space: List[int])[source]

Bases: csle_base.json_serializable.JSONSerializable

A DTO representing a Gaussian Mixture Conditional Distribution

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.gaussian_mixture_conditional.GaussianMixtureConditional[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.gaussian_mixture_conditional.GaussianMixtureConditional[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

static from_sklearn_gaussian_mixture(gmm: sklearn.mixture._gaussian_mixture.GaussianMixture, conditional_name: str, metric_name: str, num_components: int, sample_space: List[int], dim: int = 1) → csle_common.dao.system_identification.gaussian_mixture_conditional.GaussianMixtureConditional[source]

Creates the DTO from a Gaussian mixture fitted with sklearn

Parameters

  • gmm – the sklearn model

  • conditional_name – the name of the conditional

  • metric_name – the metric name

  • num_components – the number of components of the mixture

  • dim – the dimension of the mixture

  • sample_space – the sample space

Returns

a GaussianMixtureConditional instance

generate_distributions() → None[source]

Generates the combined mixture distribution by taking a convex combination of the Gaussians.

Returns

None

generate_distributions_for_samples(samples: List[float], normalize: bool = False)[source]

Generates a distribution over a given list of samples

Parameters

  • samples – the samples to generate the distribution for

  • normalize – a boolean flag indicating whether the density distribution should be normalized or not

Returns

the mixture distribution over the given samples.

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.gaussian_mixture_system_model module

class csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel(emulation_env_name: str, emulation_statistic_id: int, conditional_metric_distributions: List[List[csle_common.dao.system_identification.gaussian_mixture_conditional.GaussianMixtureConditional]], descr: str)[source]

Bases: csle_common.dao.system_identification.system_model.SystemModel

A system model (list of conditional distributions) made up of Gaussian Mixtures

compute_kl_divergences() → None[source]

Computes the KL-divergences betwen different conditional distributions

Returns

None

copy() → csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel[source]
Returns

a copy of the DTO

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.gp_conditional module

class csle_common.dao.system_identification.gp_conditional.GPConditional(conditional_name: str, metric_name: str, sample_space: List[int], observed_x: List[Union[int, float]], observed_y: List[Union[int, float]], scale_parameter: float, noise_parameter: float)[source]

Bases: csle_base.json_serializable.JSONSerializable

A DTO representing a Gaussian process conditional distribution

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.gp_conditional.GPConditional[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.gp_conditional.GPConditional[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

generate_distribution() → None[source]

Updates the distribution

Returns

None

generate_distributions_for_samples(samples) → Any[source]

Generates distributions for a given sample

Parameters

samples – the sample

Returns

the created distributions

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.gp_regression_model_with_gauissan_noise module

class csle_common.dao.system_identification.gp_regression_model_with_gauissan_noise.GPRegressionModelWithGaussianNoise(train_x, train_y, likelihood)[source]

Bases: gpytorch.models.exact_gp.ExactGP

DTO representing a GPRegression model with Gaussian Noise

forward(x) → gpytorch.distributions.multivariate_normal.MultivariateNormal[source]

Compute the prior latent distribution on a given input using the mean and covariance functions

Parameters

x – the input

Returns

the prior

training: bool

csle_common.dao.system_identification.gp_system_model module

class csle_common.dao.system_identification.gp_system_model.GPSystemModel(emulation_env_name: str, emulation_statistic_id: int, conditional_metric_distributions: List[List[csle_common.dao.system_identification.gp_conditional.GPConditional]], descr: str)[source]

Bases: csle_common.dao.system_identification.system_model.SystemModel

A system model (list of conditional distributions) made up of Gaussian Processes

compute_kl_divergences() → None[source]

Computes the KL-divergences betwen different conditional distributions

Returns

None

copy() → csle_common.dao.system_identification.gp_system_model.GPSystemModel[source]
Returns

a copy of the DTO

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.gp_system_model.GPSystemModel[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.gp_system_model.GPSystemModel[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

to_json_file(json_file_path: str) → None[source]

Saves the DTO to a json file

Parameters

json_file_path – the json file path to save the DTO to

Returns

None

to_json_str() → str[source]

Converts the DTO into a json string

Returns

the json string representation of the DTO

csle_common.dao.system_identification.mcmc_posterior module

class csle_common.dao.system_identification.mcmc_posterior.MCMCPosterior(posterior_name: str, samples: List[float], densities: List[float], sample_space: List[float])[source]

Bases: csle_base.json_serializable.JSONSerializable

A DTO representing a posterior obtained through MCMC

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.mcmc_posterior.MCMCPosterior[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.mcmc_posterior.MCMCPosterior[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.mcmc_system_model module

class csle_common.dao.system_identification.mcmc_system_model.MCMCSystemModel(emulation_env_name: str, emulation_statistic_id: int, posteriors: List[csle_common.dao.system_identification.mcmc_posterior.MCMCPosterior], descr: str)[source]

Bases: csle_common.dao.system_identification.system_model.SystemModel

A system model (list of posterior distributions) made up of posterior distributions

copy() → csle_common.dao.system_identification.mcmc_system_model.MCMCSystemModel[source]
Returns

a copy of the DTO

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.mcmc_system_model.MCMCSystemModel[source]

Converts a dict representation of the DTO into an instance

Parameters

d – the dict to convert

Returns

the converted instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.mcmc_system_model.MCMCSystemModel[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]
Returns

a dict representation of the DTO

csle_common.dao.system_identification.system_identification_config module

class csle_common.dao.system_identification.system_identification_config.SystemIdentificationConfig(model_type: csle_common.dao.system_identification.system_model_type.SystemModelType, hparams: Dict[str, csle_common.dao.training.hparam.HParam], output_dir: str, title: str, log_every: int)[source]

Bases: csle_base.json_serializable.JSONSerializable

DTO representing the configuration of a system identification job

static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.system_identification_config.SystemIdentificationConfig[source]

Converts a dict representation to an instance

Parameters

d – the dict to convert

Returns

the created instance

static from_json_file(json_file_path: str) → csle_common.dao.system_identification.system_identification_config.SystemIdentificationConfig[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

to_dict() → Dict[str, Any][source]

Converts the object to a dict representation

Returns

a dict representation of the object

csle_common.dao.system_identification.system_model module

class csle_common.dao.system_identification.system_model.SystemModel(descr: str, model_type: csle_common.dao.system_identification.system_model_type.SystemModelType)[source]

Bases: csle_base.json_serializable.JSONSerializable

Abstract system model

abstract copy() → csle_common.dao.system_identification.system_model.SystemModel[source]
Returns

a copy of the object

abstract static from_dict(d: Dict[str, Any]) → csle_common.dao.system_identification.system_model.SystemModel[source]

Converts a dict representation of the object to an instance

abstract static from_json_file(json_file_path: str) → csle_common.dao.system_identification.system_model.SystemModel[source]

Reads a json file and converts it to a DTO

Parameters

json_file_path – the json file path

Returns

the converted DTO

abstract to_dict() → Dict[str, Any][source]

Converts the object to a dict representation

Returns

a dict representation of the object

csle_common.dao.system_identification.system_model_type module

class csle_common.dao.system_identification.system_model_type.SystemModelType(value)[source]

Bases: enum.IntEnum

Enum representing the different system model types in CSLE

EMPIRICAL_DISTRIBUTION = 1
GAUSSIAN_MIXTURE = 0
GAUSSIAN_PROCESS = 2
MCMC = 3

Module contents