Source code for brian2.core.base

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

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

__all__ = [

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 {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( "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} " "(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." )