Extending algorithms

In jMetalPy, algorithms maintain a list of dependents or observers which are notified automatically after each iteration (think about event listeners). This is known as the observer pattern and can be used to extend the functionality of our algorithms by registering new observers.

For example, a basic observer will log the current number of evaluations, the objective(s) from the best individual in the population and the current computing time:

basic = BasicObserver(frequency=1)
algorithm.observable.register(observer=basic)

A progress bar observer will print a smart progress meter that increases, on each iteration, a fixed value (or step) until the maximum is reached:

max_evaluations = 25000

algorithm = GeneticAlgorithm(...)

progress_bar = ProgressBarObserver(max=max_evaluations)
algorithm.observable.register(progress_bar)

algorithm.run()

This will produce:

$ Progress:  50%|#####     | 12500/25000 [13:59<14:12, 14.66it/s]

A full list of all available observers can be found at the jmetal.util.observer module.

List of observers

observer

ProgressBarObserver(max)

Bases: Observer

Show a smart progress meter with the number of evaluations and computing time.

:param max: Number of expected iterations.

Source code in src/jmetal/util/observer.py

def __init__(self, max: int) -> None:
    """Show a smart progress meter with the number of evaluations and computing time.

    :param max: Number of expected iterations.
    """
    self.progress_bar = None
    self.progress = 0
    self._max = max

BasicObserver(frequency=1)

Bases: Observer

Show the number of evaluations, the best fitness and the computing time. :param frequency: Display frequency.

Source code in src/jmetal/util/observer.py

def __init__(self, frequency: int = 1) -> None:
    """Show the number of evaluations, the best fitness and the computing time.
    :param frequency: Display frequency."""

    self.display_frequency = frequency

PrintObjectivesObserver(frequency=1)

Bases: Observer

Show the number of evaluations, best fitness and computing time.

:param frequency: Display frequency.

Source code in src/jmetal/util/observer.py

def __init__(self, frequency: int = 1) -> None:
    """Show the number of evaluations, best fitness and computing time.

    :param frequency: Display frequency."""
    self.display_frequency = frequency

WriteFrontToFileObserver(output_directory)

Bases: Observer

Write function values of the front into files.

:param output_directory: Output directory. Each front will be saved on a file FUN.x.

Source code in src/jmetal/util/observer.py

def __init__(self, output_directory: str) -> None:
    """Write function values of the front into files.

    :param output_directory: Output directory. Each front will be saved on a file `FUN.x`."""
    self.counter = 0
    self.directory = output_directory

    if Path(self.directory).is_dir():
        LOGGER.warning(f"Directory {self.directory} exists. Removing contents.")
        for file in os.listdir(self.directory):
            os.remove(f"{self.directory}/{file}")
    else:
        LOGGER.warning(f"Directory {self.directory} does not exist. Creating it.")
        Path(self.directory).mkdir(parents=True)

PlotFrontToFileObserver(output_directory, step=100, **kwargs)

Bases: Observer

Plot and save Pareto front approximations into files.

:param output_directory: Output directory.

Source code in src/jmetal/util/observer.py

def __init__(self, output_directory: str, step: int = 100, **kwargs) -> None:
    """Plot and save Pareto front approximations into files.

    :param output_directory: Output directory.
    """
    self.directory = output_directory
    self.plot_front = Plot(title="Pareto front approximation", **kwargs)
    self.last_front: list[Solution] = []
    self.fronts: list[Solution] = []
    self.counter = 0
    self.step = step

    if Path(self.directory).is_dir():
        LOGGER.warning(f"Directory {self.directory} exists. Removing contents.")
        for file in os.listdir(self.directory):
            os.remove(f"{self.directory}/{file}")
    else:
        LOGGER.warning(f"Directory {self.directory} does not exist. Creating it.")
        Path(self.directory).mkdir(parents=True)

API

observer

This module implements the Observer design pattern for event handling in JMetalPy.

The Observer pattern allows objects to notify other objects about changes in their state. This is particularly useful for monitoring algorithm progress, logging, and visualization.

Observer

Bases: ABC

Abstract base class for observers in the Observer pattern.

Observers are objects that receive updates from Observable objects they are registered with. The type of the observable subject is not strictly enforced for better compatibility with Python's method resolution order (MRO).

Subclasses must implement the update() method to define how they handle notifications from observables.

update(subject, *args, **kwargs) abstractmethod

Receive an update from an observable subject.

This method is called whenever the observed subject changes state.

Parameters:

Name	Type	Description	Default
subject	Any	The observable object that sent the update.	required
*args	Any	Variable length argument list.	()
**kwargs	Any	Arbitrary keyword arguments containing update data. Common keys include: - 'evaluations': Current number of evaluations - 'solutions': Current population or solution set - 'computing_time': Elapsed computation time	{}

Source code in src/jmetal/core/observer.py

@abstractmethod
def update(self, subject: Any, *args: Any, **kwargs: Any) -> None:
    """Receive an update from an observable subject.

    This method is called whenever the observed subject changes state.

    Args:
        subject: The observable object that sent the update.
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments containing update data.
                Common keys include:
                - 'evaluations': Current number of evaluations
                - 'solutions': Current population or solution set
                - 'computing_time': Elapsed computation time
    """
    pass

Observable()

Bases: ABC

Abstract base class for observable subjects in the Observer pattern.

Observable objects maintain a list of observers and notify them when their state changes. This implementation is thread-safe and supports multiple observers.

Initialize the observable with an empty list of observers.

Source code in src/jmetal/core/observer.py

def __init__(self) -> None:
    """Initialize the observable with an empty list of observers."""
    self._observers: list[Observer] = []

register(observer) abstractmethod

Register an observer to receive updates.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to register.	required

Raises:

Type	Description
TypeError	If the observer is not an instance of Observer.

Source code in src/jmetal/core/observer.py

@abstractmethod
def register(self, observer: "Observer") -> None:
    """Register an observer to receive updates.

    Args:
        observer: The observer to register.

    Raises:
        TypeError: If the observer is not an instance of Observer.
    """
    pass

deregister(observer) abstractmethod

Remove an observer from the notification list.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to remove.	required

Note

If the observer is not in the list, this method does nothing.

Source code in src/jmetal/core/observer.py

@abstractmethod
def deregister(self, observer: "Observer") -> None:
    """Remove an observer from the notification list.

    Args:
        observer: The observer to remove.

    Note:
        If the observer is not in the list, this method does nothing.
    """
    pass

deregister_all() abstractmethod

Remove all observers from the notification list.

Source code in src/jmetal/core/observer.py

@abstractmethod
def deregister_all(self) -> None:
    """Remove all observers from the notification list."""
    pass

notify_all(*args, **kwargs) abstractmethod

Notify all registered observers.

This method calls the update() method on each registered observer, passing along any provided arguments.

Parameters:

Name	Type	Description	Default
*args	Any	Variable length argument list to pass to observers.	()
**kwargs	Any	Arbitrary keyword arguments to pass to observers.	{}

Source code in src/jmetal/core/observer.py

@abstractmethod
def notify_all(self, *args: Any, **kwargs: Any) -> None:
    """Notify all registered observers.

    This method calls the update() method on each registered observer,
    passing along any provided arguments.

    Args:
        *args: Variable length argument list to pass to observers.
        **kwargs: Arbitrary keyword arguments to pass to observers.
    """
    pass