from copy import copy
import cloudpickle
from sortedcontainers import SortedDict, SortedSet
from adaptive.learner.base_learner import BaseLearner
class _IgnoreFirstArgument:
"""Remove the first argument from the call signature.
The SequenceLearner's function receives a tuple ``(index, point)``
but the original function only takes ``point``.
This is the same as `lambda x: function(x[1])`, however, that is not
pickable.
"""
def __init__(self, function):
self.function = function
def __call__(self, index_point, *args, **kwargs):
index, point = index_point
return self.function(point, *args, **kwargs)
def __getstate__(self):
return self.function
def __setstate__(self, function):
self.__init__(function)
[docs]class SequenceLearner(BaseLearner):
r"""A learner that will learn a sequence. It simply returns
the points in the provided sequence when asked.
This is useful when your problem cannot be formulated in terms of
another adaptive learner, but you still want to use Adaptive's
routines to run, save, and plot.
Parameters
----------
function : callable
The function to learn. Must take a single element `sequence`.
sequence : sequence
The sequence to learn.
Attributes
----------
data : dict
The data as a mapping from "index of element in sequence" => value.
Notes
-----
From primitive tests, the `~adaptive.SequenceLearner` appears to have a
similar performance to `ipyparallel`\s ``load_balanced_view().map``. With
the added benefit of having results in the local kernel already.
"""
def __init__(self, function, sequence):
self._original_function = function
self.function = _IgnoreFirstArgument(function)
self._to_do_indices = SortedSet({i for i, _ in enumerate(sequence)})
self._ntotal = len(sequence)
self.sequence = copy(sequence)
self.data = SortedDict()
self.pending_points = set()
[docs] def ask(self, n, tell_pending=True):
indices = []
points = []
loss_improvements = []
for index in self._to_do_indices:
if len(points) >= n:
break
point = self.sequence[index]
indices.append(index)
points.append((index, point))
loss_improvements.append(1 / self._ntotal)
if tell_pending:
for i, p in zip(indices, points):
self.tell_pending((i, p))
return points, loss_improvements
[docs] def loss(self, real=True):
if not (self._to_do_indices or self.pending_points):
return 0
else:
npoints = self.npoints + (0 if real else len(self.pending_points))
return (self._ntotal - npoints) / self._ntotal
[docs] def remove_unfinished(self):
for i in self.pending_points:
self._to_do_indices.add(i)
self.pending_points = set()
[docs] def tell(self, point, value):
index, point = point
self.data[index] = value
self.pending_points.discard(index)
self._to_do_indices.discard(index)
[docs] def tell_pending(self, point):
index, point = point
self.pending_points.add(index)
self._to_do_indices.discard(index)
[docs] def done(self):
return not self._to_do_indices and not self.pending_points
[docs] def result(self):
"""Get the function values in the same order as ``sequence``."""
if not self.done():
raise Exception("Learner is not yet complete.")
return list(self.data.values())
@property
def npoints(self):
return len(self.data)
def _get_data(self):
return self.data
def _set_data(self, data):
if data:
indices, values = zip(*data.items())
# the points aren't used by tell, so we can safely pass None
points = [(i, None) for i in indices]
self.tell_many(points, values)
def __getstate__(self):
return (
cloudpickle.dumps(self._original_function),
self.sequence,
self._get_data(),
)
def __setstate__(self, state):
function, sequence, data = state
function = cloudpickle.loads(function)
self.__init__(function, sequence)
self._set_data(data)