""" Abstract class for the construction of selectors used in selecting new optimizers. """
import logging
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
from .spawncontrol import _AlwaysSpawn, BaseController
from ..core.optimizerlogger import BaseLogger
from ..core.nosettingsobject import NoSettingsObject
from ..optimizers.baseoptimizer import BaseOptimizer
from ...plams.core.settings import Settings
__all__ = ("BaseSelector",)
[docs]class BaseSelector(ABC, NoSettingsObject):
"""Base selector from which all selectors must inherit to be compatible with GloMPO.
Selectors are classes which return an optimizer and its configuration when asked by the manager. This selection will
then be used to start a new optimizer. The full manager is supplied to the selector allowing sophisticated decisions
to be designed.
:Parameters:
``*avail_opts``
A set of optimizers available to the minimization. Elements may be:
#. Subclasses or instances of :class:`.BaseOptimizer`.
#. Tuples of:
#. :class:`.BaseOptimizer` subclasses (not instances);
#. Dictionary of kwargs sent to :class:`BaseOptimizer.__init__ <.BaseOptimizer>`, or ``None``;
allow_spawn
Optional function sent to the selector which is called with the manager object as argument. If it returns
``False`` the manager will no longer spawn optimizers. See :ref:`Spawn Control`.
:Examples:
>>> DummySelector(OptimizerA(), OptimizerB(setting_a=7))
>>> DummySelector((OptimizerA, None), (OptimizerB, {'setting_a': 7}))
Both of the above are equivalent.
The ``DummySelector`` above may choose from two optimizers (``OptimizerA`` or ``OptimizerB``).
``OptimizerA`` has no special configuration settings. ``OptimizerB`` is configured with
``setting_a = 7`` at initialisation.
>>> DummySelector(OptimizerA, allow_spawn=IterSpawnStop(50_000))
In this case the selector will only spawn ``OptimizerA`` optimizers but not allow any spawning after 50000
function evaluations.
:Attributes:
allow_spawn : :class:`.BaseController`
Function used to control if a new optimizer should be allowed to be created.
avail_opts : List[Tuple[Type[BaseOptimizer], Optional[Dict[str, Any]]]]
Set of optimizers and configuration settings available to the optimizer.
logger : logging.Logger
:class:`logging.Logger` instance into which status messages may be added.
"""
def __init__(
self,
*avail_opts: Union[BaseOptimizer, Type[BaseOptimizer], Tuple[Type[BaseOptimizer], Optional[Dict[str, Any]]]],
allow_spawn: Optional[List[BaseController]] = None,
):
self.logger = logging.getLogger("glompo.selector")
self.avail_opts = []
for item in avail_opts:
try:
if isinstance(item, BaseOptimizer): # Deal with initialized optimizers
opt = type(item)
init_dict = item._init_signature
elif isinstance(item, tuple) and len(item) == 2:
opt, init_dict = item
assert issubclass(opt, BaseOptimizer)
if init_dict is None:
init_dict = {}
assert isinstance(init_dict, dict)
elif issubclass(item, BaseOptimizer):
opt = item
init_dict = {"_workers": 1}
else:
raise AssertionError
if "_workers" not in init_dict:
init_dict["_workers"] = 1
self.avail_opts.append((opt, init_dict))
except AssertionError as e:
raise ValueError(
f"Cannot parse {item}. Expected: Union[Type[BaseOptimizer], Tuple[BaseOptimizer, "
f"Dict[str, Any], Dict[str, Any]]] expected."
) from e
if callable(allow_spawn):
self.allow_spawn = [allow_spawn]
elif isinstance(allow_spawn, list):
self.allow_spawn = allow_spawn
else:
self.allow_spawn = [_AlwaysSpawn()]
def _spawnersettings(self, s: Settings) -> Settings:
for control in self.allow_spawn:
s = control.__amssettings__(s)
return s
def __amssettings__(self, s: Settings) -> Settings:
s = super().__amssettings__(s)
s = self._spawnersettings(s)
return s
[docs] @abstractmethod
def select_optimizer(
self, manager: "GloMPOManager", slots_available: int
) -> Union[Tuple[Type[BaseOptimizer], Dict[str, Any]], None, bool]:
"""Selects an optimizer to start from the available options.
:Parameters:
manager
:class:`.GloMPOManager` instance managing the optimization from which various attributes can be read.
slots_available
Number of processes/threads the manager is allowed to start according to
:class:`GloMPOManager.max_jobs <.GloMPOManager>` and the number of existing threads. GloMPO assumes that the
selector will use this parameter to return an optimizer which requires fewer processes/threads than
``slots_available``. If this is not possible then ``None`` is returned.
:Returns:
Union[Tuple[Type[BaseOptimizer], Dict[str, Any]], None, bool]
Optimizer class and configuration parameters: Tuple of optimizer class and dictionary of initialisation
parameters. Manager will use this to initialise and start a new optimizer.
``None`` is returned when no available optimizer configurations can satisfy the number of
worker slots available.
``False`` is a special return which flags that the manager must never try and start another optimizer for
the remainder of the optimization.
"""
def __contains__(self, item):
opts = (opt[0] for opt in self.avail_opts)
return item in opts