Skip to content

desdeo-emo

Methods

desdeo.emo.methods.EAs

Implements common evolutionary algorithms for multi-objective optimization.

nsga3

nsga3(*, problem: Problem, seed: int = 0, n_generations: int = 100, reference_vector_options: ReferenceVectorOptions = None, forced_verbosity: int | None = None) -> tuple[Callable[[], EMOResult], Publisher]

Implements the NSGA-III algorithm as well as its interactive version.

References

K. Deb and H. Jain, “An Evolutionary Many-Objective Optimization Algorithm Using Reference-Point-Based Nondominated Sorting Approach, Part I: Solving Problems With Box Constraints,” IEEE Transactions on Evolutionary Computation, vol. 18, no. 4, pp. 577-601, Aug. 2014.

J. Hakanen, T. Chugh, K. Sindhya, Y. Jin, and K. Miettinen, “Connections of reference vectors and different types of preference information in interactive multiobjective evolutionary algorithms,” in 2016 IEEE Symposium Series on Computational Intelligence (SSCI), Athens, Greece: IEEE, Dec. 2016, pp. 1-8.

Parameters:

Name Type Description Default
problem Problem

The problem to be solved.

required
seed int

The seed for the random number generator. Defaults to 0.

0
n_generations int

The number of generations to run the algorithm. Defaults to 100.

100
reference_vector_options ReferenceVectorOptions

The options for the reference vectors. Defaults to None. See the ReferenceVectorOptions class for the defaults. This option can be used to run an interactive version of the algorithm, using preferences provided by the user.

None
forced_verbosity int

If not None, the verbosity of the algorithm is forced to this value. Defaults to None.

None

Returns:

Type Description
tuple[Callable[[], EMOResult], Publisher]

tuple[Callable[[], EMOResult], Publisher]: A tuple containing the function to run the algorithm and the publisher object. The publisher object can be used to subscribe to the topics of the algorithm components, as well as to inject additional functionality such as archiving the results.

Source code in desdeo/emo/methods/EAs.py
def nsga3(
    *,
    problem: Problem,
    seed: int = 0,
    n_generations: int = 100,
    reference_vector_options: ReferenceVectorOptions = None,
    forced_verbosity: int | None = None,
) -> tuple[Callable[[], EMOResult], Publisher]:
    """Implements the NSGA-III algorithm as well as its interactive version.

    References:
        K. Deb and H. Jain, “An Evolutionary Many-Objective Optimization Algorithm Using Reference-Point-Based
        Nondominated Sorting Approach, Part I: Solving Problems With Box Constraints,” IEEE Transactions on Evolutionary
        Computation, vol. 18, no. 4, pp. 577-601, Aug. 2014.

        J. Hakanen, T. Chugh, K. Sindhya, Y. Jin, and K. Miettinen, “Connections of reference vectors and different
        types of preference information in interactive multiobjective evolutionary algorithms,” in 2016 IEEE Symposium
        Series on Computational Intelligence (SSCI), Athens, Greece: IEEE, Dec. 2016, pp. 1-8.

    Args:
        problem (Problem): The problem to be solved.
        seed (int, optional): The seed for the random number generator. Defaults to 0.
        n_generations (int, optional): The number of generations to run the algorithm. Defaults to 100.
        reference_vector_options (ReferenceVectorOptions, optional): The options for the reference vectors. Defaults to
            None. See the ReferenceVectorOptions class for the defaults. This option can be used to run an interactive
            version of the algorithm, using preferences provided by the user.
        forced_verbosity (int, optional): If not None, the verbosity of the algorithm is forced to this value. Defaults
            to None.

    Returns:
        tuple[Callable[[], EMOResult], Publisher]: A tuple containing the function to run the algorithm and the
            publisher object. The publisher object can be used to subscribe to the topics of the algorithm components,
            as well as to inject additional functionality such as archiving the results.
    """
    publisher = Publisher()
    evaluator = EMOEvaluator(
        problem=problem,
        publisher=publisher,
        verbosity=forced_verbosity if forced_verbosity is not None else 2,
    )

    selector = NSGAIII_select(
        problem=problem,
        publisher=publisher,
        verbosity=forced_verbosity if forced_verbosity is not None else 2,
        reference_vector_options=reference_vector_options,
    )

    # Note that the initial population size is equal to the number of reference vectors
    n_points = selector.reference_vectors.shape[0]

    generator = LHSGenerator(
        problem=problem,
        evaluator=evaluator,
        publisher=publisher,
        n_points=n_points,
        seed=seed,
        verbosity=forced_verbosity if forced_verbosity is not None else 1,
    )
    crossover = SimulatedBinaryCrossover(
        problem=problem,
        publisher=publisher,
        seed=seed,
        verbosity=forced_verbosity if forced_verbosity is not None else 1,
    )
    mutation = BoundedPolynomialMutation(
        problem=problem,
        publisher=publisher,
        seed=seed,
        verbosity=forced_verbosity if forced_verbosity is not None else 1,
    )

    terminator = MaxGenerationsTerminator(
        n_generations,
        publisher=publisher,
    )

    components = [evaluator, generator, crossover, mutation, selector, terminator]
    [publisher.auto_subscribe(x) for x in components]
    [publisher.register_topics(x.provided_topics[x.verbosity], x.__class__.__name__) for x in components]

    return (
        partial(
            template1,
            evaluator=evaluator,
            crossover=crossover,
            mutation=mutation,
            generator=generator,
            selection=selector,
            terminator=terminator,
        ),
        publisher,
    )

rvea

rvea(*, problem: Problem, seed: int = 0, n_generations=100, reference_vector_options: ReferenceVectorOptions = None, forced_verbosity: int | None = None) -> tuple[Callable[[], EMOResult], Publisher]

Implements the Reference Vector Guided Evolutionary Algorithm (RVEA), as well as its interactive version.

References

R. Cheng, Y. Jin, M. Olhofer and B. Sendhoff, "A Reference Vector Guided Evolutionary Algorithm for Many- Objective Optimization," in IEEE Transactions on Evolutionary Computation, vol. 20, no. 5, pp. 773-791, Oct. 2016, doi: 10.1109/TEVC.2016.2519378.

J. Hakanen, T. Chugh, K. Sindhya, Y. Jin, and K. Miettinen, “Connections of reference vectors and different types of preference information in interactive multiobjective evolutionary algorithms,” in 2016 IEEE Symposium Series on Computational Intelligence (SSCI), Athens, Greece: IEEE, Dec. 2016, pp. 1-8.

Parameters:

Name Type Description Default
problem Problem

The problem to be solved.

required
seed int

The seed for the random number generator. Defaults to 0.

0
n_generations int

The number of generations to run the algorithm. Defaults to 100.

100
reference_vector_options ReferenceVectorOptions

The options for the reference vectors. Defaults to None. See the ReferenceVectorOptions class for the defaults. This option can be used to run an interactive version of the algorithm, using preferences provided by the user.

None
forced_verbosity int

If not None, the verbosity of the algorithm is forced to this value. Defaults to None.

None

Returns:

Type Description
tuple[Callable[[], EMOResult], Publisher]

tuple[Callable[[], EMOResult], Publisher]: A tuple containing the function to run the algorithm and the publisher object. The publisher object can be used to subscribe to the topics of the algorithm components, as well as to inject additional functionality such as archiving the results.

Source code in desdeo/emo/methods/EAs.py
def rvea(
    *,
    problem: Problem,
    seed: int = 0,
    n_generations=100,
    reference_vector_options: ReferenceVectorOptions = None,
    forced_verbosity: int | None = None,
) -> tuple[Callable[[], EMOResult], Publisher]:
    """Implements the Reference Vector Guided Evolutionary Algorithm (RVEA), as well as its interactive version.

    References:
        R. Cheng, Y. Jin, M. Olhofer and B. Sendhoff, "A Reference Vector Guided Evolutionary Algorithm for Many-
        Objective Optimization," in IEEE Transactions on Evolutionary Computation, vol. 20, no. 5, pp. 773-791,
        Oct. 2016, doi: 10.1109/TEVC.2016.2519378.

        J. Hakanen, T. Chugh, K. Sindhya, Y. Jin, and K. Miettinen, “Connections of reference vectors and different
        types of preference information in interactive multiobjective evolutionary algorithms,” in 2016 IEEE Symposium
        Series on Computational Intelligence (SSCI), Athens, Greece: IEEE, Dec. 2016, pp. 1-8.


    Args:
        problem (Problem): The problem to be solved.
        seed (int, optional): The seed for the random number generator. Defaults to 0.
        n_generations (int, optional): The number of generations to run the algorithm. Defaults to 100.
        reference_vector_options (ReferenceVectorOptions, optional): The options for the reference vectors. Defaults to
            None. See the ReferenceVectorOptions class for the defaults. This option can be used to run an interactive
            version of the algorithm, using preferences provided by the user.
        forced_verbosity (int, optional): If not None, the verbosity of the algorithm is forced to this value. Defaults
            to None.

    Returns:
        tuple[Callable[[], EMOResult], Publisher]: A tuple containing the function to run the algorithm and the
            publisher object. The publisher object can be used to subscribe to the topics of the algorithm components,
            as well as to inject additional functionality such as archiving the results.
    """
    publisher = Publisher()
    evaluator = EMOEvaluator(
        problem=problem,
        publisher=publisher,
        verbosity=forced_verbosity if forced_verbosity is not None else 2,
    )

    selector = RVEASelector(
        problem=problem,
        publisher=publisher,
        reference_vector_options=reference_vector_options,
        verbosity=forced_verbosity if forced_verbosity is not None else 2,
    )

    # Note that the initial population size is equal to the number of reference vectors
    n_points = selector.reference_vectors.shape[0]

    generator = LHSGenerator(
        problem=problem,
        evaluator=evaluator,
        publisher=publisher,
        n_points=n_points,
        seed=seed,
        verbosity=forced_verbosity if forced_verbosity is not None else 1,
    )
    crossover = SimulatedBinaryCrossover(
        problem=problem,
        publisher=publisher,
        seed=seed,
        verbosity=forced_verbosity if forced_verbosity is not None else 1,
    )
    mutation = BoundedPolynomialMutation(
        problem=problem,
        publisher=publisher,
        seed=seed,
        verbosity=forced_verbosity if forced_verbosity is not None else 1,
    )

    terminator = MaxGenerationsTerminator(n_generations, publisher=publisher)

    components = [evaluator, generator, crossover, mutation, selector, terminator]
    [publisher.auto_subscribe(x) for x in components]
    [publisher.register_topics(x.provided_topics[x.verbosity], x.__class__.__name__) for x in components]

    return (
        partial(
            template1,
            evaluator=evaluator,
            crossover=crossover,
            mutation=mutation,
            generator=generator,
            selection=selector,
            terminator=terminator,
        ),
        publisher,
    )

Templates

desdeo.emo.methods.bases

This module contains the basic functional implementations for the EMO methods.

This can be used as a template for the implementation of the EMO methods.

EMOResult

Bases: BaseModel

Source code in desdeo/emo/methods/bases.py
class EMOResult(BaseModel):
    solutions: pl.DataFrame = Field(description="The decision vectors of the final population.")
    """The decision vectors of the final population."""
    outputs: pl.DataFrame = Field(
        description="The objective vectors, constraint vectors, and targets of the final population."
    )
    """The objective vectors, constraint vectors, and targets of the final population."""

    model_config = ConfigDict(arbitrary_types_allowed=True)
outputs class-attribute instance-attribute
outputs: DataFrame = Field(description='The objective vectors, constraint vectors, and targets of the final population.')

The objective vectors, constraint vectors, and targets of the final population.

solutions class-attribute instance-attribute
solutions: DataFrame = Field(description='The decision vectors of the final population.')

The decision vectors of the final population.

template1

template1(evaluator: EMOEvaluator, crossover: BaseCrossover, mutation: BaseMutation, generator: BaseGenerator, selection: BaseSelector, terminator: BaseTerminator) -> EMOResult

Implements a template that many EMO methods, such as RVEA and NSGA-III, follow.

Parameters:

Name Type Description Default
evaluator EMOEvaluator

A class that evaluates the solutions and provides the objective vectors, constraint vectors, and targets.

required
crossover BaseCrossover

The crossover operator.

required
mutation BaseMutation

The mutation operator.

required
generator BaseGenerator

A class that generates the initial population.

required
selection BaseSelector

The selection operator.

required
terminator BaseTerminator

The termination operator.

required

Returns:

Name Type Description
EMOResult EMOResult

The final population and their objective vectors, constraint vectors, and targets

Source code in desdeo/emo/methods/bases.py
def template1(
    evaluator: EMOEvaluator,
    crossover: BaseCrossover,
    mutation: BaseMutation,
    generator: BaseGenerator,
    selection: BaseSelector,
    terminator: BaseTerminator,
) -> EMOResult:
    """Implements a template that many EMO methods, such as RVEA and NSGA-III, follow.

    Args:
        evaluator (EMOEvaluator): A class that evaluates the solutions and provides the objective vectors, constraint
            vectors, and targets.
        crossover (BaseCrossover): The crossover operator.
        mutation (BaseMutation): The mutation operator.
        generator (BaseGenerator): A class that generates the initial population.
        selection (BaseSelector): The selection operator.
        terminator (BaseTerminator): The termination operator.

    Returns:
        EMOResult: The final population and their objective vectors, constraint vectors, and targets
    """
    solutions, outputs = generator.do()

    while not terminator.check():
        offspring = crossover.do(population=solutions)
        offspring = mutation.do(offspring, solutions)
        offspring_outputs = evaluator.evaluate(offspring)
        solutions, outputs = selection.do(parents=(solutions, outputs), offsprings=(offspring, offspring_outputs))

    return EMOResult(solutions=solutions, outputs=outputs)

Generators

desdeo.emo.operators.generator

Class for generating initial population for the evolutionary optimization algorithms.

BaseGenerator

Bases: Subscriber

Base class for generating initial population for the evolutionary optimization algorithms.

This class should be inherited by the classes that implement the initial population generation for the evolutionary optimization algorithms.

Source code in desdeo/emo/operators/generator.py
class BaseGenerator(Subscriber):
    """Base class for generating initial population for the evolutionary optimization algorithms.

    This class should be inherited by the classes that implement the initial population generation
    for the evolutionary optimization algorithms.

    """

    @property
    def provided_topics(self) -> dict[int, Sequence[GeneratorMessageTopics]]:
        """Return the topics provided by the generator.

        Returns:
            dict[int, Sequence[GeneratorMessageTopics]]: The topics provided by the generator.
        """
        return {
            0: [],
            1: [GeneratorMessageTopics.NEW_EVALUATIONS],
            2: [
                GeneratorMessageTopics.NEW_EVALUATIONS,
                GeneratorMessageTopics.POPULATION,
                GeneratorMessageTopics.OUTPUTS,
            ],
        }

    @property
    def interested_topics(self):
        """Return the message topics that the generator is interested in."""
        return []

    def __init__(self, problem: Problem, **kwargs):
        """Initialize the BaseGenerator class."""
        super().__init__(**kwargs)
        self.problem = problem
        self.variable_symbols = [var.symbol for var in problem.get_flattened_variables()]
        self.bounds = np.array([[var.lowerbound, var.upperbound] for var in problem.get_flattened_variables()])
        self.population: pl.DataFrame = None
        self.out: pl.DataFrame = None

    @abstractmethod
    def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
        """Generate the initial population.

        This method should be implemented by the inherited classes.

        Returns:
            tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
                the corresponding objectives, the constraint violations, and the targets as the
                second element.
        """

    def state(self) -> Sequence[Message]:
        """Return the state of the generator.

        This method should be implemented by the inherited classes.

        Returns:
            dict: The state of the generator.
        """
        if self.population is None or self.out is None or self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return [
                IntMessage(
                    topic=GeneratorMessageTopics.NEW_EVALUATIONS,
                    value=self.population.shape[0],
                    source=self.__class__.__name__,
                ),
            ]
        # verbosity == 2
        return [
            PolarsDataFrameMessage(
                topic=GeneratorMessageTopics.POPULATION,
                value=self.population,
                source=self.__class__.__name__,
            ),
            PolarsDataFrameMessage(
                topic=GeneratorMessageTopics.OUTPUTS,
                value=self.out,
                source=self.__class__.__name__,
            ),
            IntMessage(
                topic=GeneratorMessageTopics.NEW_EVALUATIONS,
                value=self.population.shape[0],
                source=self.__class__.__name__,
            ),
        ]
interested_topics property
interested_topics

Return the message topics that the generator is interested in.

provided_topics property
provided_topics: dict[int, Sequence[GeneratorMessageTopics]]

Return the topics provided by the generator.

Returns:

Type Description
dict[int, Sequence[GeneratorMessageTopics]]

dict[int, Sequence[GeneratorMessageTopics]]: The topics provided by the generator.

__init__
__init__(problem: Problem, **kwargs)

Initialize the BaseGenerator class.

Source code in desdeo/emo/operators/generator.py
def __init__(self, problem: Problem, **kwargs):
    """Initialize the BaseGenerator class."""
    super().__init__(**kwargs)
    self.problem = problem
    self.variable_symbols = [var.symbol for var in problem.get_flattened_variables()]
    self.bounds = np.array([[var.lowerbound, var.upperbound] for var in problem.get_flattened_variables()])
    self.population: pl.DataFrame = None
    self.out: pl.DataFrame = None
do abstractmethod
do() -> tuple[pl.DataFrame, pl.DataFrame]

Generate the initial population.

This method should be implemented by the inherited classes.

Returns:

Type Description
tuple[DataFrame, DataFrame]

tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element, the corresponding objectives, the constraint violations, and the targets as the second element.

Source code in desdeo/emo/operators/generator.py
@abstractmethod
def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Generate the initial population.

    This method should be implemented by the inherited classes.

    Returns:
        tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
            the corresponding objectives, the constraint violations, and the targets as the
            second element.
    """
state
state() -> Sequence[Message]

Return the state of the generator.

This method should be implemented by the inherited classes.

Returns:

Name Type Description
dict Sequence[Message]

The state of the generator.

Source code in desdeo/emo/operators/generator.py
def state(self) -> Sequence[Message]:
    """Return the state of the generator.

    This method should be implemented by the inherited classes.

    Returns:
        dict: The state of the generator.
    """
    if self.population is None or self.out is None or self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return [
            IntMessage(
                topic=GeneratorMessageTopics.NEW_EVALUATIONS,
                value=self.population.shape[0],
                source=self.__class__.__name__,
            ),
        ]
    # verbosity == 2
    return [
        PolarsDataFrameMessage(
            topic=GeneratorMessageTopics.POPULATION,
            value=self.population,
            source=self.__class__.__name__,
        ),
        PolarsDataFrameMessage(
            topic=GeneratorMessageTopics.OUTPUTS,
            value=self.out,
            source=self.__class__.__name__,
        ),
        IntMessage(
            topic=GeneratorMessageTopics.NEW_EVALUATIONS,
            value=self.population.shape[0],
            source=self.__class__.__name__,
        ),
    ]

LHSGenerator

Bases: RandomGenerator

Class for generating Latin Hypercube Sampling (LHS) initial population for the MOEAs.

This class generates the initial population by using the Latin Hypercube Sampling (LHS) method. If the seed is not provided, the seed is set to 0.

Source code in desdeo/emo/operators/generator.py
class LHSGenerator(RandomGenerator):
    """Class for generating Latin Hypercube Sampling (LHS) initial population for the MOEAs.

    This class generates the initial population by using the Latin Hypercube Sampling (LHS) method.
    If the seed is not provided, the seed is set to 0.
    """

    def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
        """Initialize the LHSGenerator class.

        Args:
            problem (Problem): The problem to solve.
            evaluator (BaseEvaluator): The evaluator to evaluate the population.
            n_points (int): The number of points to generate for the initial population.
            seed (int): The seed for the random number generator.
            kwargs: Additional keyword arguments. Check the Subscriber class for more information.
                At the very least, the publisher argument should be provided.
        """
        super().__init__(problem, evaluator, n_points, seed, **kwargs)
        self.lhsrng = LatinHypercube(d=len(self.variable_symbols), seed=seed)
        self.seed = seed

    def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
        """Generate the initial population.

        This method should be implemented by the inherited classes.

        Returns:
            tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
                the corresponding objectives, the constraint violations, and the targets as the second element.
        """
        if self.population is not None and self.out is not None:
            self.notify()
            return self.population, self.out
        self.population = pl.from_numpy(
            self.lhsrng.random(n=self.n_points) * (self.bounds[:, 1] - self.bounds[:, 0]) + self.bounds[:, 0],
            schema=self.variable_symbols,
        )
        self.out = self.evaluator.evaluate(self.population)
        self.notify()
        return self.population, self.out

    def update(self, message) -> None:
        """Update the generator based on the message."""
__init__
__init__(problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs)

Initialize the LHSGenerator class.

Parameters:

Name Type Description Default
problem Problem

The problem to solve.

required
evaluator BaseEvaluator

The evaluator to evaluate the population.

required
n_points int

The number of points to generate for the initial population.

required
seed int

The seed for the random number generator.

required
kwargs

Additional keyword arguments. Check the Subscriber class for more information. At the very least, the publisher argument should be provided.

{}
Source code in desdeo/emo/operators/generator.py
def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
    """Initialize the LHSGenerator class.

    Args:
        problem (Problem): The problem to solve.
        evaluator (BaseEvaluator): The evaluator to evaluate the population.
        n_points (int): The number of points to generate for the initial population.
        seed (int): The seed for the random number generator.
        kwargs: Additional keyword arguments. Check the Subscriber class for more information.
            At the very least, the publisher argument should be provided.
    """
    super().__init__(problem, evaluator, n_points, seed, **kwargs)
    self.lhsrng = LatinHypercube(d=len(self.variable_symbols), seed=seed)
    self.seed = seed
do
do() -> tuple[pl.DataFrame, pl.DataFrame]

Generate the initial population.

This method should be implemented by the inherited classes.

Returns:

Type Description
tuple[DataFrame, DataFrame]

tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element, the corresponding objectives, the constraint violations, and the targets as the second element.

Source code in desdeo/emo/operators/generator.py
def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Generate the initial population.

    This method should be implemented by the inherited classes.

    Returns:
        tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
            the corresponding objectives, the constraint violations, and the targets as the second element.
    """
    if self.population is not None and self.out is not None:
        self.notify()
        return self.population, self.out
    self.population = pl.from_numpy(
        self.lhsrng.random(n=self.n_points) * (self.bounds[:, 1] - self.bounds[:, 0]) + self.bounds[:, 0],
        schema=self.variable_symbols,
    )
    self.out = self.evaluator.evaluate(self.population)
    self.notify()
    return self.population, self.out
update
update(message) -> None

Update the generator based on the message.

Source code in desdeo/emo/operators/generator.py
def update(self, message) -> None:
    """Update the generator based on the message."""

RandomBinaryGenerator

Bases: BaseGenerator

Class for generating random initial population for problems with binary variables.

This class generates an initial population by randomly setting variable values to be either 0 or 1.

Source code in desdeo/emo/operators/generator.py
class RandomBinaryGenerator(BaseGenerator):
    """Class for generating random initial population for problems with binary variables.

    This class generates an initial population by randomly setting variable values to be either 0 or 1.
    """

    def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
        """Initialize the RandomBinaryGenerator class.

        Args:
            problem (Problem): The problem to solve.
            evaluator (BaseEvaluator): The evaluator to evaluate the population.
            n_points (int): The number of points to generate for the initial population.
            seed (int): The seed for the random number generator.
            kwargs: Additional keyword arguments. Check the Subscriber class for more information.
                At the very least, the publisher argument should be provided.
        """
        super().__init__(problem, **kwargs)
        self.n_points = n_points
        self.evaluator = evaluator
        self.rng = np.random.default_rng(seed)
        self.seed = seed

    def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
        """Generate the initial population.

        Returns:
            tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
                the corresponding objectives, the constraint violations, and the targets as the second element.
        """
        if self.population is not None and self.out is not None:
            self.notify()
            return self.population, self.out

        self.population = pl.from_numpy(
            self.rng.integers(low=0, high=2, size=(self.n_points, self.bounds.shape[0])).astype(dtype=np.float64),
            schema=self.variable_symbols,
        )

        self.out = self.evaluator.evaluate(self.population)
        self.notify()
        return self.population, self.out

    def update(self, message) -> None:
        """Update the generator based on the message."""
__init__
__init__(problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs)

Initialize the RandomBinaryGenerator class.

Parameters:

Name Type Description Default
problem Problem

The problem to solve.

required
evaluator BaseEvaluator

The evaluator to evaluate the population.

required
n_points int

The number of points to generate for the initial population.

required
seed int

The seed for the random number generator.

required
kwargs

Additional keyword arguments. Check the Subscriber class for more information. At the very least, the publisher argument should be provided.

{}
Source code in desdeo/emo/operators/generator.py
def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
    """Initialize the RandomBinaryGenerator class.

    Args:
        problem (Problem): The problem to solve.
        evaluator (BaseEvaluator): The evaluator to evaluate the population.
        n_points (int): The number of points to generate for the initial population.
        seed (int): The seed for the random number generator.
        kwargs: Additional keyword arguments. Check the Subscriber class for more information.
            At the very least, the publisher argument should be provided.
    """
    super().__init__(problem, **kwargs)
    self.n_points = n_points
    self.evaluator = evaluator
    self.rng = np.random.default_rng(seed)
    self.seed = seed
do
do() -> tuple[pl.DataFrame, pl.DataFrame]

Generate the initial population.

Returns:

Type Description
tuple[DataFrame, DataFrame]

tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element, the corresponding objectives, the constraint violations, and the targets as the second element.

Source code in desdeo/emo/operators/generator.py
def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Generate the initial population.

    Returns:
        tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
            the corresponding objectives, the constraint violations, and the targets as the second element.
    """
    if self.population is not None and self.out is not None:
        self.notify()
        return self.population, self.out

    self.population = pl.from_numpy(
        self.rng.integers(low=0, high=2, size=(self.n_points, self.bounds.shape[0])).astype(dtype=np.float64),
        schema=self.variable_symbols,
    )

    self.out = self.evaluator.evaluate(self.population)
    self.notify()
    return self.population, self.out
update
update(message) -> None

Update the generator based on the message.

Source code in desdeo/emo/operators/generator.py
def update(self, message) -> None:
    """Update the generator based on the message."""

RandomGenerator

Bases: BaseGenerator

Class for generating random initial population for the evolutionary optimization algorithms.

This class generates the initial population by randomly sampling the points from the variable bounds. The distribution of the points is uniform. If the seed is not provided, the seed is set to 0.

Source code in desdeo/emo/operators/generator.py
class RandomGenerator(BaseGenerator):
    """Class for generating random initial population for the evolutionary optimization algorithms.

    This class generates the initial population by randomly sampling the points from the variable bounds. The
    distribution of the points is uniform. If the seed is not provided, the seed is set to 0.
    """

    def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
        """Initialize the RandomGenerator class.

        Args:
            problem (Problem): The problem to solve.
            evaluator (BaseEvaluator): The evaluator to evaluate the population.
            n_points (int): The number of points to generate for the initial population.
            seed (int): The seed for the random number generator.
            kwargs: Additional keyword arguments. Check the Subscriber class for more information.
                At the very least, the publisher argument should be provided.
        """
        super().__init__(problem, **kwargs)
        self.n_points = n_points
        self.evaluator = evaluator
        self.rng = np.random.default_rng(seed)
        self.seed = seed

    def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
        """Generate the initial population.

        This method should be implemented by the inherited classes.

        Returns:
            tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
                the corresponding objectives, the constraint violations, and the targets as the second element.
        """
        if self.population is not None and self.out is not None:
            self.notify()
            return self.population, self.out
        self.population = pl.from_numpy(
            self.rng.uniform(low=self.bounds[:, 0], high=self.bounds[:, 1], size=(self.n_points, self.bounds.shape[0])),
            schema=self.variable_symbols,
        )
        self.out = self.evaluator.evaluate(self.population)
        self.notify()
        return self.population, self.out

    def update(self, message) -> None:
        """Update the generator based on the message."""
__init__
__init__(problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs)

Initialize the RandomGenerator class.

Parameters:

Name Type Description Default
problem Problem

The problem to solve.

required
evaluator BaseEvaluator

The evaluator to evaluate the population.

required
n_points int

The number of points to generate for the initial population.

required
seed int

The seed for the random number generator.

required
kwargs

Additional keyword arguments. Check the Subscriber class for more information. At the very least, the publisher argument should be provided.

{}
Source code in desdeo/emo/operators/generator.py
def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
    """Initialize the RandomGenerator class.

    Args:
        problem (Problem): The problem to solve.
        evaluator (BaseEvaluator): The evaluator to evaluate the population.
        n_points (int): The number of points to generate for the initial population.
        seed (int): The seed for the random number generator.
        kwargs: Additional keyword arguments. Check the Subscriber class for more information.
            At the very least, the publisher argument should be provided.
    """
    super().__init__(problem, **kwargs)
    self.n_points = n_points
    self.evaluator = evaluator
    self.rng = np.random.default_rng(seed)
    self.seed = seed
do
do() -> tuple[pl.DataFrame, pl.DataFrame]

Generate the initial population.

This method should be implemented by the inherited classes.

Returns:

Type Description
tuple[DataFrame, DataFrame]

tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element, the corresponding objectives, the constraint violations, and the targets as the second element.

Source code in desdeo/emo/operators/generator.py
def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Generate the initial population.

    This method should be implemented by the inherited classes.

    Returns:
        tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
            the corresponding objectives, the constraint violations, and the targets as the second element.
    """
    if self.population is not None and self.out is not None:
        self.notify()
        return self.population, self.out
    self.population = pl.from_numpy(
        self.rng.uniform(low=self.bounds[:, 0], high=self.bounds[:, 1], size=(self.n_points, self.bounds.shape[0])),
        schema=self.variable_symbols,
    )
    self.out = self.evaluator.evaluate(self.population)
    self.notify()
    return self.population, self.out
update
update(message) -> None

Update the generator based on the message.

Source code in desdeo/emo/operators/generator.py
def update(self, message) -> None:
    """Update the generator based on the message."""

RandomIntegerGenerator

Bases: BaseGenerator

Class for generating random initial population for problems with integer variables.

This class generates an initial population by randomly setting variable values to be integers between the bounds of the variables.

Source code in desdeo/emo/operators/generator.py
class RandomIntegerGenerator(BaseGenerator):
    """Class for generating random initial population for problems with integer variables.

    This class generates an initial population by randomly setting variable values to be integers between the bounds of
    the variables.
    """

    def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
        """Initialize the RandomIntegerGenerator class.

        Args:
            problem (Problem): The problem to solve.
            evaluator (BaseEvaluator): The evaluator to evaluate the population.
            n_points (int): The number of points to generate for the initial population.
            seed (int): The seed for the random number generator.
            kwargs: Additional keyword arguments. Check the Subscriber class for more information.
                At the very least, the publisher argument should be provided.
        """
        super().__init__(problem, **kwargs)
        self.n_points = n_points
        self.evaluator = evaluator
        self.rng = np.random.default_rng(seed)
        self.seed = seed

    def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
        """Generate the initial population.

        Returns:
            tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
                the corresponding objectives, the constraint violations, and the targets as the second element.
        """
        if self.population is not None and self.out is not None:
            self.notify()
            return self.population, self.out

        self.population = pl.from_numpy(
            self.rng.integers(
                low=self.bounds[:, 0],
                high=self.bounds[:, 1],
                size=(self.n_points, self.bounds.shape[0]),
                endpoint=True,
            ).astype(dtype=float),
            schema=self.variable_symbols,
        )

        self.out = self.evaluator.evaluate(self.population)
        self.notify()
        return self.population, self.out

    def update(self, message) -> None:
        """Update the generator based on the message."""
__init__
__init__(problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs)

Initialize the RandomIntegerGenerator class.

Parameters:

Name Type Description Default
problem Problem

The problem to solve.

required
evaluator BaseEvaluator

The evaluator to evaluate the population.

required
n_points int

The number of points to generate for the initial population.

required
seed int

The seed for the random number generator.

required
kwargs

Additional keyword arguments. Check the Subscriber class for more information. At the very least, the publisher argument should be provided.

{}
Source code in desdeo/emo/operators/generator.py
def __init__(self, problem: Problem, evaluator: EMOEvaluator, n_points: int, seed: int, **kwargs):
    """Initialize the RandomIntegerGenerator class.

    Args:
        problem (Problem): The problem to solve.
        evaluator (BaseEvaluator): The evaluator to evaluate the population.
        n_points (int): The number of points to generate for the initial population.
        seed (int): The seed for the random number generator.
        kwargs: Additional keyword arguments. Check the Subscriber class for more information.
            At the very least, the publisher argument should be provided.
    """
    super().__init__(problem, **kwargs)
    self.n_points = n_points
    self.evaluator = evaluator
    self.rng = np.random.default_rng(seed)
    self.seed = seed
do
do() -> tuple[pl.DataFrame, pl.DataFrame]

Generate the initial population.

Returns:

Type Description
tuple[DataFrame, DataFrame]

tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element, the corresponding objectives, the constraint violations, and the targets as the second element.

Source code in desdeo/emo/operators/generator.py
def do(self) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Generate the initial population.

    Returns:
        tuple[pl.DataFrame, pl.DataFrame]: The initial population as the first element,
            the corresponding objectives, the constraint violations, and the targets as the second element.
    """
    if self.population is not None and self.out is not None:
        self.notify()
        return self.population, self.out

    self.population = pl.from_numpy(
        self.rng.integers(
            low=self.bounds[:, 0],
            high=self.bounds[:, 1],
            size=(self.n_points, self.bounds.shape[0]),
            endpoint=True,
        ).astype(dtype=float),
        schema=self.variable_symbols,
    )

    self.out = self.evaluator.evaluate(self.population)
    self.notify()
    return self.population, self.out
update
update(message) -> None

Update the generator based on the message.

Source code in desdeo/emo/operators/generator.py
def update(self, message) -> None:
    """Update the generator based on the message."""

Evaluator

desdeo.emo.operators.evaluator

Classes for evaluating the objectives and constraints of the individuals in the population.

EMOEvaluator

Bases: Subscriber

Base class for evaluating the objectives and constraints of the individuals in the population.

This class should be inherited by the classes that implement the evaluation of the objectives and constraints of the individuals in the population.

Source code in desdeo/emo/operators/evaluator.py
class EMOEvaluator(Subscriber):
    """Base class for evaluating the objectives and constraints of the individuals in the population.

    This class should be inherited by the classes that implement the evaluation of the objectives
    and constraints of the individuals in the population.

    """

    @property
    def provided_topics(self) -> dict[int, Sequence[EvaluatorMessageTopics]]:
        """The topics provided by the Evaluator."""
        return {
            0: [],
            1: [EvaluatorMessageTopics.NEW_EVALUATIONS],
            2: [
                EvaluatorMessageTopics.NEW_EVALUATIONS,
                EvaluatorMessageTopics.POPULATION,
                EvaluatorMessageTopics.OUTPUTS,
            ],
        }

    @property
    def interested_topics(self):
        """The topics that the Evaluator is interested in."""
        return []

    def __init__(
        self,
        problem: Problem,
        verbosity: int = 1,
        **kwargs,
    ):
        """Initialize the EMOEvaluator class."""
        super().__init__(**kwargs)
        self.problem = problem
        # TODO(@light-weaver, @gialmisi): This can be so much more efficient.
        self.evaluator = lambda x: Evaluator(problem).evaluate(
            {name.symbol: x[name.symbol].to_list() for name in problem.get_flattened_variables()}
        )
        self.variable_symbols = [name.symbol for name in problem.variables]
        self.population: pl.DataFrame
        self.outs: pl.DataFrame
        self.verbosity: int = verbosity
        self.new_evals: int = 0

    def evaluate(self, population: pl.DataFrame) -> pl.DataFrame:
        """Evaluate and return the objectives.

        Args:
            population (pl.Dataframe): The set of decision variables to evaluate.

        Returns:
            pl.Dataframe: A dataframe of objective vectors, target vectors, and constraint vectors.
        """
        self.population = population
        out = self.evaluator(population)
        # remove variable_symbols from the output
        self.out = out.drop(self.variable_symbols)
        self.new_evals = len(population)
        # merge the objectives and targets

        self.notify()
        return self.out

    def state(self) -> Sequence[Message]:
        """The state of the evaluator sent to the Publisher."""
        if self.population is None or self.out is None or self.population is None or self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return [
                IntMessage(
                    topic=EvaluatorMessageTopics.NEW_EVALUATIONS,
                    value=self.new_evals,
                    source="EMOEvaluator",
                )
            ]

        if isinstance(self.population, pl.DataFrame):
            population_message = PolarsDataFrameMessage(
                topic=EvaluatorMessageTopics.POPULATION,
                value=self.population,
                source="EMOEvaluator",
            )
        else:
            population_message = GenericMessage(
                topic=EvaluatorMessageTopics.POPULATION,
                value="Population is not a polars DataFrame",
                source="EMOEvaluator",
            )
        return [
            IntMessage(
                topic=EvaluatorMessageTopics.NEW_EVALUATIONS,
                value=self.new_evals,
                source="EMOEvaluator",
            ),
            population_message,
            PolarsDataFrameMessage(
                topic=EvaluatorMessageTopics.OUTPUTS,
                value=self.out,
                source="EMOEvaluator",
            ),
        ]

    def update(self, *_, **__):
        """Update the parameters of the evaluator."""
interested_topics property
interested_topics

The topics that the Evaluator is interested in.

provided_topics property
provided_topics: dict[int, Sequence[EvaluatorMessageTopics]]

The topics provided by the Evaluator.

__init__
__init__(problem: Problem, verbosity: int = 1, **kwargs)

Initialize the EMOEvaluator class.

Source code in desdeo/emo/operators/evaluator.py
def __init__(
    self,
    problem: Problem,
    verbosity: int = 1,
    **kwargs,
):
    """Initialize the EMOEvaluator class."""
    super().__init__(**kwargs)
    self.problem = problem
    # TODO(@light-weaver, @gialmisi): This can be so much more efficient.
    self.evaluator = lambda x: Evaluator(problem).evaluate(
        {name.symbol: x[name.symbol].to_list() for name in problem.get_flattened_variables()}
    )
    self.variable_symbols = [name.symbol for name in problem.variables]
    self.population: pl.DataFrame
    self.outs: pl.DataFrame
    self.verbosity: int = verbosity
    self.new_evals: int = 0
evaluate
evaluate(population: pl.DataFrame) -> pl.DataFrame

Evaluate and return the objectives.

Parameters:

Name Type Description Default
population Dataframe

The set of decision variables to evaluate.

required

Returns:

Type Description
DataFrame

pl.Dataframe: A dataframe of objective vectors, target vectors, and constraint vectors.

Source code in desdeo/emo/operators/evaluator.py
def evaluate(self, population: pl.DataFrame) -> pl.DataFrame:
    """Evaluate and return the objectives.

    Args:
        population (pl.Dataframe): The set of decision variables to evaluate.

    Returns:
        pl.Dataframe: A dataframe of objective vectors, target vectors, and constraint vectors.
    """
    self.population = population
    out = self.evaluator(population)
    # remove variable_symbols from the output
    self.out = out.drop(self.variable_symbols)
    self.new_evals = len(population)
    # merge the objectives and targets

    self.notify()
    return self.out
state
state() -> Sequence[Message]

The state of the evaluator sent to the Publisher.

Source code in desdeo/emo/operators/evaluator.py
def state(self) -> Sequence[Message]:
    """The state of the evaluator sent to the Publisher."""
    if self.population is None or self.out is None or self.population is None or self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return [
            IntMessage(
                topic=EvaluatorMessageTopics.NEW_EVALUATIONS,
                value=self.new_evals,
                source="EMOEvaluator",
            )
        ]

    if isinstance(self.population, pl.DataFrame):
        population_message = PolarsDataFrameMessage(
            topic=EvaluatorMessageTopics.POPULATION,
            value=self.population,
            source="EMOEvaluator",
        )
    else:
        population_message = GenericMessage(
            topic=EvaluatorMessageTopics.POPULATION,
            value="Population is not a polars DataFrame",
            source="EMOEvaluator",
        )
    return [
        IntMessage(
            topic=EvaluatorMessageTopics.NEW_EVALUATIONS,
            value=self.new_evals,
            source="EMOEvaluator",
        ),
        population_message,
        PolarsDataFrameMessage(
            topic=EvaluatorMessageTopics.OUTPUTS,
            value=self.out,
            source="EMOEvaluator",
        ),
    ]
update
update(*_, **__)

Update the parameters of the evaluator.

Source code in desdeo/emo/operators/evaluator.py
def update(self, *_, **__):
    """Update the parameters of the evaluator."""

Crossover operators

desdeo.emo.operators.crossover

Evolutionary operators for recombination.

Various evolutionary operators for recombination in multiobjective optimization are defined here.

BaseCrossover

Bases: Subscriber

A base class for crossover operators.

Source code in desdeo/emo/operators/crossover.py
class BaseCrossover(Subscriber):
    """A base class for crossover operators."""

    def __init__(self, problem: Problem, **kwargs):
        """Initialize a crossover operator."""
        super().__init__(**kwargs)
        self.problem = problem
        self.variable_symbols = [var.symbol for var in problem.get_flattened_variables()]
        self.lower_bounds = [var.lowerbound for var in problem.get_flattened_variables()]
        self.upper_bounds = [var.upperbound for var in problem.get_flattened_variables()]

        self.variable_types = [var.variable_type for var in problem.get_flattened_variables()]
        self.variable_combination: VariableDomainTypeEnum = problem.variable_domain

    @abstractmethod
    def do(self, *, population: pl.DataFrame, to_mate: list[int] | None = None) -> pl.DataFrame:
        """Perform the crossover operation.

        Args:
            population (pl.DataFrame): the population to perform the crossover with. The DataFrame
                contains the decision vectors, the target vectors, and the constraint vectors.
            to_mate (list[int] | None): the indices of the population members that should
                participate in the crossover. If `None`, the whole population is subject
                to the crossover.

        Returns:
            pl.DataFrame: the offspring resulting from the crossover.
        """
__init__
__init__(problem: Problem, **kwargs)

Initialize a crossover operator.

Source code in desdeo/emo/operators/crossover.py
def __init__(self, problem: Problem, **kwargs):
    """Initialize a crossover operator."""
    super().__init__(**kwargs)
    self.problem = problem
    self.variable_symbols = [var.symbol for var in problem.get_flattened_variables()]
    self.lower_bounds = [var.lowerbound for var in problem.get_flattened_variables()]
    self.upper_bounds = [var.upperbound for var in problem.get_flattened_variables()]

    self.variable_types = [var.variable_type for var in problem.get_flattened_variables()]
    self.variable_combination: VariableDomainTypeEnum = problem.variable_domain
do abstractmethod
do(*, population: pl.DataFrame, to_mate: list[int] | None = None) -> pl.DataFrame

Perform the crossover operation.

Parameters:

Name Type Description Default
population DataFrame

the population to perform the crossover with. The DataFrame contains the decision vectors, the target vectors, and the constraint vectors.

required
to_mate list[int] | None

the indices of the population members that should participate in the crossover. If None, the whole population is subject to the crossover.

None

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring resulting from the crossover.

Source code in desdeo/emo/operators/crossover.py
@abstractmethod
def do(self, *, population: pl.DataFrame, to_mate: list[int] | None = None) -> pl.DataFrame:
    """Perform the crossover operation.

    Args:
        population (pl.DataFrame): the population to perform the crossover with. The DataFrame
            contains the decision vectors, the target vectors, and the constraint vectors.
        to_mate (list[int] | None): the indices of the population members that should
            participate in the crossover. If `None`, the whole population is subject
            to the crossover.

    Returns:
        pl.DataFrame: the offspring resulting from the crossover.
    """

SimulatedBinaryCrossover

Bases: BaseCrossover

A class for creating a simulated binary crossover operator.

Reference

Kalyanmoy Deb and Ram Bhushan Agrawal. 1995. Simulated binary crossover for continuous search space. Complex Systems 9, 2 (1995), 115-148.

Source code in desdeo/emo/operators/crossover.py
class SimulatedBinaryCrossover(BaseCrossover):
    """A class for creating a simulated binary crossover operator.

    Reference:
        Kalyanmoy Deb and Ram Bhushan Agrawal. 1995. Simulated binary crossover for continuous search space.
            Complex Systems 9, 2 (1995), 115-148.
    """

    @property
    def provided_topics(self) -> dict[str, Sequence[CrossoverMessageTopics]]:
        """The message topics provided by the crossover operator."""
        return {
            0: [],
            1: [CrossoverMessageTopics.XOVER_PROBABILITY, CrossoverMessageTopics.XOVER_DISTRIBUTION],
            2: [
                CrossoverMessageTopics.XOVER_PROBABILITY,
                CrossoverMessageTopics.XOVER_DISTRIBUTION,
                CrossoverMessageTopics.PARENTS,
                CrossoverMessageTopics.OFFSPRINGS,
            ],
        }

    @property
    def interested_topics(self):
        """The message topics the crossover operator is interested in."""
        return []

    def __init__(
        self, *, problem: Problem, seed: int, xover_probability: float = 1.0, xover_distribution: float = 30, **kwargs
    ):
        """Initialize a simulated binary crossover operator.

        Args:
            problem (Problem): the problem object.
            seed (int): the seed for the random number generator.
            xover_probability (float, optional): the crossover probability
                parameter. Ranges between 0 and 1.0. Defaults to 1.0.
            xover_distribution (float, optional): the crossover distribution
                parameter. Must be positive. Defaults to 30.
            kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
                publisher must be passed. See the Subscriber class for more information.
        """
        # Subscribes to no topics, so no need to stroe/pass the topics to the super class.
        super().__init__(problem, **kwargs)
        self.problem = problem

        if not 0 <= xover_probability <= 1:
            raise ValueError("Crossover probability must be between 0 and 1.")
        if xover_distribution <= 0:
            raise ValueError("Crossover distribution must be positive.")
        self.xover_probability = xover_probability
        self.xover_distribution = xover_distribution
        self.parent_population: pl.DataFrame
        self.offspring_population: pl.DataFrame
        self.rng = np.random.default_rng(seed)
        self.seed = seed

    def do(
        self,
        *,
        population: pl.DataFrame,
        to_mate: list[int] | None = None,
    ) -> pl.DataFrame:
        """Perform the simulated binary crossover operation.

        Args:
            population (pl.DataFrame): the population to perform the crossover with. The DataFrame
                contains the decision vectors, the target vectors, and the constraint vectors.
            to_mate (list[int] | None): the indices of the population members that should
                participate in the crossover. If `None`, the whole population is subject
                to the crossover.

        Returns:
            pl.DataFrame: the offspring resulting from the crossover.
        """
        self.parent_population = population
        pop_size = self.parent_population.shape[0]
        num_var = len(self.variable_symbols)

        parent_decvars = self.parent_population[self.variable_symbols].to_numpy()

        if to_mate is None:
            shuffled_ids = list(range(pop_size))
            shuffle(shuffled_ids)
        else:
            shuffled_ids = to_mate
        mating_pop = parent_decvars[shuffled_ids]
        mate_size = len(shuffled_ids)

        if len(shuffled_ids) % 2 == 1:
            mating_pop = np.vstack((mating_pop, mating_pop[0]))
            mate_size += 1

        offspring = np.zeros_like(mating_pop)

        HALF = 0.5  # NOQA: N806
        # TODO(@light-weaver): Extract into a numba jitted function.
        for i in range(0, mate_size, 2):
            beta = np.zeros(num_var)
            miu = self.rng.random(num_var)
            beta[miu <= HALF] = (2 * miu[miu <= HALF]) ** (1 / (self.xover_distribution + 1))
            beta[miu > HALF] = (2 - 2 * miu[miu > HALF]) ** (-1 / (self.xover_distribution + 1))
            beta = beta * ((-1) ** self.rng.integers(low=0, high=2, size=num_var))
            beta[self.rng.random(num_var) > self.xover_probability] = 1
            avg = (mating_pop[i] + mating_pop[i + 1]) / 2
            diff = (mating_pop[i] - mating_pop[i + 1]) / 2
            offspring[i] = avg + beta * diff
            offspring[i + 1] = avg - beta * diff

        self.offspring_population = pl.from_numpy(offspring, schema=self.variable_symbols)
        self.notify()

        return self.offspring_population

    def update(self, *_, **__):
        """Do nothing. This is just the basic SBX operator."""

    def state(self) -> Sequence[Message]:
        """Return the state of the crossover operator."""
        if self.parent_population is None or self.offspring_population is None:
            return []
        if self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return [
                FloatMessage(
                    topic=CrossoverMessageTopics.XOVER_PROBABILITY,
                    source="SimulatedBinaryCrossover",
                    value=self.xover_probability,
                ),
                FloatMessage(
                    topic=CrossoverMessageTopics.XOVER_DISTRIBUTION,
                    source="SimulatedBinaryCrossover",
                    value=self.xover_distribution,
                ),
            ]
        # verbosity == 2 or higher
        return [
            FloatMessage(
                topic=CrossoverMessageTopics.XOVER_PROBABILITY,
                source="SimulatedBinaryCrossover",
                value=self.xover_probability,
            ),
            FloatMessage(
                topic=CrossoverMessageTopics.XOVER_DISTRIBUTION,
                source="SimulatedBinaryCrossover",
                value=self.xover_distribution,
            ),
            PolarsDataFrameMessage(
                topic=CrossoverMessageTopics.PARENTS,
                source="SimulatedBinaryCrossover",
                value=self.parent_population,
            ),
            PolarsDataFrameMessage(
                topic=CrossoverMessageTopics.OFFSPRINGS,
                source="SimulatedBinaryCrossover",
                value=self.offspring_population,
            ),
        ]
interested_topics property
interested_topics

The message topics the crossover operator is interested in.

provided_topics property
provided_topics: dict[str, Sequence[CrossoverMessageTopics]]

The message topics provided by the crossover operator.

__init__
__init__(*, problem: Problem, seed: int, xover_probability: float = 1.0, xover_distribution: float = 30, **kwargs)

Initialize a simulated binary crossover operator.

Parameters:

Name Type Description Default
problem Problem

the problem object.

required
seed int

the seed for the random number generator.

required
xover_probability float

the crossover probability parameter. Ranges between 0 and 1.0. Defaults to 1.0.

1.0
xover_distribution float

the crossover distribution parameter. Must be positive. Defaults to 30.

30
kwargs

Additional keyword arguments. These are passed to the Subscriber class. At the very least, the publisher must be passed. See the Subscriber class for more information.

{}
Source code in desdeo/emo/operators/crossover.py
def __init__(
    self, *, problem: Problem, seed: int, xover_probability: float = 1.0, xover_distribution: float = 30, **kwargs
):
    """Initialize a simulated binary crossover operator.

    Args:
        problem (Problem): the problem object.
        seed (int): the seed for the random number generator.
        xover_probability (float, optional): the crossover probability
            parameter. Ranges between 0 and 1.0. Defaults to 1.0.
        xover_distribution (float, optional): the crossover distribution
            parameter. Must be positive. Defaults to 30.
        kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
            publisher must be passed. See the Subscriber class for more information.
    """
    # Subscribes to no topics, so no need to stroe/pass the topics to the super class.
    super().__init__(problem, **kwargs)
    self.problem = problem

    if not 0 <= xover_probability <= 1:
        raise ValueError("Crossover probability must be between 0 and 1.")
    if xover_distribution <= 0:
        raise ValueError("Crossover distribution must be positive.")
    self.xover_probability = xover_probability
    self.xover_distribution = xover_distribution
    self.parent_population: pl.DataFrame
    self.offspring_population: pl.DataFrame
    self.rng = np.random.default_rng(seed)
    self.seed = seed
do
do(*, population: pl.DataFrame, to_mate: list[int] | None = None) -> pl.DataFrame

Perform the simulated binary crossover operation.

Parameters:

Name Type Description Default
population DataFrame

the population to perform the crossover with. The DataFrame contains the decision vectors, the target vectors, and the constraint vectors.

required
to_mate list[int] | None

the indices of the population members that should participate in the crossover. If None, the whole population is subject to the crossover.

None

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring resulting from the crossover.

Source code in desdeo/emo/operators/crossover.py
def do(
    self,
    *,
    population: pl.DataFrame,
    to_mate: list[int] | None = None,
) -> pl.DataFrame:
    """Perform the simulated binary crossover operation.

    Args:
        population (pl.DataFrame): the population to perform the crossover with. The DataFrame
            contains the decision vectors, the target vectors, and the constraint vectors.
        to_mate (list[int] | None): the indices of the population members that should
            participate in the crossover. If `None`, the whole population is subject
            to the crossover.

    Returns:
        pl.DataFrame: the offspring resulting from the crossover.
    """
    self.parent_population = population
    pop_size = self.parent_population.shape[0]
    num_var = len(self.variable_symbols)

    parent_decvars = self.parent_population[self.variable_symbols].to_numpy()

    if to_mate is None:
        shuffled_ids = list(range(pop_size))
        shuffle(shuffled_ids)
    else:
        shuffled_ids = to_mate
    mating_pop = parent_decvars[shuffled_ids]
    mate_size = len(shuffled_ids)

    if len(shuffled_ids) % 2 == 1:
        mating_pop = np.vstack((mating_pop, mating_pop[0]))
        mate_size += 1

    offspring = np.zeros_like(mating_pop)

    HALF = 0.5  # NOQA: N806
    # TODO(@light-weaver): Extract into a numba jitted function.
    for i in range(0, mate_size, 2):
        beta = np.zeros(num_var)
        miu = self.rng.random(num_var)
        beta[miu <= HALF] = (2 * miu[miu <= HALF]) ** (1 / (self.xover_distribution + 1))
        beta[miu > HALF] = (2 - 2 * miu[miu > HALF]) ** (-1 / (self.xover_distribution + 1))
        beta = beta * ((-1) ** self.rng.integers(low=0, high=2, size=num_var))
        beta[self.rng.random(num_var) > self.xover_probability] = 1
        avg = (mating_pop[i] + mating_pop[i + 1]) / 2
        diff = (mating_pop[i] - mating_pop[i + 1]) / 2
        offspring[i] = avg + beta * diff
        offspring[i + 1] = avg - beta * diff

    self.offspring_population = pl.from_numpy(offspring, schema=self.variable_symbols)
    self.notify()

    return self.offspring_population
state
state() -> Sequence[Message]

Return the state of the crossover operator.

Source code in desdeo/emo/operators/crossover.py
def state(self) -> Sequence[Message]:
    """Return the state of the crossover operator."""
    if self.parent_population is None or self.offspring_population is None:
        return []
    if self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return [
            FloatMessage(
                topic=CrossoverMessageTopics.XOVER_PROBABILITY,
                source="SimulatedBinaryCrossover",
                value=self.xover_probability,
            ),
            FloatMessage(
                topic=CrossoverMessageTopics.XOVER_DISTRIBUTION,
                source="SimulatedBinaryCrossover",
                value=self.xover_distribution,
            ),
        ]
    # verbosity == 2 or higher
    return [
        FloatMessage(
            topic=CrossoverMessageTopics.XOVER_PROBABILITY,
            source="SimulatedBinaryCrossover",
            value=self.xover_probability,
        ),
        FloatMessage(
            topic=CrossoverMessageTopics.XOVER_DISTRIBUTION,
            source="SimulatedBinaryCrossover",
            value=self.xover_distribution,
        ),
        PolarsDataFrameMessage(
            topic=CrossoverMessageTopics.PARENTS,
            source="SimulatedBinaryCrossover",
            value=self.parent_population,
        ),
        PolarsDataFrameMessage(
            topic=CrossoverMessageTopics.OFFSPRINGS,
            source="SimulatedBinaryCrossover",
            value=self.offspring_population,
        ),
    ]
update
update(*_, **__)

Do nothing. This is just the basic SBX operator.

Source code in desdeo/emo/operators/crossover.py
def update(self, *_, **__):
    """Do nothing. This is just the basic SBX operator."""

SinglePointBinaryCrossover

Bases: BaseCrossover

A class that defines the single point binary crossover operation.

Source code in desdeo/emo/operators/crossover.py
class SinglePointBinaryCrossover(BaseCrossover):
    """A class that defines the single point binary crossover operation."""

    def __init__(self, *, problem: Problem, seed: int, **kwargs):
        """Initialize the single point binary crossover operator.

        Args:
            problem (Problem): the problem object.
            seed (int): the seed used in the random number generator for choosing the crossover point.
            kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
                publisher must be passed. See the Subscriber class for more information.
        """
        super().__init__(problem, **kwargs)
        self.seed = seed

        self.parent_population: pl.DataFrame
        self.offspring_population: pl.DataFrame
        self.rng = np.random.default_rng(seed)
        self.seed = seed

    @property
    def provided_topics(self) -> dict[str, Sequence[CrossoverMessageTopics]]:
        """The message topics provided by the single point binary crossover operator."""
        return {
            0: [],
            1: [],
            2: [
                CrossoverMessageTopics.PARENTS,
                CrossoverMessageTopics.OFFSPRINGS,
            ],
        }

    @property
    def interested_topics(self):
        """The message topics the single point binary crossover operator is interested in."""
        return []

    def do(
        self,
        *,
        population: pl.DataFrame,
        to_mate: list[int] | None = None,
    ) -> pl.DataFrame:
        """Perform single point binary crossover.

        Args:
            population (pl.DataFrame): the population to perform the crossover with.
            to_mate (list[int] | None, optional): indices. Defaults to None.

        Returns:
            pl.DataFrame: the offspring from the crossover.
        """
        self.parent_population = population
        pop_size = self.parent_population.shape[0]
        num_var = len(self.variable_symbols)

        parent_decision_vars = self.parent_population[self.variable_symbols].to_numpy().astype(np.bool)

        if to_mate is None:
            shuffled_ids = list(range(pop_size))
            shuffle(shuffled_ids)
        else:
            shuffled_ids = copy.copy(to_mate)

        mating_pop = parent_decision_vars[shuffled_ids]
        mating_pop_size = len(shuffled_ids)
        original_mating_pop_size = mating_pop_size

        if mating_pop_size % 2 != 0:
            # if the number of member to mate is of uneven size, copy the first member to the tail
            mating_pop = np.vstack((mating_pop, mating_pop[0]))
            mating_pop_size += 1
            shuffled_ids.append(shuffled_ids[0])

        # split the population into parents, one with members with even numbered indices, the
        # other with uneven numbered indices
        parents1 = mating_pop[[shuffled_ids[i] for i in range(0, mating_pop_size, 2)]]
        parents2 = mating_pop[[shuffled_ids[i] for i in range(1, mating_pop_size, 2)]]

        cross_over_points = self.rng.integers(1, num_var - 1, mating_pop_size // 2)

        # create a mask where, on each row, the element is 1 before the crossover point,
        # and zero after it
        cross_over_mask = np.zeros_like(parents1, dtype=np.bool)
        cross_over_mask[np.arange(cross_over_mask.shape[1]) < cross_over_points[:, None]] = 1

        # pick genes from the first parents before the crossover point
        # pick genes from the second parents after, and including, the crossover point
        offspring1_first = cross_over_mask & parents1
        offspring1_second = (~cross_over_mask) & parents2

        # combine into a first half of the whole offspring population
        offspring1 = offspring1_first | offspring1_second

        # pick genes from the first parents after, and including, the crossover point
        # pick genes from the second parents before the crossover point
        offspring2_first = (~cross_over_mask) & parents1
        offspring2_second = cross_over_mask & parents2

        # combine into the second half of the whole offspring population
        offspring2 = offspring2_first | offspring2_second

        # combine the two offspring populations into one, drop the last member if the number of
        # indices (to_mate) is uneven
        self.offspring_population = pl.from_numpy(
            np.vstack((offspring1, offspring2))[
                : (original_mating_pop_size if original_mating_pop_size % 2 == 0 else -1)
            ],
            schema=self.variable_symbols,
        ).select(pl.all().cast(pl.Float64))
        self.notify()

        return self.offspring_population

    def update(self, *_, **__):
        """Do nothing. This is just the basic single point binary crossover operator."""

    def state(self) -> Sequence[Message]:
        """Return the state of the single ponit binary crossover operator."""
        if self.parent_population is None or self.offspring_population is None:
            return []
        if self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return []
        # verbosity == 2 or higher
        return [
            PolarsDataFrameMessage(
                topic=CrossoverMessageTopics.PARENTS,
                source="SimulatedBinaryCrossover",
                value=self.parent_population,
            ),
            PolarsDataFrameMessage(
                topic=CrossoverMessageTopics.OFFSPRINGS,
                source="SimulatedBinaryCrossover",
                value=self.offspring_population,
            ),
        ]
interested_topics property
interested_topics

The message topics the single point binary crossover operator is interested in.

provided_topics property
provided_topics: dict[str, Sequence[CrossoverMessageTopics]]

The message topics provided by the single point binary crossover operator.

__init__
__init__(*, problem: Problem, seed: int, **kwargs)

Initialize the single point binary crossover operator.

Parameters:

Name Type Description Default
problem Problem

the problem object.

required
seed int

the seed used in the random number generator for choosing the crossover point.

required
kwargs

Additional keyword arguments. These are passed to the Subscriber class. At the very least, the publisher must be passed. See the Subscriber class for more information.

{}
Source code in desdeo/emo/operators/crossover.py
def __init__(self, *, problem: Problem, seed: int, **kwargs):
    """Initialize the single point binary crossover operator.

    Args:
        problem (Problem): the problem object.
        seed (int): the seed used in the random number generator for choosing the crossover point.
        kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
            publisher must be passed. See the Subscriber class for more information.
    """
    super().__init__(problem, **kwargs)
    self.seed = seed

    self.parent_population: pl.DataFrame
    self.offspring_population: pl.DataFrame
    self.rng = np.random.default_rng(seed)
    self.seed = seed
do
do(*, population: pl.DataFrame, to_mate: list[int] | None = None) -> pl.DataFrame

Perform single point binary crossover.

Parameters:

Name Type Description Default
population DataFrame

the population to perform the crossover with.

required
to_mate list[int] | None

indices. Defaults to None.

None

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring from the crossover.

Source code in desdeo/emo/operators/crossover.py
def do(
    self,
    *,
    population: pl.DataFrame,
    to_mate: list[int] | None = None,
) -> pl.DataFrame:
    """Perform single point binary crossover.

    Args:
        population (pl.DataFrame): the population to perform the crossover with.
        to_mate (list[int] | None, optional): indices. Defaults to None.

    Returns:
        pl.DataFrame: the offspring from the crossover.
    """
    self.parent_population = population
    pop_size = self.parent_population.shape[0]
    num_var = len(self.variable_symbols)

    parent_decision_vars = self.parent_population[self.variable_symbols].to_numpy().astype(np.bool)

    if to_mate is None:
        shuffled_ids = list(range(pop_size))
        shuffle(shuffled_ids)
    else:
        shuffled_ids = copy.copy(to_mate)

    mating_pop = parent_decision_vars[shuffled_ids]
    mating_pop_size = len(shuffled_ids)
    original_mating_pop_size = mating_pop_size

    if mating_pop_size % 2 != 0:
        # if the number of member to mate is of uneven size, copy the first member to the tail
        mating_pop = np.vstack((mating_pop, mating_pop[0]))
        mating_pop_size += 1
        shuffled_ids.append(shuffled_ids[0])

    # split the population into parents, one with members with even numbered indices, the
    # other with uneven numbered indices
    parents1 = mating_pop[[shuffled_ids[i] for i in range(0, mating_pop_size, 2)]]
    parents2 = mating_pop[[shuffled_ids[i] for i in range(1, mating_pop_size, 2)]]

    cross_over_points = self.rng.integers(1, num_var - 1, mating_pop_size // 2)

    # create a mask where, on each row, the element is 1 before the crossover point,
    # and zero after it
    cross_over_mask = np.zeros_like(parents1, dtype=np.bool)
    cross_over_mask[np.arange(cross_over_mask.shape[1]) < cross_over_points[:, None]] = 1

    # pick genes from the first parents before the crossover point
    # pick genes from the second parents after, and including, the crossover point
    offspring1_first = cross_over_mask & parents1
    offspring1_second = (~cross_over_mask) & parents2

    # combine into a first half of the whole offspring population
    offspring1 = offspring1_first | offspring1_second

    # pick genes from the first parents after, and including, the crossover point
    # pick genes from the second parents before the crossover point
    offspring2_first = (~cross_over_mask) & parents1
    offspring2_second = cross_over_mask & parents2

    # combine into the second half of the whole offspring population
    offspring2 = offspring2_first | offspring2_second

    # combine the two offspring populations into one, drop the last member if the number of
    # indices (to_mate) is uneven
    self.offspring_population = pl.from_numpy(
        np.vstack((offspring1, offspring2))[
            : (original_mating_pop_size if original_mating_pop_size % 2 == 0 else -1)
        ],
        schema=self.variable_symbols,
    ).select(pl.all().cast(pl.Float64))
    self.notify()

    return self.offspring_population
state
state() -> Sequence[Message]

Return the state of the single ponit binary crossover operator.

Source code in desdeo/emo/operators/crossover.py
def state(self) -> Sequence[Message]:
    """Return the state of the single ponit binary crossover operator."""
    if self.parent_population is None or self.offspring_population is None:
        return []
    if self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return []
    # verbosity == 2 or higher
    return [
        PolarsDataFrameMessage(
            topic=CrossoverMessageTopics.PARENTS,
            source="SimulatedBinaryCrossover",
            value=self.parent_population,
        ),
        PolarsDataFrameMessage(
            topic=CrossoverMessageTopics.OFFSPRINGS,
            source="SimulatedBinaryCrossover",
            value=self.offspring_population,
        ),
    ]
update
update(*_, **__)

Do nothing. This is just the basic single point binary crossover operator.

Source code in desdeo/emo/operators/crossover.py
def update(self, *_, **__):
    """Do nothing. This is just the basic single point binary crossover operator."""

UniformIntegerCrossover

Bases: BaseCrossover

A class that defines the uniform integer crossover operation.

Source code in desdeo/emo/operators/crossover.py
class UniformIntegerCrossover(BaseCrossover):
    """A class that defines the uniform integer crossover operation."""

    def __init__(self, *, problem: Problem, seed: int, **kwargs):
        """Initialize the uniform integer crossover operator.

        Args:
            problem (Problem): the problem object.
            seed (int): the seed used in the random number generator for choosing the crossover point.
            kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
                publisher must be passed. See the Subscriber class for more information.
        """
        super().__init__(problem, **kwargs)
        self.seed = seed

        self.parent_population: pl.DataFrame
        self.offspring_population: pl.DataFrame
        self.rng = np.random.default_rng(seed)
        self.seed = seed

    @property
    def provided_topics(self) -> dict[str, Sequence[CrossoverMessageTopics]]:
        """The message topics provided by the single point binary crossover operator."""
        return {
            0: [],
            1: [],
            2: [
                CrossoverMessageTopics.PARENTS,
                CrossoverMessageTopics.OFFSPRINGS,
            ],
        }

    @property
    def interested_topics(self):
        """The message topics the single point binary crossover operator is interested in."""
        return []

    def do(
        self,
        *,
        population: pl.DataFrame,
        to_mate: list[int] | None = None,
    ) -> pl.DataFrame:
        """Perform single point binary crossover.

        Args:
            population (pl.DataFrame): the population to perform the crossover with.
            to_mate (list[int] | None, optional): indices. Defaults to None.

        Returns:
            pl.DataFrame: the offspring from the crossover.
        """
        self.parent_population = population
        pop_size = self.parent_population.shape[0]
        num_var = len(self.variable_symbols)

        parent_decision_vars = self.parent_population[self.variable_symbols].to_numpy().astype(int)

        if to_mate is None:
            shuffled_ids = list(range(pop_size))
            shuffle(shuffled_ids)
        else:
            shuffled_ids = copy.copy(to_mate)

        mating_pop = parent_decision_vars[shuffled_ids]
        mating_pop_size = len(shuffled_ids)
        original_mating_pop_size = mating_pop_size

        if mating_pop_size % 2 != 0:
            # if the number of member to mate is of uneven size, copy the first member to the tail
            mating_pop = np.vstack((mating_pop, mating_pop[0]))
            mating_pop_size += 1
            shuffled_ids.append(shuffled_ids[0])

        # split the population into parents, one with members with even numbered indices, the
        # other with uneven numbered indices
        parents1 = mating_pop[[shuffled_ids[i] for i in range(0, mating_pop_size, 2)]]
        parents2 = mating_pop[[shuffled_ids[i] for i in range(1, mating_pop_size, 2)]]

        mask = self.rng.choice([True, False], size=num_var)

        offspring1 = np.where(mask, parents1, parents2)  # True, pick from parent1, False, pick from parent2
        offspring2 = np.where(mask, parents2, parents1)  # True, pick from parent2, False, pick from parent1

        # combine the two offspring populations into one, drop the last member if the number of
        # indices (to_mate) is uneven
        self.offspring_population = pl.from_numpy(
            np.vstack((offspring1, offspring2))[
                : (original_mating_pop_size if original_mating_pop_size % 2 == 0 else -1)
            ],
            schema=self.variable_symbols,
        ).select(pl.all().cast(pl.Float64))

        self.notify()

        return self.offspring_population

    def update(self, *_, **__):
        """Do nothing. This is just the basic single point binary crossover operator."""

    def state(self) -> Sequence[Message]:
        """Return the state of the single ponit binary crossover operator."""
        if self.parent_population is None or self.offspring_population is None:
            return []
        if self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return []
        # verbosity == 2 or higher
        return [
            PolarsDataFrameMessage(
                topic=CrossoverMessageTopics.PARENTS,
                source="SimulatedBinaryCrossover",
                value=self.parent_population,
            ),
            PolarsDataFrameMessage(
                topic=CrossoverMessageTopics.OFFSPRINGS,
                source="SimulatedBinaryCrossover",
                value=self.offspring_population,
            ),
        ]
interested_topics property
interested_topics

The message topics the single point binary crossover operator is interested in.

provided_topics property
provided_topics: dict[str, Sequence[CrossoverMessageTopics]]

The message topics provided by the single point binary crossover operator.

__init__
__init__(*, problem: Problem, seed: int, **kwargs)

Initialize the uniform integer crossover operator.

Parameters:

Name Type Description Default
problem Problem

the problem object.

required
seed int

the seed used in the random number generator for choosing the crossover point.

required
kwargs

Additional keyword arguments. These are passed to the Subscriber class. At the very least, the publisher must be passed. See the Subscriber class for more information.

{}
Source code in desdeo/emo/operators/crossover.py
def __init__(self, *, problem: Problem, seed: int, **kwargs):
    """Initialize the uniform integer crossover operator.

    Args:
        problem (Problem): the problem object.
        seed (int): the seed used in the random number generator for choosing the crossover point.
        kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
            publisher must be passed. See the Subscriber class for more information.
    """
    super().__init__(problem, **kwargs)
    self.seed = seed

    self.parent_population: pl.DataFrame
    self.offspring_population: pl.DataFrame
    self.rng = np.random.default_rng(seed)
    self.seed = seed
do
do(*, population: pl.DataFrame, to_mate: list[int] | None = None) -> pl.DataFrame

Perform single point binary crossover.

Parameters:

Name Type Description Default
population DataFrame

the population to perform the crossover with.

required
to_mate list[int] | None

indices. Defaults to None.

None

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring from the crossover.

Source code in desdeo/emo/operators/crossover.py
def do(
    self,
    *,
    population: pl.DataFrame,
    to_mate: list[int] | None = None,
) -> pl.DataFrame:
    """Perform single point binary crossover.

    Args:
        population (pl.DataFrame): the population to perform the crossover with.
        to_mate (list[int] | None, optional): indices. Defaults to None.

    Returns:
        pl.DataFrame: the offspring from the crossover.
    """
    self.parent_population = population
    pop_size = self.parent_population.shape[0]
    num_var = len(self.variable_symbols)

    parent_decision_vars = self.parent_population[self.variable_symbols].to_numpy().astype(int)

    if to_mate is None:
        shuffled_ids = list(range(pop_size))
        shuffle(shuffled_ids)
    else:
        shuffled_ids = copy.copy(to_mate)

    mating_pop = parent_decision_vars[shuffled_ids]
    mating_pop_size = len(shuffled_ids)
    original_mating_pop_size = mating_pop_size

    if mating_pop_size % 2 != 0:
        # if the number of member to mate is of uneven size, copy the first member to the tail
        mating_pop = np.vstack((mating_pop, mating_pop[0]))
        mating_pop_size += 1
        shuffled_ids.append(shuffled_ids[0])

    # split the population into parents, one with members with even numbered indices, the
    # other with uneven numbered indices
    parents1 = mating_pop[[shuffled_ids[i] for i in range(0, mating_pop_size, 2)]]
    parents2 = mating_pop[[shuffled_ids[i] for i in range(1, mating_pop_size, 2)]]

    mask = self.rng.choice([True, False], size=num_var)

    offspring1 = np.where(mask, parents1, parents2)  # True, pick from parent1, False, pick from parent2
    offspring2 = np.where(mask, parents2, parents1)  # True, pick from parent2, False, pick from parent1

    # combine the two offspring populations into one, drop the last member if the number of
    # indices (to_mate) is uneven
    self.offspring_population = pl.from_numpy(
        np.vstack((offspring1, offspring2))[
            : (original_mating_pop_size if original_mating_pop_size % 2 == 0 else -1)
        ],
        schema=self.variable_symbols,
    ).select(pl.all().cast(pl.Float64))

    self.notify()

    return self.offspring_population
state
state() -> Sequence[Message]

Return the state of the single ponit binary crossover operator.

Source code in desdeo/emo/operators/crossover.py
def state(self) -> Sequence[Message]:
    """Return the state of the single ponit binary crossover operator."""
    if self.parent_population is None or self.offspring_population is None:
        return []
    if self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return []
    # verbosity == 2 or higher
    return [
        PolarsDataFrameMessage(
            topic=CrossoverMessageTopics.PARENTS,
            source="SimulatedBinaryCrossover",
            value=self.parent_population,
        ),
        PolarsDataFrameMessage(
            topic=CrossoverMessageTopics.OFFSPRINGS,
            source="SimulatedBinaryCrossover",
            value=self.offspring_population,
        ),
    ]
update
update(*_, **__)

Do nothing. This is just the basic single point binary crossover operator.

Source code in desdeo/emo/operators/crossover.py
def update(self, *_, **__):
    """Do nothing. This is just the basic single point binary crossover operator."""

Mutation operators

desdeo.emo.operators.mutation

Evolutionary operators for mutation.

Various evolutionary operators for mutation in multiobjective optimization are defined here.

BaseMutation

Bases: Subscriber

A base class for mutation operators.

Source code in desdeo/emo/operators/mutation.py
class BaseMutation(Subscriber):
    """A base class for mutation operators."""

    @abstractmethod
    def __init__(self, problem: Problem, **kwargs):
        """Initialize a mu operator."""
        super().__init__(**kwargs)
        self.problem = problem
        self.variable_symbols = [var.symbol for var in problem.get_flattened_variables()]
        self.lower_bounds = [var.lowerbound for var in problem.get_flattened_variables()]
        self.upper_bounds = [var.upperbound for var in problem.get_flattened_variables()]
        self.variable_types = [var.variable_type for var in problem.get_flattened_variables()]
        self.variable_combination: VariableDomainTypeEnum = problem.variable_domain

    @abstractmethod
    def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
        """Perform the mutation operation.

        Args:
            offsprings (pl.DataFrame): the offspring population to mutate.
            parents (pl.DataFrame): the parent population from which the offspring
                was generated (via crossover).

        Returns:
            pl.DataFrame: the offspring resulting from the mutation.
        """
__init__ abstractmethod
__init__(problem: Problem, **kwargs)

Initialize a mu operator.

Source code in desdeo/emo/operators/mutation.py
@abstractmethod
def __init__(self, problem: Problem, **kwargs):
    """Initialize a mu operator."""
    super().__init__(**kwargs)
    self.problem = problem
    self.variable_symbols = [var.symbol for var in problem.get_flattened_variables()]
    self.lower_bounds = [var.lowerbound for var in problem.get_flattened_variables()]
    self.upper_bounds = [var.upperbound for var in problem.get_flattened_variables()]
    self.variable_types = [var.variable_type for var in problem.get_flattened_variables()]
    self.variable_combination: VariableDomainTypeEnum = problem.variable_domain
do abstractmethod
do(offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame

Perform the mutation operation.

Parameters:

Name Type Description Default
offsprings DataFrame

the offspring population to mutate.

required
parents DataFrame

the parent population from which the offspring was generated (via crossover).

required

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring resulting from the mutation.

Source code in desdeo/emo/operators/mutation.py
@abstractmethod
def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
    """Perform the mutation operation.

    Args:
        offsprings (pl.DataFrame): the offspring population to mutate.
        parents (pl.DataFrame): the parent population from which the offspring
            was generated (via crossover).

    Returns:
        pl.DataFrame: the offspring resulting from the mutation.
    """

BinaryFlipMutation

Bases: BaseMutation

Implements the bit flip mutation operator for binary variables.

The binary flip mutation will mutate each binary decision variable, by flipping it (0 to 1, 1 to 0) with a provided probability.

Source code in desdeo/emo/operators/mutation.py
class BinaryFlipMutation(BaseMutation):
    """Implements the bit flip mutation operator for binary variables.

    The binary flip mutation will mutate each binary decision variable,
    by flipping it (0 to 1, 1 to 0) with a provided probability.
    """

    @property
    def provided_topics(self) -> dict[int, Sequence[MutationMessageTopics]]:
        """The message topics provided by the mutation operator."""
        return {
            0: [],
            1: [
                MutationMessageTopics.MUTATION_PROBABILITY,
            ],
            2: [
                MutationMessageTopics.MUTATION_PROBABILITY,
                MutationMessageTopics.OFFSPRING_ORIGINAL,
                MutationMessageTopics.OFFSPRINGS,
            ],
        }

    @property
    def interested_topics(self):
        """The message topics that the mutation operator is interested in."""
        return []

    def __init__(
        self,
        *,
        problem: Problem,
        seed: int,
        mutation_probability: float | None = None,
        **kwargs,
    ):
        """Initialize a binary flip mutation operator.

        Args:
            problem (Problem): The problem object.
            seed (int): The seed for the random number generator.
            mutation_probability (float | None, optional): The probability of mutation. If None,
                the probability will be set to be 1/n, where n is the number of decision variables
                in the problem. Defaults to None.
            kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
                publisher must be passed. See the Subscriber class for more information.
        """
        super().__init__(problem, **kwargs)

        if self.variable_combination != VariableDomainTypeEnum.binary:
            raise ValueError("This mutation operator only works with binary variables.")
        if mutation_probability is None:
            self.mutation_probability = 1 / len(self.variable_symbols)
        else:
            self.mutation_probability = mutation_probability

        self.rng = np.random.default_rng(seed)
        self.seed = seed
        self.offspring_original: pl.DataFrame
        self.parents: pl.DataFrame
        self.offspring: pl.DataFrame

    def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
        """Perform the binary flip mutation operation.

        Args:
            offsprings (pl.DataFrame): the offspring population to mutate.
            parents (pl.DataFrame): the parent population from which the offspring
                was generated (via crossover). Not used in the mutation operator.

        Returns:
            pl.DataFrame: the offspring resulting from the mutation.
        """
        self.offspring_original = copy.copy(offsprings)
        self.parents = parents  # Not used, but kept for consistency
        offspring = offsprings.to_numpy().astype(dtype=np.bool)

        # create a boolean mask based on the mutation probability
        flip_mask = self.rng.random(offspring.shape) < self.mutation_probability

        # using XOR (^), flip the bits in the offspring when the mask is True
        # otherwise leave the bit's value as it is
        offspring = offspring ^ flip_mask

        self.offspring = pl.from_numpy(offspring, schema=self.variable_symbols).select(pl.all()).cast(pl.Float64)
        self.notify()

        return self.offspring

    def update(self, *_, **__):
        """Do nothing."""

    def state(self) -> Sequence[Message]:
        """Return the state of the mutation operator."""
        if self.offspring_original is None or self.offspring is None:
            return []
        if self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return [
                FloatMessage(
                    topic=MutationMessageTopics.MUTATION_PROBABILITY,
                    source=self.__class__.__name__,
                    value=self.mutation_probability,
                ),
            ]
        # verbosity == 2
        return [
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.OFFSPRING_ORIGINAL,
                source=self.__class__.__name__,
                value=self.offspring_original,
            ),
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.PARENTS,
                source=self.__class__.__name__,
                value=self.parents,
            ),
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.OFFSPRINGS,
                source=self.__class__.__name__,
                value=self.offspring,
            ),
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_PROBABILITY,
                source=self.__class__.__name__,
                value=self.mutation_probability,
            ),
        ]
interested_topics property
interested_topics

The message topics that the mutation operator is interested in.

provided_topics property
provided_topics: dict[int, Sequence[MutationMessageTopics]]

The message topics provided by the mutation operator.

__init__
__init__(*, problem: Problem, seed: int, mutation_probability: float | None = None, **kwargs)

Initialize a binary flip mutation operator.

Parameters:

Name Type Description Default
problem Problem

The problem object.

required
seed int

The seed for the random number generator.

required
mutation_probability float | None

The probability of mutation. If None, the probability will be set to be 1/n, where n is the number of decision variables in the problem. Defaults to None.

None
kwargs

Additional keyword arguments. These are passed to the Subscriber class. At the very least, the publisher must be passed. See the Subscriber class for more information.

{}
Source code in desdeo/emo/operators/mutation.py
def __init__(
    self,
    *,
    problem: Problem,
    seed: int,
    mutation_probability: float | None = None,
    **kwargs,
):
    """Initialize a binary flip mutation operator.

    Args:
        problem (Problem): The problem object.
        seed (int): The seed for the random number generator.
        mutation_probability (float | None, optional): The probability of mutation. If None,
            the probability will be set to be 1/n, where n is the number of decision variables
            in the problem. Defaults to None.
        kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
            publisher must be passed. See the Subscriber class for more information.
    """
    super().__init__(problem, **kwargs)

    if self.variable_combination != VariableDomainTypeEnum.binary:
        raise ValueError("This mutation operator only works with binary variables.")
    if mutation_probability is None:
        self.mutation_probability = 1 / len(self.variable_symbols)
    else:
        self.mutation_probability = mutation_probability

    self.rng = np.random.default_rng(seed)
    self.seed = seed
    self.offspring_original: pl.DataFrame
    self.parents: pl.DataFrame
    self.offspring: pl.DataFrame
do
do(offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame

Perform the binary flip mutation operation.

Parameters:

Name Type Description Default
offsprings DataFrame

the offspring population to mutate.

required
parents DataFrame

the parent population from which the offspring was generated (via crossover). Not used in the mutation operator.

required

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring resulting from the mutation.

Source code in desdeo/emo/operators/mutation.py
def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
    """Perform the binary flip mutation operation.

    Args:
        offsprings (pl.DataFrame): the offspring population to mutate.
        parents (pl.DataFrame): the parent population from which the offspring
            was generated (via crossover). Not used in the mutation operator.

    Returns:
        pl.DataFrame: the offspring resulting from the mutation.
    """
    self.offspring_original = copy.copy(offsprings)
    self.parents = parents  # Not used, but kept for consistency
    offspring = offsprings.to_numpy().astype(dtype=np.bool)

    # create a boolean mask based on the mutation probability
    flip_mask = self.rng.random(offspring.shape) < self.mutation_probability

    # using XOR (^), flip the bits in the offspring when the mask is True
    # otherwise leave the bit's value as it is
    offspring = offspring ^ flip_mask

    self.offspring = pl.from_numpy(offspring, schema=self.variable_symbols).select(pl.all()).cast(pl.Float64)
    self.notify()

    return self.offspring
state
state() -> Sequence[Message]

Return the state of the mutation operator.

Source code in desdeo/emo/operators/mutation.py
def state(self) -> Sequence[Message]:
    """Return the state of the mutation operator."""
    if self.offspring_original is None or self.offspring is None:
        return []
    if self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return [
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_PROBABILITY,
                source=self.__class__.__name__,
                value=self.mutation_probability,
            ),
        ]
    # verbosity == 2
    return [
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.OFFSPRING_ORIGINAL,
            source=self.__class__.__name__,
            value=self.offspring_original,
        ),
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.PARENTS,
            source=self.__class__.__name__,
            value=self.parents,
        ),
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.OFFSPRINGS,
            source=self.__class__.__name__,
            value=self.offspring,
        ),
        FloatMessage(
            topic=MutationMessageTopics.MUTATION_PROBABILITY,
            source=self.__class__.__name__,
            value=self.mutation_probability,
        ),
    ]
update
update(*_, **__)

Do nothing.

Source code in desdeo/emo/operators/mutation.py
def update(self, *_, **__):
    """Do nothing."""

BoundedPolynomialMutation

Bases: BaseMutation

Implements the bounded polynomial mutation operator.

Reference

Deb, K., & Goyal, M. (1996). A combined genetic adaptive search (GeneAS) for engineering design. Computer Science and informatics, 26(4), 30-45, 1996.

Source code in desdeo/emo/operators/mutation.py
class BoundedPolynomialMutation(BaseMutation):
    """Implements the bounded polynomial mutation operator.

    Reference:
        Deb, K., & Goyal, M. (1996). A combined genetic adaptive search (GeneAS) for
        engineering design. Computer Science and informatics, 26(4), 30-45, 1996.
    """

    @property
    def provided_topics(self) -> dict[int, Sequence[MutationMessageTopics]]:
        """The message topics provided by the mutation operator."""
        return {
            0: [],
            1: [
                MutationMessageTopics.MUTATION_PROBABILITY,
                MutationMessageTopics.MUTATION_DISTRIBUTION,
            ],
            2: [
                MutationMessageTopics.MUTATION_PROBABILITY,
                MutationMessageTopics.MUTATION_DISTRIBUTION,
                MutationMessageTopics.OFFSPRING_ORIGINAL,
                MutationMessageTopics.OFFSPRINGS,
            ],
        }

    @property
    def interested_topics(self):
        """The message topics that the mutation operator is interested in."""
        return []

    def __init__(
        self,
        *,
        problem: Problem,
        seed: int,
        mutation_probability: float | None = None,
        distribution_index: float = 20,
        **kwargs,
    ):
        """Initialize a bounded polynomial mutation operator.

        Args:
            problem (Problem): The problem object.
            seed (int): The seed for the random number generator.
            mutation_probability (float | None, optional): The probability of mutation. Defaults to None.
            distribution_index (float, optional): The distributaion index for polynomial mutation. Defaults to 20.
            kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
                publisher must be passed. See the Subscriber class for more information.
        """
        super().__init__(problem, **kwargs)
        if self.variable_combination != VariableDomainTypeEnum.continuous:
            raise ValueError("This mutation operator only works with continuous variables.")
        if mutation_probability is None:
            self.mutation_probability = 1 / len(self.lower_bounds)
        else:
            self.mutation_probability = mutation_probability
        self.distribution_index = distribution_index
        self.rng = np.random.default_rng(seed)
        self.seed = seed
        self.offspring_original: pl.DataFrame
        self.parents: pl.DataFrame
        self.offspring: pl.DataFrame

    def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
        """Perform the mutation operation.

        Args:
            offsprings (pl.DataFrame): the offspring population to mutate.
            parents (pl.DataFrame): the parent population from which the offspring
                was generated (via crossover).

        Returns:
            pl.DataFrame: the offspring resulting from the mutation.
        """
        # TODO(@light-weaver): Extract to a numba jitted function
        self.offspring_original = offsprings
        self.parents = parents  # Not used, but kept for consistency
        offspring = offsprings.to_numpy()
        min_val = np.ones_like(offspring) * self.lower_bounds
        max_val = np.ones_like(offspring) * self.upper_bounds
        k = self.rng.random(size=offspring.shape)
        miu = self.rng.random(size=offspring.shape)
        temp = np.logical_and((k <= self.mutation_probability), (miu < 0.5))
        offspring_scaled = (offspring - min_val) / (max_val - min_val)
        offspring[temp] = offspring[temp] + (
            (max_val[temp] - min_val[temp])
            * (
                (2 * miu[temp] + (1 - 2 * miu[temp]) * (1 - offspring_scaled[temp]) ** (self.distribution_index + 1))
                ** (1 / (self.distribution_index + 1))
                - 1
            )
        )
        temp = np.logical_and((k <= self.mutation_probability), (miu >= 0.5))
        offspring[temp] = offspring[temp] + (
            (max_val[temp] - min_val[temp])
            * (
                1
                - (
                    2 * (1 - miu[temp])
                    + 2 * (miu[temp] - 0.5) * offspring_scaled[temp] ** (self.distribution_index + 1)
                )
                ** (1 / (self.distribution_index + 1))
            )
        )
        offspring[offspring > max_val] = max_val[offspring > max_val]
        offspring[offspring < min_val] = min_val[offspring < min_val]
        self.offspring = pl.from_numpy(offspring, schema=self.variable_symbols)
        self.notify()
        return self.offspring

    def update(self, *_, **__):
        """Do nothing. This is just the basic polynomial mutation operator."""

    def state(self) -> Sequence[Message]:
        """Return the state of the mutation operator."""
        if self.offspring_original is None or self.offspring is None:
            return []
        if self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return [
                FloatMessage(
                    topic=MutationMessageTopics.MUTATION_PROBABILITY,
                    source=self.__class__.__name__,
                    value=self.mutation_probability,
                ),
                FloatMessage(
                    topic=MutationMessageTopics.MUTATION_DISTRIBUTION,
                    source=self.__class__.__name__,
                    value=self.distribution_index,
                ),
            ]
        # verbosity == 2
        return [
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.OFFSPRING_ORIGINAL,
                source=self.__class__.__name__,
                value=self.offspring_original,
            ),
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.PARENTS,
                source=self.__class__.__name__,
                value=self.parents,
            ),
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.OFFSPRINGS,
                source=self.__class__.__name__,
                value=self.offspring,
            ),
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_PROBABILITY,
                source=self.__class__.__name__,
                value=self.mutation_probability,
            ),
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_DISTRIBUTION,
                source=self.__class__.__name__,
                value=self.distribution_index,
            ),
        ]
interested_topics property
interested_topics

The message topics that the mutation operator is interested in.

provided_topics property
provided_topics: dict[int, Sequence[MutationMessageTopics]]

The message topics provided by the mutation operator.

__init__
__init__(*, problem: Problem, seed: int, mutation_probability: float | None = None, distribution_index: float = 20, **kwargs)

Initialize a bounded polynomial mutation operator.

Parameters:

Name Type Description Default
problem Problem

The problem object.

required
seed int

The seed for the random number generator.

required
mutation_probability float | None

The probability of mutation. Defaults to None.

None
distribution_index float

The distributaion index for polynomial mutation. Defaults to 20.

20
kwargs

Additional keyword arguments. These are passed to the Subscriber class. At the very least, the publisher must be passed. See the Subscriber class for more information.

{}
Source code in desdeo/emo/operators/mutation.py
def __init__(
    self,
    *,
    problem: Problem,
    seed: int,
    mutation_probability: float | None = None,
    distribution_index: float = 20,
    **kwargs,
):
    """Initialize a bounded polynomial mutation operator.

    Args:
        problem (Problem): The problem object.
        seed (int): The seed for the random number generator.
        mutation_probability (float | None, optional): The probability of mutation. Defaults to None.
        distribution_index (float, optional): The distributaion index for polynomial mutation. Defaults to 20.
        kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
            publisher must be passed. See the Subscriber class for more information.
    """
    super().__init__(problem, **kwargs)
    if self.variable_combination != VariableDomainTypeEnum.continuous:
        raise ValueError("This mutation operator only works with continuous variables.")
    if mutation_probability is None:
        self.mutation_probability = 1 / len(self.lower_bounds)
    else:
        self.mutation_probability = mutation_probability
    self.distribution_index = distribution_index
    self.rng = np.random.default_rng(seed)
    self.seed = seed
    self.offspring_original: pl.DataFrame
    self.parents: pl.DataFrame
    self.offspring: pl.DataFrame
do
do(offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame

Perform the mutation operation.

Parameters:

Name Type Description Default
offsprings DataFrame

the offspring population to mutate.

required
parents DataFrame

the parent population from which the offspring was generated (via crossover).

required

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring resulting from the mutation.

Source code in desdeo/emo/operators/mutation.py
def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
    """Perform the mutation operation.

    Args:
        offsprings (pl.DataFrame): the offspring population to mutate.
        parents (pl.DataFrame): the parent population from which the offspring
            was generated (via crossover).

    Returns:
        pl.DataFrame: the offspring resulting from the mutation.
    """
    # TODO(@light-weaver): Extract to a numba jitted function
    self.offspring_original = offsprings
    self.parents = parents  # Not used, but kept for consistency
    offspring = offsprings.to_numpy()
    min_val = np.ones_like(offspring) * self.lower_bounds
    max_val = np.ones_like(offspring) * self.upper_bounds
    k = self.rng.random(size=offspring.shape)
    miu = self.rng.random(size=offspring.shape)
    temp = np.logical_and((k <= self.mutation_probability), (miu < 0.5))
    offspring_scaled = (offspring - min_val) / (max_val - min_val)
    offspring[temp] = offspring[temp] + (
        (max_val[temp] - min_val[temp])
        * (
            (2 * miu[temp] + (1 - 2 * miu[temp]) * (1 - offspring_scaled[temp]) ** (self.distribution_index + 1))
            ** (1 / (self.distribution_index + 1))
            - 1
        )
    )
    temp = np.logical_and((k <= self.mutation_probability), (miu >= 0.5))
    offspring[temp] = offspring[temp] + (
        (max_val[temp] - min_val[temp])
        * (
            1
            - (
                2 * (1 - miu[temp])
                + 2 * (miu[temp] - 0.5) * offspring_scaled[temp] ** (self.distribution_index + 1)
            )
            ** (1 / (self.distribution_index + 1))
        )
    )
    offspring[offspring > max_val] = max_val[offspring > max_val]
    offspring[offspring < min_val] = min_val[offspring < min_val]
    self.offspring = pl.from_numpy(offspring, schema=self.variable_symbols)
    self.notify()
    return self.offspring
state
state() -> Sequence[Message]

Return the state of the mutation operator.

Source code in desdeo/emo/operators/mutation.py
def state(self) -> Sequence[Message]:
    """Return the state of the mutation operator."""
    if self.offspring_original is None or self.offspring is None:
        return []
    if self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return [
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_PROBABILITY,
                source=self.__class__.__name__,
                value=self.mutation_probability,
            ),
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_DISTRIBUTION,
                source=self.__class__.__name__,
                value=self.distribution_index,
            ),
        ]
    # verbosity == 2
    return [
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.OFFSPRING_ORIGINAL,
            source=self.__class__.__name__,
            value=self.offspring_original,
        ),
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.PARENTS,
            source=self.__class__.__name__,
            value=self.parents,
        ),
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.OFFSPRINGS,
            source=self.__class__.__name__,
            value=self.offspring,
        ),
        FloatMessage(
            topic=MutationMessageTopics.MUTATION_PROBABILITY,
            source=self.__class__.__name__,
            value=self.mutation_probability,
        ),
        FloatMessage(
            topic=MutationMessageTopics.MUTATION_DISTRIBUTION,
            source=self.__class__.__name__,
            value=self.distribution_index,
        ),
    ]
update
update(*_, **__)

Do nothing. This is just the basic polynomial mutation operator.

Source code in desdeo/emo/operators/mutation.py
def update(self, *_, **__):
    """Do nothing. This is just the basic polynomial mutation operator."""

IntegerRandomMutation

Bases: BaseMutation

Implements a random mutation operator for integer variables.

The mutation will mutate each binary integer variable, by changing its value to a random value bounded by the variable's bounds with a provided probability.

Source code in desdeo/emo/operators/mutation.py
class IntegerRandomMutation(BaseMutation):
    """Implements a random mutation operator for integer variables.

    The mutation will mutate each binary integer variable,
    by changing its value to a random value bounded by the
    variable's bounds with a provided probability.
    """

    @property
    def provided_topics(self) -> dict[int, Sequence[MutationMessageTopics]]:
        """The message topics provided by the mutation operator."""
        return {
            0: [],
            1: [
                MutationMessageTopics.MUTATION_PROBABILITY,
            ],
            2: [
                MutationMessageTopics.MUTATION_PROBABILITY,
                MutationMessageTopics.OFFSPRING_ORIGINAL,
                MutationMessageTopics.OFFSPRINGS,
            ],
        }

    @property
    def interested_topics(self):
        """The message topics that the mutation operator is interested in."""
        return []

    def __init__(
        self,
        *,
        problem: Problem,
        seed: int,
        mutation_probability: float | None = None,
        **kwargs,
    ):
        """Initialize a random integer mutation operator.

        Args:
            problem (Problem): The problem object.
            seed (int): The seed for the random number generator.
            mutation_probability (float | None, optional): The probability of mutation. If None,
                the probability will be set to be 1/n, where n is the number of decision variables
                in the problem. Defaults to None.
            kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
                publisher must be passed. See the Subscriber class for more information.
        """
        super().__init__(problem, **kwargs)

        if self.variable_combination != VariableDomainTypeEnum.integer:
            raise ValueError("This mutation operator only works with integer variables.")
        if mutation_probability is None:
            self.mutation_probability = 1 / len(self.variable_symbols)
        else:
            self.mutation_probability = mutation_probability

        self.rng = np.random.default_rng(seed)
        self.seed = seed
        self.offspring_original: pl.DataFrame
        self.parents: pl.DataFrame
        self.offspring: pl.DataFrame

    def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
        """Perform the random integer mutation operation.

        Args:
            offsprings (pl.DataFrame): the offspring population to mutate.
            parents (pl.DataFrame): the parent population from which the offspring
                was generated (via crossover). Not used in the mutation operator.

        Returns:
            pl.DataFrame: the offspring resulting from the mutation.
        """
        self.offspring_original = copy.copy(offsprings)
        self.parents = parents  # Not used, but kept for consistency

        population = offsprings.to_numpy().astype(int)

        # create a boolean mask based on the mutation probability
        mutation_mask = self.rng.random(population.shape) < self.mutation_probability

        mutated = np.where(
            mutation_mask,
            self.rng.integers(self.lower_bounds, self.upper_bounds, size=population.shape, dtype=int, endpoint=True),
            population,
        )

        self.offspring = pl.from_numpy(mutated, schema=self.variable_symbols).select(pl.all()).cast(pl.Float64)
        self.notify()

        return self.offspring

    def update(self, *_, **__):
        """Do nothing."""

    def state(self) -> Sequence[Message]:
        """Return the state of the mutation operator."""
        if self.offspring_original is None or self.offspring is None:
            return []
        if self.verbosity == 0:
            return []
        if self.verbosity == 1:
            return [
                FloatMessage(
                    topic=MutationMessageTopics.MUTATION_PROBABILITY,
                    source=self.__class__.__name__,
                    value=self.mutation_probability,
                ),
            ]
        # verbosity == 2
        return [
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.OFFSPRING_ORIGINAL,
                source=self.__class__.__name__,
                value=self.offspring_original,
            ),
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.PARENTS,
                source=self.__class__.__name__,
                value=self.parents,
            ),
            PolarsDataFrameMessage(
                topic=MutationMessageTopics.OFFSPRINGS,
                source=self.__class__.__name__,
                value=self.offspring,
            ),
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_PROBABILITY,
                source=self.__class__.__name__,
                value=self.mutation_probability,
            ),
        ]
interested_topics property
interested_topics

The message topics that the mutation operator is interested in.

provided_topics property
provided_topics: dict[int, Sequence[MutationMessageTopics]]

The message topics provided by the mutation operator.

__init__
__init__(*, problem: Problem, seed: int, mutation_probability: float | None = None, **kwargs)

Initialize a random integer mutation operator.

Parameters:

Name Type Description Default
problem Problem

The problem object.

required
seed int

The seed for the random number generator.

required
mutation_probability float | None

The probability of mutation. If None, the probability will be set to be 1/n, where n is the number of decision variables in the problem. Defaults to None.

None
kwargs

Additional keyword arguments. These are passed to the Subscriber class. At the very least, the publisher must be passed. See the Subscriber class for more information.

{}
Source code in desdeo/emo/operators/mutation.py
def __init__(
    self,
    *,
    problem: Problem,
    seed: int,
    mutation_probability: float | None = None,
    **kwargs,
):
    """Initialize a random integer mutation operator.

    Args:
        problem (Problem): The problem object.
        seed (int): The seed for the random number generator.
        mutation_probability (float | None, optional): The probability of mutation. If None,
            the probability will be set to be 1/n, where n is the number of decision variables
            in the problem. Defaults to None.
        kwargs: Additional keyword arguments. These are passed to the Subscriber class. At the very least, the
            publisher must be passed. See the Subscriber class for more information.
    """
    super().__init__(problem, **kwargs)

    if self.variable_combination != VariableDomainTypeEnum.integer:
        raise ValueError("This mutation operator only works with integer variables.")
    if mutation_probability is None:
        self.mutation_probability = 1 / len(self.variable_symbols)
    else:
        self.mutation_probability = mutation_probability

    self.rng = np.random.default_rng(seed)
    self.seed = seed
    self.offspring_original: pl.DataFrame
    self.parents: pl.DataFrame
    self.offspring: pl.DataFrame
do
do(offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame

Perform the random integer mutation operation.

Parameters:

Name Type Description Default
offsprings DataFrame

the offspring population to mutate.

required
parents DataFrame

the parent population from which the offspring was generated (via crossover). Not used in the mutation operator.

required

Returns:

Type Description
DataFrame

pl.DataFrame: the offspring resulting from the mutation.

Source code in desdeo/emo/operators/mutation.py
def do(self, offsprings: pl.DataFrame, parents: pl.DataFrame) -> pl.DataFrame:
    """Perform the random integer mutation operation.

    Args:
        offsprings (pl.DataFrame): the offspring population to mutate.
        parents (pl.DataFrame): the parent population from which the offspring
            was generated (via crossover). Not used in the mutation operator.

    Returns:
        pl.DataFrame: the offspring resulting from the mutation.
    """
    self.offspring_original = copy.copy(offsprings)
    self.parents = parents  # Not used, but kept for consistency

    population = offsprings.to_numpy().astype(int)

    # create a boolean mask based on the mutation probability
    mutation_mask = self.rng.random(population.shape) < self.mutation_probability

    mutated = np.where(
        mutation_mask,
        self.rng.integers(self.lower_bounds, self.upper_bounds, size=population.shape, dtype=int, endpoint=True),
        population,
    )

    self.offspring = pl.from_numpy(mutated, schema=self.variable_symbols).select(pl.all()).cast(pl.Float64)
    self.notify()

    return self.offspring
state
state() -> Sequence[Message]

Return the state of the mutation operator.

Source code in desdeo/emo/operators/mutation.py
def state(self) -> Sequence[Message]:
    """Return the state of the mutation operator."""
    if self.offspring_original is None or self.offspring is None:
        return []
    if self.verbosity == 0:
        return []
    if self.verbosity == 1:
        return [
            FloatMessage(
                topic=MutationMessageTopics.MUTATION_PROBABILITY,
                source=self.__class__.__name__,
                value=self.mutation_probability,
            ),
        ]
    # verbosity == 2
    return [
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.OFFSPRING_ORIGINAL,
            source=self.__class__.__name__,
            value=self.offspring_original,
        ),
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.PARENTS,
            source=self.__class__.__name__,
            value=self.parents,
        ),
        PolarsDataFrameMessage(
            topic=MutationMessageTopics.OFFSPRINGS,
            source=self.__class__.__name__,
            value=self.offspring,
        ),
        FloatMessage(
            topic=MutationMessageTopics.MUTATION_PROBABILITY,
            source=self.__class__.__name__,
            value=self.mutation_probability,
        ),
    ]
update
update(*_, **__)

Do nothing.

Source code in desdeo/emo/operators/mutation.py
def update(self, *_, **__):
    """Do nothing."""

Selection operators

desdeo.emo.operators.selection

The base class for selection operators.

This whole file should be rewritten. Everything is a mess. Moreover, the selectors do not yet take seeds as input for reproducibility. TODO:@light-weaver

BaseDecompositionSelector

Bases: BaseSelector

Base class for decomposition based selection operators.

Source code in desdeo/emo/operators/selection.py
class BaseDecompositionSelector(BaseSelector):
    """Base class for decomposition based selection operators."""

    def __init__(self, problem: Problem, reference_vector_options: ReferenceVectorOptions, **kwargs):
        super().__init__(problem, **kwargs)
        self.reference_vector_options = reference_vector_options
        self.reference_vectors: np.ndarray
        self.reference_vectors_initial: np.ndarray

        # Set default values
        if "creation_type" not in self.reference_vector_options:
            self.reference_vector_options["creation_type"] = "simplex"
        if "vector_type" not in self.reference_vector_options:
            self.reference_vector_options["vector_type"] = "spherical"
        if "adaptation_frequency" not in self.reference_vector_options:
            self.reference_vector_options["adaptation_frequency"] = 100
        if self.reference_vector_options["creation_type"] == "simplex":
            self._create_simplex()
        elif self.reference_vector_options["creation_type"] == "s_energy":
            raise NotImplementedError("Riesz s-energy criterion is not yet implemented.")

        if "interactive_adaptation" not in self.reference_vector_options:
            self.reference_vector_options["interactive_adaptation"] = "none"
        elif self.reference_vector_options["interactive_adaptation"] != "none":
            self.reference_vector_options["adaptation_frequency"] = 0
        if "adaptation_distance" not in self.reference_vector_options:
            self.reference_vector_options["adaptation_distance"] = 0.2
        self._create_simplex()

        if self.reference_vector_options["interactive_adaptation"] == "reference_point":
            if "reference_point" not in self.reference_vector_options:
                raise ValueError("Reference point must be specified for interactive adaptation.")
            self.interactive_adapt_3(
                np.array([self.reference_vector_options["reference_point"][x] for x in self.target_symbols]),
                translation_param=self.reference_vector_options["adaptation_distance"],
            )
        elif self.reference_vector_options["interactive_adaptation"] == "preferred_solutions":
            if "preferred_solutions" not in self.reference_vector_options:
                raise ValueError("Preferred solutions must be specified for interactive adaptation.")
            self.interactive_adapt_1(
                np.array([self.reference_vector_options["preferred_solutions"][x] for x in self.target_symbols]).T,
                translation_param=self.reference_vector_options["adaptation_distance"],
            )
        elif self.reference_vector_options["interactive_adaptation"] == "non_preferred_solutions":
            if "non_preferred_solutions" not in self.reference_vector_options:
                raise ValueError("Non-preferred solutions must be specified for interactive adaptation.")
            self.interactive_adapt_2(
                np.array([self.reference_vector_options["non_preferred_solutions"][x] for x in self.target_symbols]).T,
                predefined_distance=self.reference_vector_options["adaptation_distance"],
            )
        elif self.reference_vector_options["interactive_adaptation"] == "preferred_ranges":
            if "preferred_ranges" not in self.reference_vector_options:
                raise ValueError("Preferred ranges must be specified for interactive adaptation.")
            self.interactive_adapt_4(
                np.array([self.reference_vector_options["preferred_ranges"][x] for x in self.target_symbols]).T,
            )

    def _create_simplex(self):
        """Create the reference vectors using simplex lattice design."""

        def approx_lattice_resolution(number_of_vectors: int, num_dims: int) -> int:
            """Approximate the lattice resolution based on the number of vectors."""
            temp_lattice_resolution = 0
            while True:
                temp_lattice_resolution += 1
                temp_number_of_vectors = comb(
                    temp_lattice_resolution + num_dims - 1,
                    num_dims - 1,
                    exact=True,
                )
                if temp_number_of_vectors > number_of_vectors:
                    break
            return temp_lattice_resolution - 1

        if "lattice_resolution" in self.reference_vector_options:
            lattice_resolution = self.reference_vector_options["lattice_resolution"]
        elif "number_of_vectors" in self.reference_vector_options:
            lattice_resolution = approx_lattice_resolution(
                self.reference_vector_options["number_of_vectors"], num_dims=self.num_dims
            )
        else:
            lattice_resolution = approx_lattice_resolution(500, num_dims=self.num_dims)

        number_of_vectors: int = comb(
            lattice_resolution + self.num_dims - 1,
            self.num_dims - 1,
            exact=True,
        )

        self.reference_vector_options["number_of_vectors"] = number_of_vectors
        self.reference_vector_options["lattice_resolution"] = lattice_resolution

        temp1 = range(1, self.num_dims + lattice_resolution)
        temp1 = np.array(list(combinations(temp1, self.num_dims - 1)))
        temp2 = np.array([range(self.num_dims - 1)] * number_of_vectors)
        temp = temp1 - temp2 - 1
        weight = np.zeros((number_of_vectors, self.num_dims), dtype=int)
        weight[:, 0] = temp[:, 0]
        for i in range(1, self.num_dims - 1):
            weight[:, i] = temp[:, i] - temp[:, i - 1]
        weight[:, -1] = lattice_resolution - temp[:, -1]
        self.reference_vectors = weight / lattice_resolution
        self.reference_vectors_initial = np.copy(self.reference_vectors)
        self._normalize_rvs()

    def _normalize_rvs(self):
        """Normalize the reference vectors to a unit hypersphere."""
        if self.reference_vector_options["vector_type"] == "spherical":
            norm = np.linalg.norm(self.reference_vectors, axis=1).reshape(-1, 1)
            norm[norm == 0] = np.finfo(float).eps
        elif self.reference_vector_options["vector_type"] == "planar":
            norm = np.sum(self.reference_vectors, axis=1).reshape(-1, 1)
        else:
            raise ValueError("Invalid vector type. Must be either 'spherical' or 'planar'.")
        self.reference_vectors = np.divide(self.reference_vectors, norm)

    def interactive_adapt_1(self, z: np.ndarray, translation_param: float) -> None:
        """Adapt reference vectors using the information about prefererred solution(s) selected by the Decision maker.

        Args:
            z (np.ndarray): Preferred solution(s).
            translation_param (float): Parameter determining how close the reference vectors are to the central vector
            **v** defined by using the selected solution(s) z.
        """
        if z.shape[0] == 1:
            # single preferred solution
            # calculate new reference vectors
            self.reference_vectors = translation_param * self.reference_vectors_initial + ((1 - translation_param) * z)

        else:
            # multiple preferred solutions
            # calculate new reference vectors for each preferred solution
            values = [translation_param * self.reference_vectors_initial + ((1 - translation_param) * z_i) for z_i in z]

            # combine arrays of reference vectors into a single array and update reference vectors
            self.reference_vectors = np.concatenate(values)

        self._normalize_rvs()
        self.add_edge_vectors()

    def interactive_adapt_2(self, z: np.ndarray, predefined_distance: float) -> None:
        """Adapt reference vectors by using the information about non-preferred solution(s) selected by the Decision maker.

        After the Decision maker has specified non-preferred solution(s), Euclidian distance between normalized solution
        vector(s) and each of the reference vectors are calculated. Those reference vectors that are **closer** than a
        predefined distance are either **removed** or **re-positioned** somewhere else.

        Note:
            At the moment, only the **removal** of reference vectors is supported. Repositioning of the reference
            vectors is **not** supported.

        Note:
            In case the Decision maker specifies multiple non-preferred solutions, the reference vector(s) for which the
            distance to **any** of the non-preferred solutions is less than predefined distance are removed.

        Note:
            Future developer should implement a way for a user to say: "Remove some percentage of
            objecive space/reference vectors" rather than giving a predefined distance value.

        Args:
            z (np.ndarray): Non-preferred solution(s).
            predefined_distance (float): The reference vectors that are closer than this distance are either removed or
            re-positioned somewhere else.
            Default value: 0.2
        """
        # calculate L1 norm of non-preferred solution(s)
        z = np.atleast_2d(z)
        norm = np.linalg.norm(z, ord=2, axis=1).reshape(np.shape(z)[0], 1)

        # non-preferred solutions normalized
        v_c = np.divide(z, norm)

        # distances from non-preferred solution(s) to each reference vector
        distances = np.array(
            [
                list(
                    map(
                        lambda solution: np.linalg.norm(solution - value, ord=2),
                        v_c,
                    )
                )
                for value in self.reference_vectors
            ]
        )

        # find out reference vectors that are not closer than threshold value to any non-preferred solution
        mask = [all(d >= predefined_distance) for d in distances]

        # set those reference vectors that met previous condition as new reference vectors, drop others
        self.reference_vectors = self.reference_vectors[mask]

        self._normalize_rvs()
        self.add_edge_vectors()

    def interactive_adapt_3(self, ref_point, translation_param):
        """Adapt reference vectors linearly towards a reference point. Then normalize.

        The details can be found in the following paper: Hakanen, Jussi &
        Chugh, Tinkle & Sindhya, Karthik & Jin, Yaochu & Miettinen, Kaisa.
        (2016). Connections of Reference Vectors and Different Types of
        Preference Information in Interactive Multiobjective Evolutionary
        Algorithms.

        Parameters
        ----------
        ref_point :

        translation_param :
            (Default value = 0.2)

        """
        self.reference_vectors = self.reference_vectors_initial * translation_param + (
            (1 - translation_param) * ref_point
        )
        self._normalize_rvs()
        self.add_edge_vectors()

    def interactive_adapt_4(self, preferred_ranges: np.ndarray) -> None:
        """Adapt reference vectors by using the information about the Decision maker's preferred range for each of the objective.

        Using these ranges, Latin hypercube sampling is applied to generate m number of samples between
        within these ranges, where m is the number of reference vectors. Normalized vectors constructed of these samples
        are then set as new reference vectors.

        Args:
            preferred_ranges (np.ndarray): Preferred lower and upper bound for each of the objective function values.
        """
        # bounds
        lower_limits = np.min(preferred_ranges, axis=0)
        upper_limits = np.max(preferred_ranges, axis=0)

        # generate samples using Latin hypercube sampling
        lhs = LatinHypercube(d=self.num_dims)
        w = lhs.random(n=self.reference_vectors_initial.shape[0])

        # scale between bounds
        w = w * (upper_limits - lower_limits) + lower_limits

        # set new reference vectors and normalize them
        self.reference_vectors = w
        self._normalize_rvs()
        self.add_edge_vectors()

    def add_edge_vectors(self):
        """Add edge vectors to the list of reference vectors.

        Used to cover the entire orthant when preference information is
        provided.

        """
        edge_vectors = np.eye(self.reference_vectors.shape[1])
        self.reference_vectors = np.vstack([self.reference_vectors, edge_vectors])
        self._normalize_rvs()
_create_simplex
_create_simplex()

Create the reference vectors using simplex lattice design.

Source code in desdeo/emo/operators/selection.py
def _create_simplex(self):
    """Create the reference vectors using simplex lattice design."""

    def approx_lattice_resolution(number_of_vectors: int, num_dims: int) -> int:
        """Approximate the lattice resolution based on the number of vectors."""
        temp_lattice_resolution = 0
        while True:
            temp_lattice_resolution += 1
            temp_number_of_vectors = comb(
                temp_lattice_resolution + num_dims - 1,
                num_dims - 1,
                exact=True,
            )
            if temp_number_of_vectors > number_of_vectors:
                break
        return temp_lattice_resolution - 1

    if "lattice_resolution" in self.reference_vector_options:
        lattice_resolution = self.reference_vector_options["lattice_resolution"]
    elif "number_of_vectors" in self.reference_vector_options:
        lattice_resolution = approx_lattice_resolution(
            self.reference_vector_options["number_of_vectors"], num_dims=self.num_dims
        )
    else:
        lattice_resolution = approx_lattice_resolution(500, num_dims=self.num_dims)

    number_of_vectors: int = comb(
        lattice_resolution + self.num_dims - 1,
        self.num_dims - 1,
        exact=True,
    )

    self.reference_vector_options["number_of_vectors"] = number_of_vectors
    self.reference_vector_options["lattice_resolution"] = lattice_resolution

    temp1 = range(1, self.num_dims + lattice_resolution)
    temp1 = np.array(list(combinations(temp1, self.num_dims - 1)))
    temp2 = np.array([range(self.num_dims - 1)] * number_of_vectors)
    temp = temp1 - temp2 - 1
    weight = np.zeros((number_of_vectors, self.num_dims), dtype=int)
    weight[:, 0] = temp[:, 0]
    for i in range(1, self.num_dims - 1):
        weight[:, i] = temp[:, i] - temp[:, i - 1]
    weight[:, -1] = lattice_resolution - temp[:, -1]
    self.reference_vectors = weight / lattice_resolution
    self.reference_vectors_initial = np.copy(self.reference_vectors)
    self._normalize_rvs()
_normalize_rvs
_normalize_rvs()

Normalize the reference vectors to a unit hypersphere.

Source code in desdeo/emo/operators/selection.py
def _normalize_rvs(self):
    """Normalize the reference vectors to a unit hypersphere."""
    if self.reference_vector_options["vector_type"] == "spherical":
        norm = np.linalg.norm(self.reference_vectors, axis=1).reshape(-1, 1)
        norm[norm == 0] = np.finfo(float).eps
    elif self.reference_vector_options["vector_type"] == "planar":
        norm = np.sum(self.reference_vectors, axis=1).reshape(-1, 1)
    else:
        raise ValueError("Invalid vector type. Must be either 'spherical' or 'planar'.")
    self.reference_vectors = np.divide(self.reference_vectors, norm)
add_edge_vectors
add_edge_vectors()

Add edge vectors to the list of reference vectors.

Used to cover the entire orthant when preference information is provided.

Source code in desdeo/emo/operators/selection.py
def add_edge_vectors(self):
    """Add edge vectors to the list of reference vectors.

    Used to cover the entire orthant when preference information is
    provided.

    """
    edge_vectors = np.eye(self.reference_vectors.shape[1])
    self.reference_vectors = np.vstack([self.reference_vectors, edge_vectors])
    self._normalize_rvs()
interactive_adapt_1
interactive_adapt_1(z: np.ndarray, translation_param: float) -> None

Adapt reference vectors using the information about prefererred solution(s) selected by the Decision maker.

Parameters:

Name Type Description Default
z ndarray

Preferred solution(s).

required
translation_param float

Parameter determining how close the reference vectors are to the central vector

required
Source code in desdeo/emo/operators/selection.py
def interactive_adapt_1(self, z: np.ndarray, translation_param: float) -> None:
    """Adapt reference vectors using the information about prefererred solution(s) selected by the Decision maker.

    Args:
        z (np.ndarray): Preferred solution(s).
        translation_param (float): Parameter determining how close the reference vectors are to the central vector
        **v** defined by using the selected solution(s) z.
    """
    if z.shape[0] == 1:
        # single preferred solution
        # calculate new reference vectors
        self.reference_vectors = translation_param * self.reference_vectors_initial + ((1 - translation_param) * z)

    else:
        # multiple preferred solutions
        # calculate new reference vectors for each preferred solution
        values = [translation_param * self.reference_vectors_initial + ((1 - translation_param) * z_i) for z_i in z]

        # combine arrays of reference vectors into a single array and update reference vectors
        self.reference_vectors = np.concatenate(values)

    self._normalize_rvs()
    self.add_edge_vectors()
interactive_adapt_2
interactive_adapt_2(z: np.ndarray, predefined_distance: float) -> None

Adapt reference vectors by using the information about non-preferred solution(s) selected by the Decision maker.

After the Decision maker has specified non-preferred solution(s), Euclidian distance between normalized solution vector(s) and each of the reference vectors are calculated. Those reference vectors that are closer than a predefined distance are either removed or re-positioned somewhere else.

Note

At the moment, only the removal of reference vectors is supported. Repositioning of the reference vectors is not supported.

Note

In case the Decision maker specifies multiple non-preferred solutions, the reference vector(s) for which the distance to any of the non-preferred solutions is less than predefined distance are removed.

Note

Future developer should implement a way for a user to say: "Remove some percentage of objecive space/reference vectors" rather than giving a predefined distance value.

Parameters:

Name Type Description Default
z ndarray

Non-preferred solution(s).

required
predefined_distance float

The reference vectors that are closer than this distance are either removed or

required
Default value

0.2

required
Source code in desdeo/emo/operators/selection.py