Source code for nengo.base

from copy import copy
import warnings

import numpy as np

import nengo
from nengo.config import SupportDefaultsMixin
from nengo.exceptions import NotAddedToNetworkWarning, ValidationError
from nengo.params import (
    FrozenObject,
    IntParam,
    iter_params,
    NumberParam,
    Parameter,
    StringParam,
    Unconfigurable,
)
from nengo.utils.compat import is_integer, iteritems, range, with_metaclass
from nengo.utils.numpy import as_shape, maxint


class NetworkMember(type):
    """A metaclass used to add instances of derived classes to networks.

    Inheriting from this class means that Network.add will be invoked after
    initializing the object, unless add_to_container=False is passed to the
    derived class constructor.
    """

    def __call__(cls, *args, **kwargs):
        """Override default __call__ behavior so that Network.add is called."""
        inst = cls.__new__(cls)
        add_to_container = kwargs.pop('add_to_container', True)
        # Do the __init__ before adding in case __init__ errors out
        inst.__init__(*args, **kwargs)
        if add_to_container:
            nengo.Network.add(inst)
        inst._initialized = True  # value doesn't matter, just existance
        return inst


class NengoObject(with_metaclass(NetworkMember, SupportDefaultsMixin)):
    """A base class for Nengo objects.

    Parameters
    ----------
    label : string
        A descriptive label for the object.
    seed : int
        The seed used for random number generation.

    Attributes
    ----------
    label : string
        A descriptive label for the object.
    seed : int
        The seed used for random number generation.
    """

    # Order in which parameters have to be initialized.
    # Missing parameters will be initialized last in an undefined order.
    # This is needed for pickling and copying of Nengo objects when the
    # parameter initialization order matters.
    _param_init_order = []

    label = StringParam('label', default=None, optional=True)
    seed = IntParam('seed', default=None, optional=True)

    def __init__(self, label, seed):
        super(NengoObject, self).__init__()
        self.label = label
        self.seed = seed

    def __getstate__(self):
        state = self.__dict__.copy()
        del state['_initialized']

        for attr in self.params:
            param = getattr(type(self), attr)
            if self in param:
                state[attr] = getattr(self, attr)

        return state

    def __setstate__(self, state):
        for attr in self._param_init_order:
            setattr(self, attr, state.pop(attr))

        for attr in self.params:
            if attr in state:
                setattr(self, attr, state.pop(attr))

        for k, v in iteritems(state):
            setattr(self, k, v)

        self._initialized = True
        if len(nengo.Network.context) > 0:
            warnings.warn(NotAddedToNetworkWarning(self))

    def __setattr__(self, name, val):
        if hasattr(self, '_initialized') and not hasattr(self, name):
            warnings.warn(
                "Creating new attribute '%s' on '%s'. "
                "Did you mean to change an existing attribute?" % (name, self),
                SyntaxWarning)
        super(NengoObject, self).__setattr__(name, val)

    def __str__(self):
        return self._str(
            include_id=not hasattr(self, 'label') or self.label is None)

    def __repr__(self):
        return self._str(include_id=True)

    def _str(self, include_id):
        return "<%s%s%s>" % (
            type(self).__name__,
            "" if not hasattr(self, 'label') else
            " (unlabeled)" if self.label is None else
            ' "%s"' % self.label,
            " at 0x%x" % id(self) if include_id else "")

    @property
    def params(self):
        """Returns a list of parameter names that can be set."""
        return list(iter_params(self))

    def copy(self, add_to_container=True):
        with warnings.catch_warnings():
            # We warn when copying since we can't change add_to_container.
            # However, we deal with it here, so we ignore the warning.
            warnings.simplefilter('ignore', category=NotAddedToNetworkWarning)
            c = copy(self)
        if add_to_container:
            nengo.Network.add(c)
        return c


class ObjView(object):
    """Container for a slice with respect to some object.

    This is used by the __getitem__ of Neurons, Node, and Ensemble, in order
    to pass slices of those objects to Connection. This is a notational
    convenience for creating transforms. See Connection for details.

    Does not currently support any other view-like operations.
    """

    def __init__(self, obj, key=slice(None)):
        self.obj = obj

        # Node.size_in != size_out, so one of these can be invalid
        try:
            self.size_in = np.arange(self.obj.size_in)[key].size
        except IndexError:
            self.size_in = None
        try:
            self.size_out = np.arange(self.obj.size_out)[key].size
        except IndexError:
            self.size_out = None
        if self.size_in is None and self.size_out is None:
            raise IndexError("Invalid slice '%s' of %s" % (key, self.obj))

        if is_integer(key):
            # single slices of the form [i] should be cast into
            # slice objects for convenience
            if key == -1:
                # special case because slice(-1, 0) gives the empty list
                self.slice = slice(key, None)
            else:
                self.slice = slice(key, key+1)
        else:
            self.slice = key

    def copy(self):
        return copy(self)

    def __len__(self):
        return self.size_out

    def __str__(self):
        return "%s[%s]" % (self.obj, self._slice_string)

    def __repr__(self):
        return "%r[%s]" % (self.obj, self._slice_string)

    @property
    def _slice_string(self):
        if isinstance(self.slice, slice):
            sl_start = "" if self.slice.start is None else self.slice.start
            sl_stop = "" if self.slice.stop is None else self.slice.stop
            if self.slice.step is None:
                return "%s:%s" % (sl_start, sl_stop)
            else:
                return "%s:%s:%s" % (sl_start, sl_stop, self.slice.step)
        else:
            return str(self.slice)


class NengoObjectParam(Parameter):
    def __init__(self, name, optional=False, readonly=True,
                 nonzero_size_in=False, nonzero_size_out=False):
        default = Unconfigurable  # These can't have defaults
        self.nonzero_size_in = nonzero_size_in
        self.nonzero_size_out = nonzero_size_out
        super(NengoObjectParam, self).__init__(
            name, default, optional, readonly)

    def validate(self, instance, nengo_obj):
        nengo_objects = (
            NengoObject,
            ObjView,
            nengo.ensemble.Neurons,
            nengo.connection.LearningRule
        )
        if not isinstance(nengo_obj, nengo_objects):
            raise ValidationError("'%s' is not a Nengo object" % nengo_obj,
                                  attr=self.name, obj=instance)
        if self.nonzero_size_in and nengo_obj.size_in < 1:
            raise ValidationError("'%s' must have size_in > 0." % nengo_obj,
                                  attr=self.name, obj=instance)
        if self.nonzero_size_out and nengo_obj.size_out < 1:
            raise ValidationError("'%s' must have size_out > 0." % nengo_obj,
                                  attr=self.name, obj=instance)
        super(NengoObjectParam, self).validate(instance, nengo_obj)


[docs]class Process(FrozenObject): """A general system with input, output, and state. For more details on how to use processes and make custom process subclasses, see :doc:`examples/processes`. Parameters ---------- default_size_in : int (Default: 0) Sets the default size in for nodes using this process. default_size_out : int (Default: 1) Sets the default size out for nodes running this process. Also, if ``d`` is not specified in `~.Process.run` or `~.Process.run_steps`, this will be used. default_dt : float (Default: 0.001 (1 millisecond)) If ``dt`` is not specified in `~.Process.run`, `~.Process.run_steps`, `~.Process.ntrange`, or `~.Process.trange`, this will be used. seed : int, optional (Default: None) Random number seed. Ensures random factors will be the same each run. Attributes ---------- default_dt : float If ``dt`` is not specified in `~.Process.run`, `~.Process.run_steps`, `~.Process.ntrange`, or `~.Process.trange`, this will be used. default_size_in : int The default size in for nodes using this process. default_size_out : int The default size out for nodes running this process. Also, if ``d`` is not specified in `~.Process.run` or `~.Process.run_steps`, this will be used. seed : int or None Random number seed. Ensures random factors will be the same each run. """ default_size_in = IntParam('default_size_in', low=0) default_size_out = IntParam('default_size_out', low=0) default_dt = NumberParam('default_dt', low=0, low_open=True) seed = IntParam('seed', low=0, high=maxint, optional=True) def __init__(self, default_size_in=0, default_size_out=1, default_dt=0.001, seed=None): super(Process, self).__init__() self.default_size_in = default_size_in self.default_size_out = default_size_out self.default_dt = default_dt self.seed = seed
[docs] def apply(self, x, d=None, dt=None, rng=np.random, copy=True, **kwargs): """Run process on a given input. Keyword arguments that do not appear in the parameter list below will be passed to the ``make_step`` function of this process. Parameters ---------- x : ndarray The input signal given to the process. d : int, optional (Default: None) Output dimensionality. If None, ``default_size_out`` will be used. dt : float, optional (Default: None) Simulation timestep. If None, ``default_dt`` will be used. rng : `numpy.random.RandomState` (Default: ``numpy.random``) Random number generator used for stochstic processes. copy : bool, optional (Default: True) If True, a new output array will be created for output. If False, the input signal ``x`` will be overwritten. """ shape_in = as_shape(np.asarray(x[0]).shape, min_dim=1) shape_out = as_shape(self.default_size_out if d is None else d) dt = self.default_dt if dt is None else dt rng = self.get_rng(rng) step = self.make_step(shape_in, shape_out, dt, rng, **kwargs) output = np.zeros((len(x),) + shape_out) if copy else x for i, xi in enumerate(x): output[i] = step((i+1) * dt, xi) return output
[docs] def get_rng(self, rng): """Get a properly seeded independent RNG for the process step. Parameters ---------- rng : `numpy.random.RandomState` The parent random number generator to use if the seed is not set. """ seed = rng.randint(maxint) if self.seed is None else self.seed return np.random.RandomState(seed)
[docs] def make_step(self, shape_in, shape_out, dt, rng): """Create function that advances the process forward one time step. This must be implemented by all custom processes. Parameters ---------- shape_in : tuple The shape of the input signal. shape_out : tuple The shape of the output signal. dt : float The simulation timestep. rng : `numpy.random.RandomState` A random number generator. """ raise NotImplementedError("Process must implement `make_step` method.")
[docs] def run(self, t, d=None, dt=None, rng=np.random, **kwargs): """Run process without input for given length of time. Keyword arguments that do not appear in the parameter list below will be passed to the ``make_step`` function of this process. Parameters ---------- t : float The length of time to run. d : int, optional (Default: None) Output dimensionality. If None, ``default_size_out`` will be used. dt : float, optional (Default: None) Simulation timestep. If None, ``default_dt`` will be used. rng : `numpy.random.RandomState` (Default: ``numpy.random``) Random number generator used for stochstic processes. """ dt = self.default_dt if dt is None else dt n_steps = int(np.round(float(t) / dt)) return self.run_steps(n_steps, d=d, dt=dt, rng=rng, **kwargs)
[docs] def run_steps(self, n_steps, d=None, dt=None, rng=np.random, **kwargs): """Run process without input for given number of steps. Keyword arguments that do not appear in the parameter list below will be passed to the ``make_step`` function of this process. Parameters ---------- n_steps : int The number of steps to run. d : int, optional (Default: None) Output dimensionality. If None, ``default_size_out`` will be used. dt : float, optional (Default: None) Simulation timestep. If None, ``default_dt`` will be used. rng : `numpy.random.RandomState` (Default: ``numpy.random``) Random number generator used for stochstic processes. """ shape_in = as_shape(0) shape_out = as_shape(self.default_size_out if d is None else d) dt = self.default_dt if dt is None else dt rng = self.get_rng(rng) step = self.make_step(shape_in, shape_out, dt, rng, **kwargs) output = np.zeros((n_steps,) + shape_out) for i in range(n_steps): output[i] = step((i+1) * dt) return output
[docs] def ntrange(self, n_steps, dt=None): """Create time points corresponding to a given number of steps. Parameters ---------- n_steps : int The given number of steps. dt : float, optional (Default: None) Simulation timestep. If None, ``default_dt`` will be used. """ dt = self.default_dt if dt is None else dt return dt * np.arange(1, n_steps + 1)
[docs] def trange(self, t, dt=None): """Create time points corresponding to a given length of time. Parameters ---------- t : float The given length of time. dt : float, optional (Default: None) Simulation timestep. If None, ``default_dt`` will be used. """ dt = self.default_dt if dt is None else dt n_steps = int(np.round(float(t) / dt)) return self.ntrange(n_steps, dt=dt)
class ProcessParam(Parameter): """Must be a Process.""" def validate(self, instance, process): super(ProcessParam, self).validate(instance, process) if process is not None and not isinstance(process, Process): raise ValidationError( "Must be Process (got type %r)" % type(process).__name__, attr=self.name, obj=instance) return process