Source code for brian2.core.base

All Brian objects should derive from `BrianObject`.
import functools
import weakref
import traceback
import os
import sys

from brian2.utils.logger import get_logger
from brian2.core.names import Nameable
from brian2.units.allunits import second
from brian2.units.fundamentalunits import check_units

__all__ = ['BrianObject',

logger = get_logger(__name__)

[docs]class BrianObject(Nameable): """ All Brian objects derive from this class, defines magic tracking and update. See the documentation for `Network` for an explanation of which objects get updated in which order. Parameters ---------- dt : `Quantity`, optional The time step to be used for the simulation. Cannot be combined with the `clock` argument. clock : `Clock`, optional The update clock to be used. If neither a clock, nor the `dt` argument is specified, the `defaultclock` will be used. when : str, optional In which scheduling slot to simulate the object during a time step. Defaults to ``'start'``. See :ref:`scheduling` for possible values. order : int, optional The priority of this object for operations occurring at the same time step and in the same scheduling slot. Defaults to 0. namespace: dict, optional A dictionary mapping identifier names to objects. If not given, the namespace will be filled in at the time of the call of ``, with either the values from the ``namespace`` argument of the `` method or from the local context, if no such argument is given. name : str, optional A unique name for the object - one will be assigned automatically if not provided (of the form ``brianobject_1``, etc.). Notes ----- The set of all `BrianObject` objects is stored in ``BrianObject.__instances__()``. """ @check_units(dt=second) def __init__(self, dt=None, clock=None, when='start', order=0, namespace=None, name='brianobject*'): # Setup traceback information for this object creation_stack = [] bases = [] for modulename in ['brian2']: if modulename in sys.modules: base, _ = os.path.split(sys.modules[modulename].__file__) bases.append(base) for fname, linenum, funcname, line in traceback.extract_stack(): if all(base not in fname for base in bases): s = (f" File '{fname}', line {linenum}, in {funcname}\n" f" {line}") creation_stack.append(s) creation_stack = [''] + creation_stack #: A string indicating where this object was created (traceback with any parts of Brian code removed) self._creation_stack = ("Object was created here (most recent call only, full details in " "debug log):\n" + creation_stack[-1]) self._full_creation_stack = "Object was created here:\n"+"\n".join(creation_stack) if dt is not None and clock is not None: raise ValueError("Can only specify either a dt or a clock, not both.") if not isinstance(when, str): from brian2.core.clocks import Clock # Give some helpful error messages for users coming from the alpha # version if isinstance(when, Clock): raise TypeError(("Do not use the 'when' argument for " "specifying a clock, either provide a " "timestep for the 'dt' argument or a Clock " "object for 'clock'.")) if isinstance(when, tuple): raise TypeError("Use the separate keyword arguments, 'dt' (or " "'clock'), 'when', and 'order' instead of " "providing a tuple for 'when'. Only use the " "'when' argument for the scheduling slot.") # General error raise TypeError("The 'when' argument has to be a string " "specifying the scheduling slot (e.g. 'start').") Nameable.__init__(self, name) #: The clock used for simulating this object self._clock = clock if clock is None: from brian2.core.clocks import Clock, defaultclock if dt is not None: self._clock = Clock(dt=dt,'_clock*') else: self._clock = defaultclock if getattr(self._clock, '_is_proxy', False): from brian2.devices.device import get_device self._clock = get_device().defaultclock #: Used to remember the `Network` in which this object has been included #: before, to raise an error if it is included in a new `Network` self._network = None #: The ID string determining when the object should be updated in ``. self.when = when #: The order in which objects with the same clock and ``when`` should be updated self.order = order self._dependencies = set() self._contained_objects = [] self._code_objects = [] self._active = True #: The scope key is used to determine which objects are collected by magic self._scope_key = self._scope_current_key # Make sure that keys in the namespace are valid if namespace is None: # Do not overwrite namespace if already set (e.g. in StateMonitor) namespace = getattr(self, 'namespace', {}) for key in namespace: if key.startswith('_'): raise ValueError("Names starting with underscores are " "reserved for internal use an cannot be " "defined in the namespace argument.") #: The group-specific namespace self.namespace = namespace logger.diagnostic(f"Created BrianObject with name {}, " f"clock={self._clock}, " f"when={self.when}, order={self.order}") #: Global key value for ipython cell restrict magic _scope_current_key = 0 #: Whether or not `MagicNetwork` is invalidated when a new `BrianObject` of this type is added invalidates_magic_network = True #: Whether or not the object should be added to a `MagicNetwork`. Note that #: all objects in `BrianObject.contained_objects` are automatically added #: when the parent object is added, therefore e.g. `NeuronGroup` should set #: `add_to_magic_network` to ``True``, but it should not be set for all the #: dependent objects such as `StateUpdater` add_to_magic_network = False
[docs] def add_dependency(self, obj): """ Add an object to the list of dependencies. Takes care of handling subgroups correctly (i.e., adds its parent object). Parameters ---------- obj : `BrianObject` The object that this object depends on. """ from brian2.groups.subgroup import Subgroup if isinstance(obj, Subgroup): self._dependencies.add( else: self._dependencies.add(
[docs] def before_run(self, run_namespace): """ Optional method to prepare the object before a run. Called by `Network.after_run` before the main simulation loop starts. """ for codeobj in self._code_objects: codeobj.before_run()
[docs] def after_run(self): """ Optional method to do work after a run is finished. Called by `Network.after_run` after the main simulation loop terminated. """ for codeobj in self._code_objects: codeobj.after_run()
[docs] def run(self): for codeobj in self._code_objects: codeobj()
contained_objects = property(fget=lambda self:self._contained_objects, doc=""" The list of objects contained within the `BrianObject`. When a `BrianObject` is added to a `Network`, its contained objects will be added as well. This allows for compound objects which contain a mini-network structure. Note that this attribute cannot be set directly, you need to modify the underlying list, e.g. ``obj.contained_objects.extend([A, B])``. """) code_objects = property(fget=lambda self:self._code_objects, doc=""" The list of `CodeObject` contained within the `BrianObject`. TODO: more details. Note that this attribute cannot be set directly, you need to modify the underlying list, e.g. ``obj.code_objects.extend([A, B])``. """) updaters = property(fget=lambda self:self._updaters, doc=""" The list of `Updater` that define the runtime behaviour of this object. TODO: more details. Note that this attribute cannot be set directly, you need to modify the underlying list, e.g. ``obj.updaters.extend([A, B])``. """) clock = property(fget=lambda self: self._clock, doc=""" The `Clock` determining when the object should be updated. Note that this cannot be changed after the object is created. """) def _set_active(self, val): val = bool(val) self._active = val for obj in self.contained_objects: = val active = property(fget=lambda self:self._active, fset=_set_active, doc=""" Whether or not the object should be run. Inactive objects will not have their `update` method called in ``. Note that setting or unsetting the `active` attribute will set or unset it for all `contained_objects`. """) def __repr__(self): classname = self.__class__.__name__ description = (f"{classname}(clock={self._clock}, when={self.when}, " f"order={self.order}, name={!r})") return description # This is a repeat from, but we want to get the documentation # here again name =
[docs]def weakproxy_with_fallback(obj): """ Attempts to create a `weakproxy` to the object, but falls back to the object if not possible. """ try: return weakref.proxy(obj) except TypeError: return obj
[docs]def device_override(name): """ Decorates a function/method to allow it to be overridden by the current `Device`. The ``name`` is the function name in the `Device` to use as an override if it exists. The returned function has an additional attribute ``original_function`` which is a reference to the original, undecorated function. """ def device_override_decorator(func): def device_override_decorated_function(*args, **kwds): from brian2.devices.device import get_device curdev = get_device() if hasattr(curdev, name): return getattr(curdev, name)(*args, **kwds) else: return func(*args, **kwds) device_override_decorated_function.original_function = func functools.update_wrapper(device_override_decorated_function, func) return device_override_decorated_function return device_override_decorator
[docs]class BrianObjectException(Exception): """ High level exception that adds extra Brian-specific information to exceptions This exception should only be raised at a fairly high level in Brian code to pass information back to the user. It adds extra information about where a `BrianObject` was defined to better enable users to locate the source of problems. Parameters ---------- message : str Additional error information to add to the original exception. brianobj : BrianObject The object that caused the error to happen. original_exception : Exception The original exception that was raised. """ def __init__(self, message, brianobj): self._brian_message = message self._brian_objname = self._brian_objcreate = brianobj._creation_stack logger.diagnostic(f"Error was encountered with object " f"'{self._brian_objname}':\n" f"{brianobj._full_creation_stack}") def __str__(self): return (f"Error encountered with object named '{self._brian_objname}'.\n" f"{self._brian_objcreate}\n\n" f"{self._brian_message} " f"(See above for original error message and traceback.)")
[docs]def brian_object_exception(message, brianobj, original_exception): """ Returns a `BrianObjectException` derived from the original exception. Creates a new class derived from the class of the original exception and `BrianObjectException`. This allows exception handling code to respond both to the original exception class and `BrianObjectException`. See `BrianObjectException` for arguments and notes. """ raise NotImplementedError("The brian_object_exception function is no longer used. " "Raise a BrianObjectException directly.")