Utils

Utility functions and auxiliary data structures. The majority of the utility functions are compiled with Numba - Jit, for improved performance.

Submodules

star_pso.utils.auxiliary module

class star_pso.utils.auxiliary.BlockType(value)[source]

Bases: Enum

Description:

BlockType enumeration defines the types that a data block can take.

BINARY = 2
CATEGORICAL = 3
FLOAT = 0
INTEGER = 1
class star_pso.utils.auxiliary.SpecialMode(value)[source]

Bases: Enum

Description:

SpecialMode enumeration defines specific modes that the GenericPSO can accommodate. These are handled internally (i.e. NOT by the end user) to call the evaluate_function methods and perform operations directly related to the specific versions of PSO method.

CATEGORICAL = 1
JACK_OF_ALL_TRADES = 2
NORMAL = 0
star_pso.utils.auxiliary.check_velocity_parameters(options: dict) → None[source]

Checks that the options dictionary has all the additional parameters to estimate the velocities of the optimization algorithm:

  • ‘w0’: inertia weight

  • ‘c1’: cognitive coefficient

  • ‘c2’: social coefficient

  • ‘mode’: mode of operation

Parameters:

options – dictionary to check for missing parameters.

Returns:

None.

star_pso.utils.auxiliary.cost_function(func: Callable | None = None, minimize: bool = False)[source]

Decorator for the function we want to optimize. The default setting is maximization.

Parameters:

  • func – the function to be optimized.

  • minimize – if True it will return the negative function value to allow for the minimization. Default is False.

Returns:

the ‘function_wrapper’ method.

star_pso.utils.auxiliary.identify_global_optima(swarm_population: list[Particle], epsilon: float = 1e-05, radius: float = 0.1, f_opt: float | None = None) → list[source]

This auxiliary method will search if the global optimal solution(s) are found in the swarm population.

Parameters:

  • swarm_population – a list[Particle] of potential solutions.

  • epsilon – accuracy level of the global optimal solution.

  • radius – niche radius of the distance between two particles.

  • f_opt – function value for the global optimal solution.

Returns:

a list of best-fit individuals identified as solutions.

star_pso.utils.auxiliary.kl_divergence_array(p: ndarray[Any, dtype[_ScalarType_co]], q: ndarray[Any, dtype[_ScalarType_co]]) → ndarray[Any, dtype[_ScalarType_co]][source]

Calculates the Kullback-Leibler divergence between two distributions. Note that KL divergence is not symmetric, thus KL(p, q) != KL(q, p).

NOTE: Both distributions ‘p’ and ‘q’ should already be normalized to sum to one.

This method is equivalent to entropy(p, q, axis=1) from scipy_stats, only it’s much faster.

Parameters:

  • p – (NDArray) probability distribution.

  • q – (NDArray) probability distribution.

Returns:

(NDArray) Kullback-Leibler divergence.

star_pso.utils.auxiliary.kl_divergence_item(p: ndarray[Any, dtype[_ScalarType_co]], q: ndarray[Any, dtype[_ScalarType_co]]) → float[source]

Calculates the Kullback-Leibler divergence between two distributions. Note that KL divergence is not symmetric, thus KL(p, q) != KL(q, p).

NOTE: Both distributions ‘p’ and ‘q’ should already be normalized to sum to one.

This method is equivalent to entropy(p, q) from scipy_stats, only it’s much faster!

Parameters:

  • p – (NDArray) probability distribution.

  • q – (NDArray) probability distribution.

Returns:

(float) Kullback-Leibler divergence.

star_pso.utils.auxiliary.linear_rank_probabilities(p_size: int) → tuple[ndarray[Any, dtype[_ScalarType_co]], float][source]

Calculate the rank probability distribution over the population size. The function is cached so repeated calls with the same input should not recompute the same array since the population size of the swarm is not expected to change.

NOTE: Probabilities are returned in ascending order.

Parameters:

p_size – (int) population size.

Returns:

(array, float) rank probability distribution in ascending order along with their sum. Note: The sum should be one but due to small errors it might be less.

star_pso.utils.auxiliary.nb_cdist(x_pos: ndarray[Any, dtype[_ScalarType_co]], scaled: bool = False) → ndarray[Any, dtype[_ScalarType_co]][source]

This is equivalent to the scipy.spatial.distance.cdist method with Euclidean distance metric. It is a tailored version for the purposes of the multimodal operation mode.

Parameters:

  • x_pos – a numpy array of positions. The dimensions of the input array should [n_rows, n_cols], where n_rows is the number of particles and n_cols are the number of positions.

  • scaled – boolean flag that allows the input array to be scaled, using the MaxAbsScaler, before computing the distances.

Returns:

a square [n_rows, n_rows] numpy array of distances.

star_pso.utils.auxiliary.nb_centroid(x_pos: ndarray[Any, dtype[_ScalarType_co]]) → ndarray[Any, dtype[_ScalarType_co]][source]

Provides an auxiliary optimized function that calculate the centroid (i.e. x_pos.mean(axis=0)) for the input array.

Parameters:

x_pos – (array) A 2D numpy array (n_particles, n_features) representing the positions of the swarm.

Returns:

the mean values.

star_pso.utils.auxiliary.nb_clip_array(x_new: ndarray[Any, dtype[_ScalarType_co]], lower_limit: float | ndarray[Any, dtype[_ScalarType_co]], upper_limit: float | ndarray[Any, dtype[_ScalarType_co]]) → ndarray[Any, dtype[_ScalarType_co]][source]

Local version of numba clip which limits the values of an array. Given an interval values outside the interval are clipped to the interval edges.

Parameters:

  • x_new – array values to be clipped.

  • lower_limit – lower limit.

  • upper_limit – upper limit.

Returns:

the clipped array values.

star_pso.utils.auxiliary.nb_clip_inplace(x: ndarray[Any, dtype[_ScalarType_co]], x_min: float | ndarray[Any, dtype[_ScalarType_co]], x_max: float | ndarray[Any, dtype[_ScalarType_co]]) → None[source]

Local auxiliary function that is used to clip the values of input array ‘x’ to [x_min, x_max] range, and put the output inplace.

Parameters:

  • x – the numpy array we want to clip its values.

  • x_min – the minimum (lower bound).

  • x_max – the maximum (upper bound).

star_pso.utils.auxiliary.nb_clip_item(x_new: float | ndarray[Any, dtype[_ScalarType_co]], lower_limit: float | ndarray[Any, dtype[_ScalarType_co]], upper_limit: float | ndarray[Any, dtype[_ScalarType_co]]) → int | float[source]

Local version of numba clip which limits the values of a scalar. Given an interval values outside the interval are clipped to the interval edges. The final value is returned with item().

Parameters:

  • x_new – scalar value to be clipped.

  • lower_limit – lower limit.

  • upper_limit – upper limit.

Returns:

the clipped item value.

star_pso.utils.auxiliary.nb_median_euclidean_distance(x_pos: ndarray[Any, dtype[_ScalarType_co]], normal: bool = False) → float[source]

Calculates a measure of the particles spread, when their position is defined by continuous variables in ‘R’. First we calculate the centroid position of the swarm, and then we compute its distance from all the particles. To get an estimate in [0,1] we can divide them with the maximum distance (optional).

Parameters:

  • x_pos – (array) A 2D numpy array (n_particles, n_features) representing the positions of the swarm.

  • normal – (bool) if “True”, normalize the distances by their maximum distance.

Returns:

the median Euclidean distance.

star_pso.utils.auxiliary.nb_median_hamming_distance(x_pos: ndarray[Any, dtype[_ScalarType_co]], normal: bool = False) → float[source]

Compute the median Hamming distance of the input array. It is assumed that the input ‘x_pos’, represents the 2D array of particle binary positions {0, 1}.

Parameters:

  • x_pos – (array) A 2D numpy array (n_particles, n_features) representing the positions of the swarm.

  • normal – whether to compute the normalized average Hamming distance. This will yield a value between 0 and 1. A value of ‘0’ indicates that all rows are identical. On the contrary a value of ‘1’ indicates that all rows are completely different across all positions.

The normalization provides a clearer understanding of the diversity of the binary strings in relation to their length.

Returns:

the (normalized) median Hamming distance value.

star_pso.utils.auxiliary.nb_median_kl_divergence(x_pos: ndarray[Any, dtype[_ScalarType_co]], normal: bool = False) → float[source]

Calculate the ‘median KL divergence’ value of the input array. It is assumed that each row is a distribution (i.e. sum to 1).

Parameters:

  • x_pos – (array) A 2D numpy array (n_particles, n_features) representing the positions of the swarm.

  • normal – If enabled the KL values will be normalized using the maximum KL divergence from the data.

Returns:

The median KL divergence of the swarm positions.

star_pso.utils.auxiliary.nb_median_taxicab_distance(x_pos: ndarray[Any, dtype[_ScalarType_co]], normal: bool = False) → float[source]

Calculates a measure of the particles spread, when their position is defined by integer variables in ‘Z’. First we calculate the centroid position of the swarm, and then we compute its distance from all the particles. To get an estimate in [0,1] we can divide them with the maximum distance (optional).

Parameters:

  • x_pos – (array) A 2D numpy array (n_particles, n_features) representing the positions of the swarm.

  • normal – (bool) if “True”, normalize the distances by their maximum distance.

Returns:

the median TaxiCab (Manhattan) distance.

star_pso.utils.auxiliary.pareto_front(points: ndarray[Any, dtype[_ScalarType_co]]) → ndarray[Any, dtype[_ScalarType_co]][source]

Simple function that calculates the pareto (optimal) front points from a given input points numpy array.

Parameters:

points – array of points as [(fx1, fx2, …, fxn), (fy1, fy2, …, fyn), ……………….., (fk1, fk2, …, fkn)]

Returns:

array of points that lie on the pareto front.

star_pso.utils.auxiliary.spread_methods: dict = {BlockType.BINARY: CPUDispatcher(<function nb_median_hamming_distance>), BlockType.CATEGORICAL: CPUDispatcher(<function nb_median_kl_divergence>), BlockType.FLOAT: CPUDispatcher(<function nb_median_euclidean_distance>), BlockType.INTEGER: CPUDispatcher(<function nb_median_taxicab_distance>)}

Create a dictionary with block types as keys and their corresponding spread estimation methods as values.

star_pso.utils.auxiliary.time_it(func: Callable)[source]

Timing decorator function.

Parameters:

func – the function we want to time.

Returns:

the time wrapper method.

star_pso.utils.data_block module

class star_pso.utils.data_block.DataBlock(position: int | float | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], btype: BlockType, valid_set: list | tuple | None = None, lower_bound: list | Number | None = None, upper_bound: list | Number | None = None)[source]

Bases: object

Description:

This is the main class that encodes the data of a single particle variable. The class encapsulates not only the data (position and best position), but also the way that this data can be updated using a specific type dependent functions.

property best_position: int | float | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]

Accessor (getter) of the data block’s best position.

Returns:

the best recorded position value of the data block.

property block_t: BlockType

Accessor (getter) of the data block’s type.

Returns:

the data block type.

classmethod init_binary(**kwargs) → int[source]

It is used to initialize randomly the positions of a discrete ‘binary’ data blocks.

Parameters:

kwargs – contains the required parameters.

Returns:

a new random (integer) position.

static init_categorical(**kwargs) → int[source]

It is used to initialize randomly the positions of a discrete ‘categorical’ data blocks.

Parameters:

kwargs – contains the required parameters.

Returns:

a new random (integer) position.

classmethod init_float(**kwargs) → float[source]

It is used to initialize randomly the positions of a continuous ‘float’ data blocks.

Parameters:

kwargs – contains the required parameters.

Returns:

a new random (float) position.

classmethod init_integer(**kwargs) → int[source]

It is used to initialize randomly the positions of a discrete ‘int’ data blocks.

Parameters:

kwargs – contains the required parameters.

Returns:

a new random (integer) position.

static init_methods() → dict[source]

Create a dictionary with method names as keys and their corresponding initialization methods as values.

Returns:

a (cached) dictionary with functions that correspond to the correct block types.

property position: int | float | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]

Accessor (getter) of the data block’s position.

Returns:

the position value of the data block.

reset_position() → None[source]

This method provides a public interface for the reset of the new positions, for all types of data blocks.

Returns:

None.

rng: Generator = Generator(PCG64) at 0x7C2F37F83220

Random Number Generator for the whole class.

classmethod set_seed(new_seed=None) → None[source]

Sets a new seed for the random number generator.

Parameters:

new_seed – New seed value (default=None).

Returns:

None.

classmethod upd_binary(params: Params) → int[source]

It is used to update the positions of discrete ‘binary’ data blocks.

Parameters:

params – tuple which contains the parameters for the update equation.

Returns:

a new (binary) position.

classmethod upd_categorical(params: Params) → _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes][source]

It is used to update the positions of discrete ‘categorical’ data blocks.

Parameters:

params – tuple which contains the parameters for the update equation.

Returns:

a new array like with probabilities.

static upd_float(params: Params) → float[source]

It is used to update the positions of continuous ‘float’ data blocks.

Parameters:

params – tuple which contains the parameters for the update equation.

Returns:

a new (float) position.

static upd_integer(params: Params) → int[source]

It is used to update the positions of discrete ‘int’ data blocks.

Parameters:

params – tuple which contains the parameters for the update equation.

Returns:

a new (int) position.

static update_methods() → dict[source]

Return a dictionary with keys the method names and their corresponding update methods as values.

Returns:

a (cached) dictionary with functions that correspond to the correct block types.

property valid_set: list | tuple

Accessor (getter) method for the valid set.

Returns:

The set of valid values of a CATEGORICAL block.

Module contents

star_pso.utils.ScalarOrArray

Make a type alias for the position’s type.

alias of int | float | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]

class star_pso.utils.VOptions(w0, c1, c2, mode=False)

Bases: tuple

Declare a named tuple with the parameters we want to use in the velocity equations:

  • “w0”: inertia weight

  • “c1”: cognitive coefficient

  • “c2”: social coefficient

  • “mode”: “fipso”, “g_best”, “multimodal”.

c1

Alias for field number 1

c2

Alias for field number 2

mode

Alias for field number 3

w0

Alias for field number 0