diff --git a/.gitignore b/.gitignore index 15e25256c..2e005a2ca 100644 --- a/.gitignore +++ b/.gitignore @@ -140,3 +140,7 @@ ENV/ # mypy .mypy_cache/ + +# pixi environments +.pixi +*.egg-info diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 000000000..76db8032d --- /dev/null +++ b/mypy.ini @@ -0,0 +1,9 @@ +[mypy] +ignore_missing_imports = True +files = pymoo/core/algorithm.py + +disable_error_code = misc, arg-type, name-defined, attr-defined, empty-body, annotation-unchecked + +;follow_imports = skip +disallow_untyped_defs = False +check_untyped_defs = False \ No newline at end of file diff --git a/pymoo/core/algorithm.py b/pymoo/core/algorithm.py index 54e7694f4..440ef94e2 100644 --- a/pymoo/core/algorithm.py +++ b/pymoo/core/algorithm.py @@ -1,13 +1,16 @@ import copy import time +from typing import Any import numpy as np from pymoo.core.callback import Callback from pymoo.core.evaluator import Evaluator from pymoo.core.population import Population +from pymoo.core.problem import Problem from pymoo.core.result import Result -from pymoo.termination.default import DefaultMultiObjectiveTermination, DefaultSingleObjectiveTermination +from pymoo.termination.default import DefaultMultiObjectiveTermination, DefaultSingleObjectiveTermination, Termination +from pymoo.util.archive import Archive from pymoo.util.display.display import Display from pymoo.util.function_loader import FunctionLoader from pymoo.util.misc import termination_from_tuple @@ -17,16 +20,16 @@ class Algorithm: def __init__(self, - termination=None, - output=None, - display=None, - callback=None, - archive=None, - return_least_infeasible=False, - save_history=False, - verbose=False, - seed=None, - evaluator=None, + termination: Termination | str | tuple[str, ...] | None = None, + output: str | None = None, + display: Display | None = None, + callback: Callback | None = None, + archive: Archive | None = None, + return_least_infeasible: bool = False, + save_history: bool = False, + verbose: bool = False, + seed: int | None = None, + evaluator: Evaluator | None = None, **kwargs): super().__init__() @@ -35,7 +38,7 @@ def __init__(self, FunctionLoader.get_instance() # the problem to be solved (will be set later on) - self.problem = None + self.problem: Problem | None = None # the termination criterion to be used by the algorithm - might be specific for an algorithm self.termination = termination @@ -72,30 +75,30 @@ def __init__(self, self.evaluator = evaluator # the history object which contains the list - self.history = list() + self.history: list[Algorithm] = [] # the current solutions stored - here considered as population - self.pop = None + self.pop: Population | None = None # a placeholder object for implementation to store solutions in each iteration - self.off = None + self.off: Population | None = None # the optimum found by the algorithm - self.opt = None + self.opt: Population | None = None # the current number of generation or iteration - self.n_iter = None + self.n_iter: int | None = None # can be used to store additional data in submodules - self.data = {} + self.data: dict[Any, Any] = {} # if the initialized method has been called before or not self.is_initialized = False # the time when the algorithm has been setup for the first time - self.start_time = None + self.start_time: float | None = None - def setup(self, problem, **kwargs): + def setup(self, problem: Problem, **kwargs) -> "Algorithm": # the problem to be solved by the algorithm self.problem = problem @@ -133,13 +136,14 @@ def setup(self, problem, **kwargs): return self - def run(self): + def run(self) -> Result: while self.has_next(): self.next() return self.result() - def has_next(self): - return not self.termination.has_terminated() + def has_next(self) -> bool: + assert self.termination is not None, "Algorithm has no termination" + return not self.termination.has_terminated() # type: ignore def finalize(self): @@ -148,7 +152,7 @@ def finalize(self): return self._finalize() - def next(self): + def next(self) -> None: # get the infill solutions infills = self.infill() @@ -162,7 +166,7 @@ def next(self): else: self.advance() - def _initialize(self): + def _initialize(self) -> None: # the time starts whenever this method is called self.start_time = time.time() @@ -172,7 +176,7 @@ def _initialize(self): self.pop = Population.empty() self.opt = None - def infill(self): + def infill(self) -> Population | None: if self.problem is None: raise Exception("Please call `setup(problem)` before calling next().") @@ -196,7 +200,7 @@ def infill(self): return infills - def advance(self, infills=None, **kwargs): + def advance(self, infills: Population | None = None, **kwargs) -> Population | Result: # if infills have been provided set them as offsprings and feed them into advance self.off = infills @@ -229,13 +233,13 @@ def advance(self, infills=None, **kwargs): self._post_advance() # if the algorithm has terminated, then do the finalization steps and return the result - if self.termination.has_terminated(): + if self.termination.has_terminated(): # type: ignore self.finalize() ret = self.result() # otherwise just increase the iteration counter for the next step and return the current optimum else: - ret = self.opt + ret = self.opt # type: ignore # add the infill solutions to an archive if self.archive is not None and infills is not None: @@ -243,13 +247,13 @@ def advance(self, infills=None, **kwargs): return ret - def result(self): + def result(self) -> Result: res = Result() # store the time when the algorithm as finished res.start_time = self.start_time res.end_time = time.time() - res.exec_time = res.end_time - res.start_time + res.exec_time = res.end_time - res.start_time # type: ignore res.pop = self.pop res.archive = self.archive @@ -257,7 +261,7 @@ def result(self): # get the optimal solution found opt = self.opt - if opt is None or len(opt) == 0: + if opt is None or len(opt) == 0: # type: ignore opt = None # if no feasible solution has been found @@ -274,10 +278,10 @@ def result(self): # otherwise get the values from the population else: - X, F, CV, G, H = self.opt.get("X", "F", "CV", "G", "H") + X, F, CV, G, H = self.opt.get("X", "F", "CV", "G", "H") # type: ignore # if single-objective problem and only one solution was found - create a 1d array - if self.problem.n_obj == 1 and len(X) == 1: + if self.problem.n_obj == 1 and len(X) == 1: # type: ignore X, F, CV, G, H = X[0], F[0], CV[0], G[0], H[0] # set all the individual values @@ -289,22 +293,22 @@ def result(self): return res - def ask(self): + def ask(self) -> Population | None: return self.infill() - def tell(self, *args, **kwargs): + def tell(self, *args, **kwargs) -> Population | Result: return self.advance(*args, **kwargs) - def _set_optimum(self): + def _set_optimum(self) -> None: self.opt = filter_optimum(self.pop, least_infeasible=True) - def _post_advance(self): + def _post_advance(self) -> None: # update the current optimum of the algorithm self._set_optimum() # update the current termination condition of the algorithm - self.termination.update(self) + self.termination.update(self) # type: ignore # display the output if defined by the algorithm self.display(self) @@ -315,32 +319,32 @@ def _post_advance(self): if self.save_history: _hist, _callback, _display = self.history, self.callback, self.display - self.history, self.callback, self.display = None, None, None + self.history, self.callback, self.display = None, None, None # type: ignore obj = copy.deepcopy(self) self.history, self.callback, self.display = _hist, _callback, _display self.history.append(obj) - self.n_iter += 1 + self.n_iter += 1 # type: ignore # ========================================================================================================= # TO BE OVERWRITTEN # ========================================================================================================= - def _setup(self, problem, **kwargs): + def _setup(self, problem, **kwargs) -> None: pass - def _initialize_infill(self): + def _initialize_infill(self) -> Population: pass - def _initialize_advance(self, infills=None, **kwargs): + def _initialize_advance(self, infills=None, **kwargs) -> None: pass - def _infill(self): + def _infill(self) -> Population: pass - def _advance(self, infills=None, **kwargs): - pass + def _advance(self, infills=None, **kwargs) -> Any: + raise NotImplementedError() def _finalize(self): pass @@ -386,7 +390,7 @@ def _advance(self, infills=None, **kwargs): return False -def default_termination(problem): +def default_termination(problem: Problem) -> Termination: if problem.n_obj > 1: termination = DefaultMultiObjectiveTermination() else: diff --git a/pymoo/core/callback.py b/pymoo/core/callback.py index e768cc09b..83f91fd23 100644 --- a/pymoo/core/callback.py +++ b/pymoo/core/callback.py @@ -1,23 +1,30 @@ +from typing import Any +import typing + +if typing.TYPE_CHECKING: + from pymoo.core.algorithm import Algorithm + + class Callback: def __init__(self) -> None: super().__init__() - self.data = {} - self.is_initialized = False + self.data: dict[Any, Any] = {} + self.is_initialized: bool = False - def initialize(self, algorithm): + def initialize(self, algorithm: "Algorithm") -> None: pass - def notify(self, algorithm): + def notify(self, algorithm: "Algorithm") -> None: pass - def update(self, algorithm): + def update(self, algorithm: "Algorithm") -> Any: return self._update(algorithm) - def _update(self, algorithm): - pass + def _update(self, algorithm: "Algorithm") -> Any: + return None - def __call__(self, algorithm): + def __call__(self, algorithm: "Algorithm"): if not self.is_initialized: self.initialize(algorithm) @@ -31,8 +38,8 @@ class CallbackCollection(Callback): def __init__(self, *args) -> None: super().__init__() - self.callbacks = args + self.callbacks: typing.Iterable[Callback] = args - def update(self, algorithm): + def update(self, algorithm) -> None: [callback.update(algorithm) for callback in self.callbacks] diff --git a/pymoo/core/evaluator.py b/pymoo/core/evaluator.py index eb72d2155..65c134dd5 100644 --- a/pymoo/core/evaluator.py +++ b/pymoo/core/evaluator.py @@ -1,3 +1,5 @@ +from typing import Any + import numpy as np from pymoo.core.individual import Individual @@ -38,8 +40,8 @@ def __init__(self, def eval(self, problem: Problem, pop: Population, - skip_already_evaluated: bool = None, - evaluate_values_of: list = None, + skip_already_evaluated: bool | None = None, + evaluate_values_of: list[Any] | None = None, count_evals: bool = True, **kwargs): @@ -56,7 +58,7 @@ def eval(self, # filter the index to have individual where not all attributes have been evaluated if skip_already_evaluated: - I = [i for i, ind in enumerate(pop) if not all([e in ind.evaluated for e in evaluate_values_of])] + I = np.array([i for i, ind in enumerate(pop) if not all([e in ind.evaluated for e in evaluate_values_of])]) # if skipping is deactivated simply make the index being all individuals else: diff --git a/pymoo/core/individual.py b/pymoo/core/individual.py index f5d9c2ae8..7bc631f12 100644 --- a/pymoo/core/individual.py +++ b/pymoo/core/individual.py @@ -1,3 +1,5 @@ +# mypy: disable-error-code="return-value" + """ Module containing infrastructure for representing individuals in population-based optimization algorithms. @@ -12,7 +14,7 @@ ] import copy -from typing import Optional +from typing import Optional, Callable, Any from typing import Tuple from typing import Union from warnings import warn @@ -63,36 +65,36 @@ def __init__( in the ``Individual``. """ # set decision variable vector to None - self._X = None + self._X: np.ndarray | None = None # set values objective(s), inequality constraint(s), equality # contstraint(s) to None - self._F = None - self._G = None - self._H = None + self._F: np.ndarray | None = None + self._G: np.ndarray | None = None + self._H: np.ndarray | None = None # set first derivatives of objective(s), inequality constraint(s), # equality contstraint(s) to None - self._dF = None - self._dG = None - self._dH = None + self._dF: np.ndarray | None = None + self._dG: np.ndarray | None = None + self._dH: np.ndarray | None = None # set second derivatives of objective(s), inequality constraint(s), # equality contstraint(s) to None - self._ddF = None - self._ddG = None - self._ddH = None + self._ddF: np.ndarray | None = None + self._ddG: np.ndarray | None = None + self._ddH: np.ndarray | None = None # set constraint violation value to None - self._CV = None + self._CV: np.ndarray | None = None - self.evaluated = None + self.evaluated: set[Any] | None = None # initialize all the local variables self.reset() # a local storage for data - self.data = {} + self.data: dict[Any, Any] = {} # the config for this individual if config is None: @@ -594,8 +596,8 @@ def set( def get( self, - *keys: Tuple[str,...], - ) -> Union[tuple,object]: + *keys: Tuple[str, ...], + ) -> Union[tuple, object]: """ Get the values for one or more keys for an individual. @@ -614,7 +616,7 @@ def get( for key in keys: if hasattr(self, key): - v = getattr(self, key) + v = getattr(self, key) # type: ignore elif key in self.data: v = self.data[key] else: @@ -731,14 +733,14 @@ def calc_cv( elif G.ndim == 1: ieq_cv = constr_to_cv(G, **config["cv_ieq"]) else: - ieq_cv = [constr_to_cv(g, **config["cv_ieq"]) for g in G] + ieq_cv = [constr_to_cv(g, **config["cv_ieq"]) for g in G] # type: ignore if H is None: eq_cv = [0.0] elif H.ndim == 1: eq_cv = constr_to_cv(np.abs(H), **config["cv_eq"]) else: - eq_cv = [constr_to_cv(np.abs(h), **config["cv_eq"]) for h in H] + eq_cv = [constr_to_cv(np.abs(h), **config["cv_eq"]) for h in H] # type: ignore return np.array(ieq_cv) + np.array(eq_cv) @@ -748,7 +750,7 @@ def constr_to_cv( eps: float = 0.0, scale: Optional[float] = None, pow: Optional[float] = None, - func: object = np.mean, + func: Callable = np.mean, ) -> float: """ Convert a constraint to a constraint violation. diff --git a/pymoo/core/population.py b/pymoo/core/population.py index cc3feab19..2f9169489 100644 --- a/pymoo/core/population.py +++ b/pymoo/core/population.py @@ -1,3 +1,5 @@ +from typing import Callable, Any, Optional + import numpy as np from pymoo.core.individual import Individual @@ -5,30 +7,31 @@ class Population(np.ndarray): - def __new__(cls, individuals=[]): + def __new__(cls, individuals: Individual | list[Individual] | None = None): + individuals = individuals if individuals is not None else [] if isinstance(individuals, Individual): individuals = [individuals] return np.array(individuals).view(cls) - def has(self, key): + def has(self, key: str) -> bool: return all([ind.has(key) for ind in self]) - def collect(self, func, to_numpy=True): - val = [] + def collect(self, func: Callable, to_numpy: bool = True) -> list[Any] | np.ndarray: + val: list[Any] = [] for i in range(len(self)): val.append(func(self[i])) if to_numpy: - val = np.array(val) + return np.array(val) return val - def apply(self, func): + def apply(self, func: Callable) -> None: self.collect(func, to_numpy=False) - def set(self, *args, **kwargs): + def set(self, *args, **kwargs) -> Optional["Population"]: # if population is empty just return if self.size == 0: - return + return None # done for the old interface with the interleaving variable definition kwargs = interleaving_args(*args, kwargs=kwargs) @@ -51,9 +54,9 @@ def set(self, *args, **kwargs): return self - def get(self, *args, to_numpy=True, **kwargs): + def get(self, *args, to_numpy: bool = True, **kwargs) -> Any | tuple[Any, ...]: - val = {} + val: dict[Any, list[Any]] = {} for c in args: val[c] = [] @@ -78,7 +81,7 @@ def get(self, *args, to_numpy=True, **kwargs): return tuple(res) @classmethod - def merge(cls, a, b, *args): + def merge(cls, a, b, *args) -> "Population": # do the regular merge between first and second element m = merge(a, b) @@ -91,16 +94,16 @@ def merge(cls, a, b, *args): return m @classmethod - def create(cls, *args): + def create(cls, *args) -> "Population": return Population.__new__(cls, args) @classmethod - def empty(cls, size=0): + def empty(cls, size: int = 0) -> "Population": individuals = [Individual() for _ in range(size)] return Population.__new__(cls, individuals) @classmethod - def new(cls, *args, **kwargs): + def new(cls, *args, **kwargs) -> "Population": kwargs = interleaving_args(*args, kwargs=kwargs) if len(kwargs) > 0: @@ -118,7 +121,7 @@ def new(cls, *args, **kwargs): return pop -def pop_from_array_or_individual(array, pop=None): +def pop_from_array_or_individual(array: Population | np.ndarray | Individual, pop: Population | None = None) -> Population: # the population type can be different - (different type of individuals) if pop is None: pop = Population.empty() @@ -132,16 +135,18 @@ def pop_from_array_or_individual(array, pop=None): pop = Population.empty(1) pop[0] = array else: - return None + return None # type: ignore return pop -def merge(a, b): +def merge(a: Population | np.ndarray | Individual | None, b: Population | np.ndarray | Individual | None) -> Population: if a is None: - return b + assert b is not None, "Merge requires at least on non-empty Individual" + return pop_from_array_or_individual(b) elif b is None: - return a + assert a is not None, "Merge requires at least on non-empty Individual" + return pop_from_array_or_individual(a) a, b = pop_from_array_or_individual(a), pop_from_array_or_individual(b) @@ -167,7 +172,7 @@ def interleaving_args(*args, kwargs=None): return kwargs -def calc_cv(pop, config=None): +def calc_cv(pop: Population, config: dict[Any, Any] | None = None) -> np.ndarray: if config is None: config = Individual.default_config() diff --git a/pymoo/core/result.py b/pymoo/core/result.py index 33d2b3057..287bc1c20 100644 --- a/pymoo/core/result.py +++ b/pymoo/core/result.py @@ -1,3 +1,15 @@ +import typing +from typing import Any, Optional + +import numpy as np + +if typing.TYPE_CHECKING: + from pymoo.core.population import Population + from pymoo.core.problem import Problem + from pymoo.util.archive import Archive + from pymoo.core.algorithm import Algorithm + + class Result: """ The resulting object of an optimization run. @@ -6,38 +18,44 @@ class Result: def __init__(self) -> None: super().__init__() - self.opt = None + self.opt: Population | None = None self.success = None self.message = None # ! other attributes to be set as well # the problem that was solved - self.problem = None + self.problem: Optional[Problem] = None # the archive stored during the run - self.archive = None + self.archive: Optional[Archive] = None # the optimal solution for that problem self.pf = None # the algorithm that was used for optimization - self.algorithm = None + self.algorithm: Optional[Algorithm] = None # the final population if it applies - self.pop = None + self.pop: Optional[Population] = None # directly the values of opt - self.X, self.F, self.CV, self.G, self.H = None, None, None, None, None + self.X: Optional[np.ndarray] = None + self.F: Optional[np.ndarray] = None + self.CV: Optional[np.ndarray] = None + self.G: Optional[np.ndarray] = None + self.H: Optional[np.ndarray] = None # all the timings that are stored of the run - self.start_time, self.end_time, self.exec_time = None, None, None + self.start_time: Optional[float] = None + self.end_time: Optional[float] = None + self.exec_time: Optional[float] = None # the history of the optimization run is they were saved - self.history = [] + self.history: list[Algorithm] = [] # data stored within the algorithm - self.data = None + self.data: dict[Any, Any] | None = None @property def cv(self): diff --git a/pymoo/core/variable.py b/pymoo/core/variable.py index a8f1f5b3e..b7646c15e 100644 --- a/pymoo/core/variable.py +++ b/pymoo/core/variable.py @@ -137,7 +137,7 @@ class BoundedVariable(Variable): def __init__( self, value: Optional[object] = None, - bounds: Tuple[Optional[object],Optional[object]] = (None, None), + bounds: Tuple[Optional[int | float], Optional[int | float]] = (None, None), strict: Optional[Tuple[Optional[object],Optional[object]]] = None, **kwargs: dict, ) -> None: @@ -255,6 +255,7 @@ def _sample( decision variables. """ low, high = self.bounds + assert high is not None, "The upper bound must be specified" return np.random.randint(low, high=high + 1, size=n) @@ -355,10 +356,10 @@ def _sample( def get( - *args: Tuple[Union[Variable,object],...], - size: Optional[Union[tuple,int]] = None, + *args: Tuple[Union[Variable, object], ...], + size: Optional[Union[tuple, int]] = None, **kwargs: dict - ) -> Union[tuple,object,None]: + ) -> Union[tuple, object, None]: """ Get decision variable values from a tuple of ``Variable`` objects. @@ -378,7 +379,7 @@ def get( Decision variable value(s). """ if len(args) == 0: - return + return None ret = [] for arg in args: diff --git a/pymoo/termination/__init__.py b/pymoo/termination/__init__.py index 058506f3b..a9a060b2c 100644 --- a/pymoo/termination/__init__.py +++ b/pymoo/termination/__init__.py @@ -1,5 +1,10 @@ +import typing -def get_termination(name, *args, **kwargs): +if typing.TYPE_CHECKING: + from pymoo.core.termination import Termination + + +def get_termination(name, *args, **kwargs) -> "Termination": from pymoo.termination.default import DefaultMultiObjectiveTermination, DefaultSingleObjectiveTermination from pymoo.termination.max_eval import MaximumFunctionCallTermination from pymoo.termination.max_gen import MaximumGenerationTermination diff --git a/pymoo/util/archive.py b/pymoo/util/archive.py index d507db597..4e4f65090 100644 --- a/pymoo/util/archive.py +++ b/pymoo/util/archive.py @@ -25,7 +25,8 @@ def __init__(self, survival, problem=None) -> None: if problem is None: from pymoo.core.problem import Problem - problem = Problem() + # TODO: this line is probably never evaluated. It would raise as Problem is ABC + problem = Problem() # type: ignore self.problem = problem diff --git a/pymoo/util/misc.py b/pymoo/util/misc.py index 480bd7378..8e8afe14c 100644 --- a/pymoo/util/misc.py +++ b/pymoo/util/misc.py @@ -1,3 +1,5 @@ +import typing + from collections import OrderedDict from datetime import datetime from itertools import combinations @@ -8,6 +10,9 @@ from pymoo.core.population import Population from pymoo.core.sampling import Sampling +if typing.TYPE_CHECKING: + from pymoo.core.termination import Termination + def parameter_less(F, CV, fmax=None, inplace=False): @@ -367,7 +372,7 @@ def to_numpy(a): return np.array(a) -def termination_from_tuple(termination): +def termination_from_tuple(termination: typing.Union["Termination", str, tuple[str, ...]]) -> "Termination": from pymoo.core.termination import Termination # get the termination if provided as a tuple - create an object