Source code for scm.glompo.opt_selectors.baseselector

""" 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