culebra.trainer.aco.CPACO class

class CPACO(fitness_func: FitnessFunction, solution_cls: type[Ant], species: Species, initial_pheromone: float | Sequence[float, ...], heuristic: ndarray[float] | Sequence[ndarray[float], ...] | None = None, pheromone_influence: float | Sequence[float, ...] | None = None, heuristic_influence: float | Sequence[float, ...] | None = None, exploitation_prob: float | None = None, col_size: int | None = None, pop_size: int | None = None, custom_termination_func: Callable[[CPACO], bool] | None = None, max_num_iters: int | None = None, checkpoint_activation: bool | None = None, checkpoint_freq: int | None = None, checkpoint_basename: str | None = None, verbosity: bool | None = None, random_seed: int | None = None)

Bases: PACO

Create a new Crowding PACO trainer.

Parameters:
Raises:
  • TypeError – If any argument is not of the appropriate type

  • ValueError – If any argument has an incorrect value

Class methods

classmethod CPACO.load(filename: str) Base

Load a serialized object from a file.

Parameters:

filename (str) – The file name.

Returns:

The loaded object

Raises:

Properties

property CPACO.checkpoint_activation: bool

Checkpointing activation.

Returns:

True if checkpointing is active, or False otherwise

Return type:

bool

Setter:

Modify the checkpointing activation

Parameters:

value (bool) – New value for the checkpoint activation. If set to None, _default_checkpoint_activation is chosen

Raises:

TypeError – If value is not a boolean value

property CPACO.checkpoint_basename: str

Checkpoint base file path.

Return type:

str

Setter:

Modify the checkpoint base file path

Parameters:

value (str) – New value for the checkpoint base file path. If set to None, _default_checkpoint_basename is chosen

Raises:

TypeError – If value is not a valid file name

property CPACO.checkpoint_filename: str

Checkpoint file path.

Returns:

The file path to store checkpoints

Return type:

str

property CPACO.checkpoint_freq: int

Checkpoint frequency.

Return type:

int

Setter:

Modify the checkpoint frequency

Parameters:

value (int) – New value for the checkpoint frequency. If set to None, _default_checkpoint_freq is chosen

Raises:
property CPACO.choice_info: ndarray[float] | None

Choice information for all the graph’s arcs.

The choice information is generated from both the pheromone and the heuristic matrices, modified by other parameters (depending on the ACO approach) and is used to obtain the probalility of following the next feasible arc for the node.

Returns:

The choice information or None if the training process has not begun

Return type:

ndarray[float]

property CPACO.col: list[Ant] | None

Colony.

Returns:

The colony or None if it has not been generated yet

Return type:

list[Ant]

property CPACO.col_size: int

Colony size.

Return type:

int

Setter:

Set a new value for the colony size

Parameters:

size (int) – The new colony size. If set to None, _default_col_size is chosen

Raises:
property CPACO.container: DistributedTrainer | None

Container of this trainer.

The trainer container is only used by distributed trainers. For the rest of trainers defaults to None.

Return type:

DistributedTrainer

Setter:

Set a new value for container of this trainer

Parameters:

value (DistributedTrainer) – New value for the container or None

Raises:

TypeError – If value is not a valid trainer

property CPACO.cooperative_fitness_estimation_func: Callable[Sequence[Sequence[float]], Sequence[float]]

Cooperative fitness estimation function.

Funtion to estimate the cooperative fitness of a solution from several fitness trials with different cooperators. Only used by cooperative trainers.

Return type:

Callable

Setter:

Set a new function

Parameters:

func (Callable) – The new function

Raises:

TypeError – If func is not a valid function

property CPACO.cooperators: list[list[Solution | None]] | None

Cooperators of the other species.

Only used by cooperative trainers. If the trainer does not use cooperators, None is returned.

Return type:

list[list[Solution]]

property CPACO.current_iter: int | None

Current iteration.

Returns:

The current iteration or None if the training has not been done yet

Return type:

int

property CPACO.custom_termination_func: Callable[[CentralizedTrainer], bool] | None

Custom termination criterion.

Although the trainer will always stop when the max_num_iters are reached, a custom termination criterion can be set to detect convergente and stop the trainer earlier. This custom termination criterion must be a function which receives the trainer as its unique argument and returns a boolean value, True if the training should terminate or False otherwise.

If more than one arguments are needed to define the termination condition, functools.partial() can be used:

from functools import partial

def my_crit(trainer, max_iters):
    if trainer.current_iter < max_iters:
        return False
    return True

trainer.custom_termination_func = partial(my_crit, max_iters=10)
Setter:

Set a new custom termination criterion

Parameters:

func (Callable) – The new custom termination criterion. If set to None, the default termination criterion is used

Raises:

TypeError – If func is not callable

property CPACO.exploitation_prob: float

Exploitation probability (\({q_0}\)).

Return type:

float

Setter:

Set a new value for the exploitation probability

Parameters:

prob (float) – The new probability. If set to None, _default_exploitation_prob is chosen

Raises:
property CPACO.fitness_func: FitnessFunction

Training fitness function.

Return type:

FitnessFunction

Setter:

Set a new fitness function

Parameters:

value (FitnessFunction) – The new training fitness function

Raises:

TypeError – If value is not a valid fitness function

property CPACO.heuristic: tuple[ndarray[float], ...]

Heuristic matrices.

Return type:

tuple[ndarray[float]]

Setter:

Set new heuristic matrices

Parameters:

values (ndarray[float] | Sequence[ndarray[float], ...]) – The new heuristic matrices. Both a single matrix or a sequence of matrices are allowed. If a single matrix is provided, it will be replicated for all the heuristic matrices. If set to None, _default_heuristic is chosen

Raises:
property CPACO.heuristic_influence: tuple[float, ...]

Relative influence of heuristic (\({\beta}\)).

Returns:

One value for each heuristic matrix

Return type:

tuple[float]

Setter:

Set new values for the relative influence of each heuristic matrix

Parameters:

values (float | Sequence[float]) – New value for the relative influence of each heuristic matrix. Both a scalar value or a sequence of values are allowed. If a scalar value is provided, it will be used for all the heuristic matrices. If set to None, _default_heuristic_influence is chosen

Raises:
abstract property CPACO.heuristic_shapes: tuple[tuple[int, int], ...]

Shape of the heuristic matrices.

This property must be overridden by subclasses to return a correct value.

Return type:

tuple[tuple[int]]

Raises:

NotImplementedError – If has not been overridden

property CPACO.index: int

Trainer index.

The trainer index is only used within distributed trainers.

Return type:

int

Setter:

Set a new value for trainer index.

Parameters:

value (int) – New value for the trainer index. If set to None, _default_index is chosen

Raises:
property CPACO.initial_pheromone: tuple[float, ...]

Initial value for each pheromone matrix.

Returns:

One initial value for each pheromone matrix

Return type:

tuple[float]

Setter:

Set the initial value for each pheromone matrix

Parameters:

values (float | Sequence[float]) – New initial value for each pheromone matrix. Both a scalar value or a sequence of values are allowed. If a scalar value is provided, it will be used for all the pheromone matrices

Raises:
property CPACO.iteration_metric_names: tuple(str)

Names of the metrics recorded each iteration.

Return type:

tuple[str]

property CPACO.iteration_obj_stats: dict(str, Callable)

Stats applied to each objective every iteration.

Return type:

dict

property CPACO.logbook: Logbook | None

Trainer logbook.

Returns:

A logbook with the statistics of the training or None if the training has not been done yet

Return type:

Logbook

property CPACO.max_num_iters: int

Maximum number of iterations.

Return type:

int

Setter:

Set a new value for the maximum number of iterations

Parameters:

value (int) – The new maximum number of iterations. If set to None, _default_max_num_iters is chosen

Raises:
property CPACO.num_evals: int | None

Number of evaluations performed while training.

Returns:

The number of evaluations or None if the training has not been done yet

Return type:

int

property CPACO.num_iters: int | None

Number of iterations performed while training.

Returns:

The number of iterations or None if the training has not been done yet

Return type:

int

property CPACO.num_heuristic_matrices: int

Number of heuristic matrices used by this trainer.

Return type:

int

property CPACO.num_pheromone_matrices: int

Number of pheromone matrices used by this trainer.

Return type:

int

property CPACO.pheromone: list[ndarray[float], ...] | None

Pheromone matrices.

Returns:

The pheromone matrices or None if the training process has not begun

Return type:

list[ndarray[float]]

property CPACO.pheromone_influence: tuple[float, ...]

Relative influence of pheromone (\({\alpha}\)).

Returns:

One value for each pheromone matrix

Return type:

tuple[float]

Setter:

Set new values for the relative influence of each pheromone matrix

Parameters:

values (float | Sequence[float]) – New value for the relative influence of each pheromone matrix. Both a scalar value or a sequence of values are allowed. If a scalar value is provided, it will be used for all the pheromone matrices. If set to None, _default_pheromone_influence is chosen

Raises:
abstract property CPACO.pheromone_shapes: tuple[tuple[int, int], ...]

Shape of the pheromone matrices.

This property must be overridden by subclasses to return a correct value.

Return type:

tuple[tuple[int]]

Raises:

NotImplementedError – If has not been overridden

property CPACO.pop: list[Ant] | None

Population.

Returns:

The population or None if it has not been generated

Return type:

list[Ant]

property CPACO.pop_size: int

Population size.

Return type:

int

Setter:

Set the population size

Parameters:

size (int) – The new population size. If set to None, _default_pop_size is chosen

Raises:
property CPACO.random_seed: int

Random seed used by this trainer.

Return type:

int

Setter:

Set a new value for the random seed

Parameters:

value (int) – New value

property CPACO.receive_representatives_func: Callable[[CentralizedTrainer], None]

Representatives reception function.

Distributed trainers should set this property to their subtrainers in order to implement an adequate representatives exchange mechanism.

Return type:

Callable

Setter:

Set a new function

Parameters:

func (Callable) – The new function

Raises:

TypeError – If func is not a valid function

property CPACO.runtime: float | None

Training runtime.

Returns:

The training runtime or None if the training has not been done yet.

Return type:

float

property CPACO.send_representatives_func: Callable[[CentralizedTrainer], None]

Representatives sending function.

Distributed trainers should set this property to their subtrainers in order to implement an adequate representatives exchange mechanism.

Return type:

Callable

Setter:

Set a new function

Parameters:

func (Callable) – The new function

Raises:

TypeError – If func is not a valid function

property CPACO.solution_cls: type[Solution]

Solution class.

Return type:

type[Solution]

Setter:

Set the new solution class

Parameters:

value (type[Solution]) – The new class

Raises:

TypeError – If value is not a Solution subclass

property CPACO.species: Species

Species.

Return type:

Species

Setter:

Set the new species

Parameters:

value (Species) – The new species

Raises:

TypeError – If value is not a Species

property CPACO.state_proxy: DictProxy | None

Proxy for the state of this trainer.

The proxy is used to copy the state of the trainer only when training is executed within a multiprocess.Process. Defaults to None

Return type:

DictProxy

Setter:

Set a new value for the state proxy of this trainer

Parameters:

value (DictProxy) – New value for the state proxy or None

Raises:

TypeError – If value is not a valid proxy

property CPACO.training_finished: bool

Check if training has finished.

Returns:

:data`True` if training has finished

Return type:

bool

property CPACO.verbosity: bool

Verbosity of this trainer.

Return type:

bool

Setter:

Set a new value for the verbosity

Parameters:

value (bool) – The verbosity. If set to None, _default_verbosity is chosen

Raises:

TypeError – If value is not boolean

Private properties

property CPACO._default_checkpoint_activation: bool

Default checkpointing activation.

Returns:

DEFAULT_CHECKPOINT_ACTIVATION

Return type:

bool

property CPACO._default_checkpoint_basename: str

Default checkpointing base file name.

Returns:

DEFAULT_CHECKPOINT_BASENAME

Return type:

str

property CPACO._default_checkpoint_freq: int

Default checkpointing frequency.

Returns:

DEFAULT_CHECKPOINT_FREQ

Return type:

int

abstract property CPACO._default_col_size: int

Default colony size.

This property must be overridden by subclasses to return a correct default value.

Return type:

int

Raises:

NotImplementedError – If has not been overridden

property CPACO._default_cooperative_fitness_estimation_func: Callable[Sequence[Sequence[float]], Sequence[float]]

Default cooperative fitness estimation function.

Return the average of all fitness trials.

property CPACO._default_exploitation_prob: float

Default exploitation probability (\({q_0}\)).

Returns:

DEFAULT_EXPLOITATION_PROB

Return type:

float

abstract property CPACO._default_heuristic: tuple[ndarray[float], ...]

Default heuristic matrices.

This property must be overridden by subclasses to return a correct value.

Return type:

tuple[ndarray[float]]

Raises:

NotImplementedError – If has not been overridden

property CPACO._default_heuristic_influence: tuple[float, ...]

Default relative influence of heuristic (\({\beta}\)).

Returns:

The DEFAULT_HEURISTIC_INFLUENCE for each pheromone matrix

Return type:

tuple[float]

property CPACO._default_index: int

Default index.

Returns:

DEFAULT_INDEX

Return type:

int

property CPACO._default_max_num_iters: int

Default maximum number of iterations.

Returns:

DEFAULT_MAX_NUM_ITERS

Return type:

int

property CPACO._default_pheromone_influence: tuple[float, ...]

Default relative influence of pheromone (\({\alpha}\)).

Returns:

The DEFAULT_PHEROMONE_INFLUENCE for each pheromone matrix

Return type:

tuple[float]

property CPACO._default_pop_size: int

Default Population size.

Returns:

col_size

Return type:

int

property CPACO._default_receive_representatives_func: Callable[[CentralizedTrainer], None]

Default implementation for the representatives reception function.

It does nothing.

property CPACO._default_send_representatives_func: Callable[[CentralizedTrainer], None]

Default implementation for the representatives sending function.

It does nothing.

property CPACO._default_verbosity: bool

Default verbosity.

Returns:

DEFAULT_VERBOSITY

Return type:

bool

Methods

CPACO.best_cooperators() list[list[Solution | None]] | None

Return a list of cooperators from each species.

Only used for cooperative trainers.

Returns:

A list of cooperators lists if the trainer is cooperative or None in other cases

Return type:

list[list[Solution]]

CPACO.best_solutions() tuple[HallOfFame]

Get the best solutions found for each species.

Returns:

One Hall of Fame for each species

Return type:

tuple[HallOfFame]

CPACO.dump(filename: str) None

Serialize this object and save it to a file.

Parameters:

filename (str) – The file name.

Raises:
CPACO.evaluate(sol: Solution, fitness_func: FitnessFunction | None = None, index: int | None = None, cooperators: Sequence[Sequence[Solution | None]] | None = None) int

Evaluate one solution.

Its fitness will be modified according with the fitness function results.

Parameters:
  • sol (Solution) – The solution

  • fitness_func (FitnessFunction) – The fitness function. If omitted, the training function is used

  • index (int) – Index where sol should be inserted in the cooperators sequence to form a complete solution for the problem

  • cooperators (Sequence[Sequence[Solution]]) – Sequence of cooperators of other species or None (if no cooperators are needed to evaluate sol)

Returns:

The number of evaluations performed

Return type:

int

CPACO.integrate_representatives(representatives: list[Ant]) None

Integrate representative solutions.

This method is intended to be called within distributed trainers to make the implementation of migrations easier.

Parameters:

representatives (list[Ant]) – A list of solutions

CPACO.reset() None

Reset the trainer.

Delete the state of the trainer (with _reset_state()) and also all the internal data structures needed to perform the training (with _reset_internals()).

This method should be invoqued each time a hyper parameter is modified.

CPACO.select_representatives() list[Ant]

Select representative solutions.

This method is intended to be called within distributed trainers to make the implementation of migrations easier.

Returns:

A list of solutions

Return type:

list[Ant]

CPACO.test(best_found: Sequence[HallOfFame], fitness_func: FitnessFunction, cooperators: Sequence[Sequence[Solution]] | None = None) None

Apply the test fitness function to the solutions found.

Update the solutions in best_found with their test fitness.

Parameters:
Raises:
  • TypeError – If any parameter has a wrong type

  • ValueError – If any parameter has an invalid value.

CPACO.train()

Perform the training process.

Private methods

abstract CPACO._ant_choice_info(ant: Ant) ndarray[float]

Return the choice info to obtain the next node the ant will visit.

Parameters:

ant (Ant) – The ant

Return type:

ndarray[float]

This method should be overridden by subclasses.

Raises:

NotImplementedError – If has not been overridden

CPACO._calculate_choice_info() None

Calculate the choice info matrix.

The choice info matrix is re-calculated every time an ant is generated since in CPACO each ant uses its own heuristic influence correction factors.

CPACO._default_termination_func() bool

Default termination criterion.

Returns:

True if max_num_iters iterations have been run

Return type:

bool

CPACO._deposit_pheromone(ants: Sequence[Ant], weight: float = 1.0) None

Make some ants deposit weighted pheromone.

This method must be overridden by subclasses to take into account the correct number and shape of the pheromone matrices.

Parameters:
Raises:

NotImplementedError – If has not been overridden

CPACO._do_iteration() None

Implement an iteration of the training process.

CPACO._do_training() None

Apply the training algorithm.

Execute the trainer until the termination condition is met. Each iteration is composed by the following steps:

CPACO._finish_iteration() None

Finish an iteration.

Finish the iteration metrics (number of evaluations, execution time) after each iteration is run.

CPACO._finish_training() None

Finish the training process.

This method is called after the training has finished. It can be overridden to perform any treatment of the solutions found.

CPACO._generate_ant() Ant

Generate a new ant.

A new set of heuristic influence correction factors is generated for the ant. Then, the choice info for the ant is generated according to these correction factors. Finally, the ant makes its path and gets evaluated.

Returns:

The new ant

Return type:

Ant

CPACO._generate_col() None

Fill the colony with evaluated ants.

CPACO._generate_cooperators() Sequence[Sequence[Solution | None]] | None

Generate cooperators from other species.

Returns:

The cooperators

Return type:

Sequence[ Sequence[Solution | None]] | None

CPACO._get_iteration_metrics() dict

Collect the iteration metrics.

Returns:

The metrics

Return type:

dict

CPACO._get_objective_stats() dict

Gather the objective stats.

Returns:

The stats

Return type:

dict

CPACO._get_state() dict[str, Any]

Return the state of this trainer.

Overridden to add the current population to the trainer’s state.

Return type:

dict

CPACO._init_internals() None

Set up the trainer internal data structures to start training.

Create all the internal objects, functions and data structures needed to run the training process. For the CPACO class, the heuristic influence correction factors are created. Subclasses which need more objects or data structures should override this method.

CPACO._init_pheromone() None

Init the pheromone matrix(ces) according to the initial value(s).

CPACO._init_state() None

Init the trainer state.

If there is any checkpoint file, the state is initialized from it with the _load_state() method. Otherwise a new initial state is generated with the _new_state() method.

CPACO._init_training() None

Init the training process.

Initialize the state of the trainer and all the internal data structures needed to perform the training.

CPACO._load_state() None

Load the state of the last checkpoint.

Raises:

Exception – If the checkpoint file can’t be loaded

CPACO._new_state() None

Generate a new trainer state.

Overridden to initialize the population with random ants.

CPACO._next_choice(ant: Ant) int | None

Choose the next node for an ant.

The election is made from the feasible neighborhood of the current node, which is composed of those nodes neither discarded nor visited yet by the ant and connected to its current node. Any node forbidden by the constraints of the :class:~culebra.abc.Species defining the problem is also unfeasible.

The best possible node is selected with probability exploitation_prob. In case the best node is not chosen, the next node is selected probabilistically according to the choice_info matrix.

Parameters:

ant (Ant) – The ant

Returns:

The index of the chosen node or None if there isn’t any feasible node

Return type:

int

CPACO._pheromone_amount(ant: Ant) tuple[float, ...]

Return the amount of pheromone to be deposited by an ant.

Each ant deposit an amount of pheromone according to its rank + 1.

Parameters:

ant (Ant) – The ant

Returns:

The amount of pheromone to be deposited for each objective

Return type:

tuple[float]

CPACO._reset_internals() None

Reset the internal structures of the trainer.

Overridden to reset the pheromone matrices. If subclasses overwrite the _init_internals() method to add any new internal object, this method should also be overridden to reset all the internal objects of the trainer.

CPACO._reset_state() None

Reset the trainer state.

Overridden to reset the population.

CPACO._save_state() None

Save the state at a new checkpoint.

Raises:

Exception – If the checkpoint file can’t be written

CPACO._set_state(state: dict[str, Any]) None

Set the state of this trainer.

Overridden to add the current population to the trainer’s state.

Parameters:

state (dict) – The last loaded state

CPACO._start_iteration() None

Start an iteration.

Prepare the iteration metrics (number of evaluations, execution time) before each iteration is run and create an empty ant colony. Overridden to avoid the calculation of the choice information before executing the next iteration because CPACO applies a different choice probability for each ant.

CPACO._termination_criterion() bool

Control the training termination.

Returns:

True if either the default termination criterion or a custom termination criterion is met. The default termination criterion is implemented by the _default_termination_func() method. Another custom termination criterion can be set with custom_termination_func method

Return type:

bool

CPACO._unfeasible_nodes(ant: Ant) ndarray[float]

Return the indices of all unfeasible nodes.

Subclasses should also filter out any node forbidden by the constraints of the :class:~culebra.abc.Species defining the problem.

Parameters:

ant (Ant) – The ant

Returns:

All nodes visited by the ant, whether selected or discarded

Return type:

ndarray[float]

CPACO._update_logbook() None

Append the iteration dato to the logbook.

If verbosity is activated, also output the log data to the console.

CPACO._update_pheromone() None

Update the pheromone trails.

Pheromone trails are reinitialized and updated according to the current population.

CPACO._update_pop() None

Update the population.