eta_utility.eta_x.common.common module

eta_utility.eta_x.common.common.vectorize_environment(env: type[BaseEnv], config_run: ConfigOptRun, env_settings: EnvSettings, callback: Callable[[BaseEnv], None], verbose: int = 2, vectorizer: type[DummyVecEnv] = <class 'stable_baselines3.common.vec_env.dummy_vec_env.DummyVecEnv'>, n: int = 1, *, training: bool = False, monitor_wrapper: bool = False, norm_wrapper_obs: bool = False, norm_wrapper_reward: bool = False) → VecNormalize | VecEnv[source]

Vectorize the environment and automatically apply normalization wrappers if configured. If the environment is initialized as an interaction_env it will not have normalization wrappers and use the appropriate configuration automatically.

Parameters:

  • env – Environment class which will be instantiated and vectorized.

  • config_run – Configuration for a specific optimization run.

  • env_settings – Configuration settings dictionary for the environment which is being initialized.

  • callback – Callback to call with an environment instance.

  • verbose – Logging verbosity to use in the environment.

  • vectorizer – Vectorizer class to use for vectorizing the environments.

  • n – Number of vectorized environments to create.

  • training – Flag to identify whether the environment should be initialized for training or playing. If true, it will be initialized for training.

  • norm_wrapper_obs – Flag to determine whether observations from the environments should be normalized.

  • norm_wrapper_reward – Flag to determine whether rewards from the environments should be normalized.

Returns:

Vectorized environments, possibly also wrapped in a normalizer.

eta_utility.eta_x.common.common.initialize_model(algo: type[BaseAlgorithm], policy: type[BasePolicy], envs: VecEnv | VecNormalize, algo_settings: AlgoSettings, seed: int | None = None, *, tensorboard_log: bool = False, log_path: Path | None = None) → BaseAlgorithm[source]

Initialize a new model or algorithm.

Parameters:

  • algo – Algorithm to initialize.

  • policy – The policy that should be used by the algorithm.

  • envs – The environment which the algorithm operates on.

  • algo_settings – Additional settings for the algorithm.

  • seed – Random seed to be used by the algorithm.

  • tensorboard_log – Flag to enable logging to tensorboard.

  • log_path – Path for tensorboard log. Online required if logging is true

Returns:

Initialized model.

eta_utility.eta_x.common.common.load_model(algo: type[BaseAlgorithm], envs: VecEnv | VecNormalize, algo_settings: AlgoSettings, path_model: Path, *, tensorboard_log: bool = False, log_path: Path | None = None) → BaseAlgorithm[source]

Load an existing model.

Parameters:

  • algo – Algorithm type of the model to be loaded.

  • envs – The environment which the algorithm operates on.

  • algo_settings – Additional settings for the algorithm.

  • path_model – Path to load the model from.

  • tensorboard_log – Flag to enable logging to tensorboard.

  • log_path – Path for tensorboard log. Online required if logging is true

Returns:

Initialized model.

eta_utility.eta_x.common.common.log_to_file(config: ConfigOpt, config_run: ConfigOptRun) → None[source]

Log output in terminal to the run_info file.

Parameters:

  • config – Configuration to figure out the logging settings.

  • config_run – Configuration for this optimization run.

eta_utility.eta_x.common.common.log_run_info(config: ConfigOpt, config_run: ConfigOptRun) → None[source]

Save run configuration to the run_info file.

Parameters:

  • config – Configuration for the framework.

  • config_run – Configuration for this optimization run.

eta_utility.eta_x.common.common.deserialize_net_arch(net_arch: Sequence[Mapping[str, Any]], in_features: int, device: th.device | str = 'auto') → th.nn.Sequential[source]

Deserialize_net_arch can take a list of dictionaries describing a sequential torch network and deserialize it by instantiating the corresponding classes.

An example for a possible net_arch would be:

[{"layer": "Linear", "out_features": 60},
 {"activation_func": "Tanh"},
 {"layer": "Linear", "out_features": 60},
 {"activation_func": "Tanh"}]

One key of the dictionary should be either ‘layer’, ‘activation_func’ or ‘process’. If the ‘layer’ key is present, a layer from the torch.nn module is instantiated, if the ‘activation_func’ key is present, the value will be instantiated as an activation function from torch.nn. If the key ‘process’ is present, the value will be interpreted as a data processor from eta_utility.eta_x.common.processors.

All other keys of each dictionary will be used as keyword parameters to the instantiation of the layer, activation function or processor.

Only the number of input features for the first layer must be specified (using the ‘in_features’) parameter. The function will then automatically determine the number of input features for all other layers in the sequential network.

Parameters:

  • net_arch – List of dictionaries describing the network architecture.

  • in_features – Number of input features for the first layer.

  • device – Torch device to use for training the network.

Returns:

Sequential torch network.

eta_utility.eta_x.common.common.log_net_arch(model: BaseAlgorithm, config_run: ConfigOptRun) → None[source]

Store network architecture or policy information in a file. This requires for the model to be initialized, otherwise it will raise a ValueError.

Parameters:

  • model – The algorithm whose network architecture is stored.

  • config_run – Optimization run configuration (which contains info about the file to store info in).

Raises:

ValueError.

eta_utility.eta_x.common.common.is_vectorized_env(env: BaseEnv | VecEnv | VecNormalize | None) → bool[source]

Check if an environment is vectorized.

Parameters:

env – The environment to check.

eta_utility.eta_x.common.common.is_env_closed(env: BaseEnv | VecEnv | VecNormalize | None) → bool[source]

Check whether an environment has been closed.

Parameters:

env – The environment to check.

eta_utility.eta_x.common.common.episode_results_path(series_results_path: Path, run_name: str, episode: int, env_id: int = 1) → pathlib.Path[source]

Generate a filepath which can be used for storing episode results of a specific environment as a csv file.

Name is of the format: ThisRun_001_01.csv (run name _ episode number _ environment id .csv)

Parameters:

  • series_results_path – Path for results of the series of optimization runs.

  • run_name – Name of the optimization run.

  • episode – Number of the episode the environment is working on.

  • env_id – Identification of the environment.

eta_utility.eta_x.common.common.episode_name_string(run_name: str, episode: int, env_id: int = 1) → str[source]

Generate a name which can be used to pre or postfix files from a specific episode and run of an environment.

Name is of the format: ThisRun_001_01 (run name _ episode number _ environment id)

Parameters:

  • run_name – Name of the optimization run.

  • episode – Number of the episode the environment is working on.

  • env_id – Identification of the environment.