Source code for scm.glompo.common.namedtuples

""" Named tuples used throughout the package to make code clearer. """

from multiprocessing.context import Process
from multiprocessing.synchronize import Event
from multiprocessing.connection import Connection
from threading import Thread
from typing import Any, Callable, Dict, NamedTuple, Sequence, Type, Union

__all__ = ("Result", "Bound", "OptimizerPackage", "IterationResult", "ProcessPackage", "OptimizerCheckpoint")


[docs]class Result(NamedTuple):
    """Final result delivered by :class:`.GloMPOManager`."""

    x: Sequence[float]
    """ Sequence[float]: Location in input space producing the lowest function value. """
    fx: float
    """ float: The lowest function value found. """
    stats: Dict[str, Any]
    """ Dict[str, Any]: Dictionary of statistics surrounding the optimization. """
    origin: Dict[str, Any]
    """ Dict[str, Any]: Optimizer name, settings, starting point and termination condition. """


[docs]class Bound(NamedTuple):
    """Class of parameter bounds."""

    min: float
    """ float: Lower parameter bound. """
    max: float
    """ float: Upper parameter bound. """


[docs]class OptimizerPackage(NamedTuple):
    """Package of an initialized optimizer, its unique identifier and multiprocessing variables."""

    opt_id: int
    """ int: Unique optimizer identification number. """
    optimizer: Callable
    """ Callable: Instance of :class:`.BaseOptimizer` """
    signal_pipe: Connection
    """ :class:`~multiprocessing.connection.Connection`: Used to send messages between the optimizer and manager. """
    allow_run_event: Event
    """ :class:`multiprocessing.Event`: Used by the manager to pause the optimizer. """
    slots: int
    """ int: Maximum number of thread/processes the optimizer may spawn. """


[docs]class IterationResult(NamedTuple):
    """Return type of each optimizer iteration."""

    opt_id: int
    """ int: Unique optimizer identification number. """
    x: Sequence[float]
    """ Sequence[float]: Location in input space. """
    fx: float
    """ float: Corresponding function evaluation. """
    extras: Sequence[Any]
    """ Sequence[Any]: Possible returns if :meth:`~.BaseFunction.detailed_call` is used. """


[docs]class ProcessPackage(NamedTuple):
    """Package of a running process and its communication channels."""

    process: Union[Process, Thread]
    """ Union[:class:`~multiprocessing.Process`, :class:`~threading.Thread`]: Process within which optimizer is run. """
    signal_pipe: Connection
    """ :class:`~multiprocessing.connection.Connection`: Used to send messages between the optimizer and manager. """
    allow_run_event: Event
    """ :class:`multiprocessing.Event`: Used by the manager to pause the optimizer. """
    slots: int
    """ int: Maximum number of thread/processes the optimizer may spawn. """


[docs]class OptimizerCheckpoint(NamedTuple):
    """Information needed in the manager about initialized optimizers for checkpoint loading."""

    opt_type: Type["BaseOptimizer"]
    """ Type[:class:`.BaseOptimizer`]: Class of optimizer to be restarted. """
    slots: int
    """ int: Maximum number of thread/processes the optimizer may spawn. """