use 'from __future__ import annotations' (#346)

Author: basnijholt

adaptive/learner/average_learner.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 from math import sqrt
-from typing import Callable, Dict, List, Optional, Tuple
+from typing import Callable
 
 import cloudpickle
 import numpy as np
@@ -38,8 +40,8 @@ class AverageLearner(BaseLearner):
     def __init__(
         self,
         function: Callable[[int], Real],
-        atol: Optional[float] = None,
-        rtol: Optional[float] = None,
+        atol: float | None = None,
+        rtol: float | None = None,
         min_npoints: int = 2,
     ) -> None:
         if atol is None and rtol is None:
@@ -68,7 +70,7 @@ def to_numpy(self):
         """Data as NumPy array of size (npoints, 2) with seeds and values."""
         return np.array(sorted(self.data.items()))
 
-    def ask(self, n: int, tell_pending: bool = True) -> Tuple[List[int], List[Float]]:
+    def ask(self, n: int, tell_pending: bool = True) -> tuple[list[int], list[Float]]:
         points = list(range(self.n_requested, self.n_requested + n))
 
         if any(p in self.data or p in self.pending_points for p in points):
@@ -159,10 +161,10 @@ def plot(self):
         vals = hv.Points(vals)
         return hv.operation.histogram(vals, num_bins=num_bins, dimension="y")
 
-    def _get_data(self) -> Tuple[Dict[int, Real], int, Real, Real]:
+    def _get_data(self) -> tuple[dict[int, Real], int, Real, Real]:
         return (self.data, self.npoints, self.sum_f, self.sum_f_sq)
 
-    def _set_data(self, data: Tuple[Dict[int, Real], int, Real, Real]) -> None:
+    def _set_data(self, data: tuple[dict[int, Real], int, Real, Real]) -> None:
         self.data, self.npoints, self.sum_f, self.sum_f_sq = data
 
     def __getstate__(self):

adaptive/learner/average_learner1D.py

@@ -1,19 +1,11 @@
+from __future__ import annotations
+
 import math
 import sys
 from collections import defaultdict
 from copy import deepcopy
 from math import hypot
-from typing import (
-    Callable,
-    DefaultDict,
-    Dict,
-    Iterable,
-    List,
-    Optional,
-    Sequence,
-    Set,
-    Tuple,
-)
+from typing import Callable, DefaultDict, Iterable, List, Sequence, Tuple
 
 import numpy as np
 import scipy.stats
@@ -27,7 +19,7 @@
 Point = Tuple[int, Real]
 Points = List[Point]
 
-__all__: List[str] = ["AverageLearner1D"]
+__all__: list[str] = ["AverageLearner1D"]
 
 
 class AverageLearner1D(Learner1D):
@@ -75,11 +67,10 @@ class AverageLearner1D(Learner1D):
 
     def __init__(
         self,
-        function: Callable[[Tuple[int, Real]], Real],
-        bounds: Tuple[Real, Real],
-        loss_per_interval: Optional[
-            Callable[[Sequence[Real], Sequence[Real]], float]
-        ] = None,
+        function: Callable[[tuple[int, Real]], Real],
+        bounds: tuple[Real, Real],
+        loss_per_interval: None
+        | (Callable[[Sequence[Real], Sequence[Real]], float]) = None,
         delta: float = 0.2,
         alpha: float = 0.005,
         neighbor_sampling: float = 0.3,
@@ -115,15 +106,15 @@ def __init__(
         self._number_samples = SortedDict()
         # This set contains the points x that have less than min_samples
         # samples or less than a (neighbor_sampling*100)% of their neighbors
-        self._undersampled_points: Set[Real] = set()
+        self._undersampled_points: set[Real] = set()
         # Contains the error in the estimate of the
         # mean at each point x in the form {x0: error(x0), ...}
-        self.error: Dict[Real, float] = decreasing_dict()
+        self.error: dict[Real, float] = decreasing_dict()
         #  Distance between two neighboring points in the
         # form {xi: ((xii-xi)^2 + (yii-yi)^2)^0.5, ...}
-        self._distances: Dict[Real, float] = decreasing_dict()
+        self._distances: dict[Real, float] = decreasing_dict()
         # {xii: error[xii]/min(_distances[xi], _distances[xii], ...}
-        self.rescaled_error: Dict[Real, float] = decreasing_dict()
+        self.rescaled_error: dict[Real, float] = decreasing_dict()
 
     @property
     def nsamples(self) -> int:
@@ -136,7 +127,7 @@ def min_samples_per_point(self) -> int:
             return 0
         return min(self._number_samples.values())
 
-    def ask(self, n: int, tell_pending: bool = True) -> Tuple[Points, List[float]]:
+    def ask(self, n: int, tell_pending: bool = True) -> tuple[Points, list[float]]:
         """Return 'n' points that are expected to maximally reduce the loss."""
         # If some point is undersampled, resample it
         if len(self._undersampled_points):
@@ -165,7 +156,7 @@ def ask(self, n: int, tell_pending: bool = True) -> Tuple[Points, List[float]]:
 
         return points, loss_improvements
 
-    def _ask_for_more_samples(self, x: Real, n: int) -> Tuple[Points, List[float]]:
+    def _ask_for_more_samples(self, x: Real, n: int) -> tuple[Points, list[float]]:
         """When asking for n points, the learner returns n times an existing point
         to be resampled, since in general n << min_samples and this point will
         need to be resampled many more times"""
@@ -184,7 +175,7 @@ def _ask_for_more_samples(self, x: Real, n: int) -> Tuple[Points, List[float]]:
         loss_improvements = [loss_improvement / n] * n
         return points, loss_improvements
 
-    def _ask_for_new_point(self, n: int) -> Tuple[Points, List[float]]:
+    def _ask_for_new_point(self, n: int) -> tuple[Points, list[float]]:
         """When asking for n new points, the learner returns n times a single
         new point, since in general n << min_samples and this point will need
         to be resampled many more times"""
@@ -398,7 +389,7 @@ def tell_many(self, xs: Points, ys: Sequence[Real]) -> None:
                 # simultaneously, before we move on to a new x
                 self.tell_many_at_point(x, seed_y_mapping)
 
-    def tell_many_at_point(self, x: Real, seed_y_mapping: Dict[int, Real]) -> None:
+    def tell_many_at_point(self, x: Real, seed_y_mapping: dict[int, Real]) -> None:
         """Tell the learner about many samples at a certain location x.
 
         Parameters
@@ -454,10 +445,10 @@ def tell_many_at_point(self, x: Real, seed_y_mapping: Dict[int, Real]) -> None:
                     self._update_interpolated_loss_in_interval(*interval)
                 self._oldscale = deepcopy(self._scale)
 
-    def _get_data(self) -> Dict[Real, Real]:
+    def _get_data(self) -> dict[Real, Real]:
         return self._data_samples
 
-    def _set_data(self, data: Dict[Real, Real]) -> None:
+    def _set_data(self, data: dict[Real, Real]) -> None:
         if data:
             for x, samples in data.items():
                 self.tell_many_at_point(x, samples)
@@ -491,7 +482,7 @@ def plot(self):
         return p.redim(x=dict(range=plot_bounds))
 
 
-def decreasing_dict() -> Dict:
+def decreasing_dict() -> dict:
     """This initialization orders the dictionary from large to small values"""
 
     def sorting_rule(key, value):

adaptive/learner/base_learner.py

@@ -110,12 +110,10 @@ def tell_many(self, xs, ys):
     def tell_pending(self, x):
         """Tell the learner that 'x' has been requested such
         that it's not suggested again."""
-        pass
 
     @abc.abstractmethod
     def remove_unfinished(self):
         """Remove uncomputed data from the learner."""
-        pass
 
     @abc.abstractmethod
     def loss(self, real=True):
@@ -142,7 +140,6 @@ def ask(self, n, tell_pending=True):
             `pending_points`. Set this to False if you do not
             want to modify the state of the learner.
         """
-        pass
 
     @abc.abstractmethod
     def _get_data(self):

adaptive/learner/learner1D.py

@@ -1,8 +1,10 @@
+from __future__ import annotations
+
 import collections.abc
 import itertools
 import math
 from copy import copy, deepcopy
-from typing import Any, Callable, Dict, List, Optional, Sequence, Set, Tuple, Union
+from typing import Any, Callable, Dict, List, Sequence, Tuple, Union
 
 import cloudpickle
 import numpy as np
@@ -170,7 +172,7 @@ def curvature_loss(xs: XsType1, ys: YsType1) -> Float:
     return curvature_loss
 
 
-def linspace(x_left: Real, x_right: Real, n: Int) -> List[Float]:
+def linspace(x_left: Real, x_right: Real, n: Int) -> list[Float]:
     """This is equivalent to
     'np.linspace(x_left, x_right, n, endpoint=False)[1:]',
     but it is 15-30 times faster for small 'n'."""
@@ -194,7 +196,7 @@ def _get_neighbors_from_array(xs: np.ndarray) -> NeighborsType:
 
 def _get_intervals(
     x: float, neighbors: NeighborsType, nth_neighbors: int
-) -> List[Tuple[float, float]]:
+) -> list[tuple[float, float]]:
     nn = nth_neighbors
     i = neighbors.index(x)
     start = max(0, i - nn - 1)
@@ -249,9 +251,9 @@ class Learner1D(BaseLearner):
 
     def __init__(
         self,
-        function: Callable[[Real], Union[Float, np.ndarray]],
-        bounds: Tuple[Real, Real],
-        loss_per_interval: Optional[Callable[[XsTypeN, YsTypeN], Float]] = None,
+        function: Callable[[Real], Float | np.ndarray],
+        bounds: tuple[Real, Real],
+        loss_per_interval: Callable[[XsTypeN, YsTypeN], Float] | None = None,
     ):
         self.function = function  # type: ignore
 
@@ -267,8 +269,8 @@ def __init__(
         # the learners behavior in the tests.
         self._recompute_losses_factor = 2
 
-        self.data: Dict[Real, Real] = {}
-        self.pending_points: Set[Real] = set()
+        self.data: dict[Real, Real] = {}
+        self.pending_points: set[Real] = set()
 
         # A dict {x_n: [x_{n-1}, x_{n+1}]} for quick checking of local
         # properties.
@@ -292,7 +294,7 @@ def __init__(
         self.bounds = list(bounds)
         self.__missing_bounds = set(self.bounds)  # cache of missing bounds
 
-        self._vdim: Optional[int] = None
+        self._vdim: int | None = None
 
     @property
     def vdim(self) -> int:
@@ -334,20 +336,18 @@ def loss(self, real: bool = True) -> float:
         max_interval, max_loss = losses.peekitem(0)
         return max_loss
 
-    def _scale_x(self, x: Optional[Float]) -> Optional[Float]:
+    def _scale_x(self, x: Float | None) -> Float | None:
         if x is None:
             return None
         return x / self._scale[0]
 
-    def _scale_y(
-        self, y: Union[Float, np.ndarray, None]
-    ) -> Union[Float, np.ndarray, None]:
+    def _scale_y(self, y: Float | np.ndarray | None) -> Float | np.ndarray | None:
         if y is None:
             return None
         y_scale = self._scale[1] or 1
         return y / y_scale
 
-    def _get_point_by_index(self, ind: int) -> Optional[float]:
+    def _get_point_by_index(self, ind: int) -> float | None:
         if ind < 0 or ind >= len(self.neighbors):
             return None
         return self.neighbors.keys()[ind]
@@ -449,7 +449,7 @@ def _update_neighbors(self, x: float, neighbors: NeighborsType) -> None:
             neighbors.get(x_left, [None, None])[1] = x
             neighbors.get(x_right, [None, None])[0] = x
 
-    def _update_scale(self, x: float, y: Union[Float, np.ndarray]) -> None:
+    def _update_scale(self, x: float, y: Float | np.ndarray) -> None:
         """Update the scale with which the x and y-values are scaled.
 
         For a learner where the function returns a single scalar the scale
@@ -476,7 +476,7 @@ def _update_scale(self, x: float, y: Union[Float, np.ndarray]) -> None:
                 self._bbox[1][1] = max(self._bbox[1][1], y)
                 self._scale[1] = self._bbox[1][1] - self._bbox[1][0]
 
-    def tell(self, x: float, y: Union[Float, Sequence[Float], np.ndarray]) -> None:
+    def tell(self, x: float, y: Float | Sequence[Float] | np.ndarray) -> None:
         if x in self.data:
             # The point is already evaluated before
             return
@@ -522,13 +522,9 @@ def tell_pending(self, x: float) -> None:
     def tell_many(
         self,
         xs: Sequence[Float],
-        ys: Union[
-            Sequence[Float],
-            Sequence[Sequence[Float]],
-            Sequence[np.ndarray],
-        ],
+        ys: (Sequence[Float] | Sequence[Sequence[Float]] | Sequence[np.ndarray]),
         *,
-        force: bool = False
+        force: bool = False,
     ) -> None:
         if not force and not (len(xs) > 0.5 * len(self.data) and len(xs) > 2):
             # Only run this more efficient method if there are
@@ -597,7 +593,7 @@ def tell_many(
                 # have an inf loss.
                 self._update_interpolated_loss_in_interval(*ival)
 
-    def ask(self, n: int, tell_pending: bool = True) -> Tuple[List[float], List[float]]:
+    def ask(self, n: int, tell_pending: bool = True) -> tuple[list[float], list[float]]:
         """Return 'n' points that are expected to maximally reduce the loss."""
         points, loss_improvements = self._ask_points_without_adding(n)
 
@@ -607,7 +603,7 @@ def ask(self, n: int, tell_pending: bool = True) -> Tuple[List[float], List[floa
 
         return points, loss_improvements
 
-    def _missing_bounds(self) -> List[Real]:
+    def _missing_bounds(self) -> list[Real]:
         missing_bounds = []
         for b in copy(self.__missing_bounds):
             if b in self.data:
@@ -616,7 +612,7 @@ def _missing_bounds(self) -> List[Real]:
                 missing_bounds.append(b)
         return sorted(missing_bounds)
 
-    def _ask_points_without_adding(self, n: int) -> Tuple[List[float], List[float]]:
+    def _ask_points_without_adding(self, n: int) -> tuple[list[float], list[float]]:
         """Return 'n' points that are expected to maximally reduce the loss.
         Without altering the state of the learner"""
         # Find out how to divide the n points over the intervals
@@ -691,8 +687,8 @@ def _ask_points_without_adding(self, n: int) -> Tuple[List[float], List[float]]:
         return points, loss_improvements
 
     def _loss(
-        self, mapping: Dict[Interval, float], ival: Interval
-    ) -> Tuple[float, Interval]:
+        self, mapping: dict[Interval, float], ival: Interval
+    ) -> tuple[float, Interval]:
         loss = mapping[ival]
         return finite_loss(ival, loss, self._scale[0])
 
@@ -736,10 +732,10 @@ def remove_unfinished(self) -> None:
         self.losses_combined = deepcopy(self.losses)
         self.neighbors_combined = deepcopy(self.neighbors)
 
-    def _get_data(self) -> Dict[float, float]:
+    def _get_data(self) -> dict[float, float]:
         return self.data
 
-    def _set_data(self, data: Dict[float, float]) -> None:
+    def _set_data(self, data: dict[float, float]) -> None:
         if data:
             xs, ys = zip(*data.items())
             self.tell_many(xs, ys)
@@ -763,7 +759,7 @@ def __setstate__(self, state):
         self.losses_combined.update(losses_combined)
 
 
-def loss_manager(x_scale: float) -> Dict[Interval, float]:
+def loss_manager(x_scale: float) -> dict[Interval, float]:
     def sort_key(ival, loss):
         loss, ival = finite_loss(ival, loss, x_scale)
         return -loss, ival
@@ -772,7 +768,7 @@ def sort_key(ival, loss):
     return sorted_dict
 
 
-def finite_loss(ival: Interval, loss: float, x_scale: float) -> Tuple[float, Interval]:
+def finite_loss(ival: Interval, loss: float, x_scale: float) -> tuple[float, Interval]:
     """Get the so-called finite_loss of an interval in order to be able to
     sort intervals that have infinite loss."""
     # If the loss is infinite we return the

adaptive/runner.py

@@ -286,12 +286,10 @@ def elapsed_time(self):
 
         Is called in `overhead`.
         """
-        pass
 
     @abc.abstractmethod
     def _submit(self, x):
         """Is called in `_get_futures`."""
-        pass
 
     @property
     def tracebacks(self):