# encoding: utf-8
"""
Common implementation of the Projection class, to be sub-classed by
backend-specific Projection classes.
:copyright: Copyright 2006-2024 by the PyNN team, see AUTHORS.
:license: CeCILL, see LICENSE for details.
"""
from functools import reduce
import logging
import operator
from copy import deepcopy
from warnings import warn
import numpy as np
from .. import recording, errors, models, core, descriptions
from ..parameters import ParameterSpace, LazyArray
from ..space import Space
from ..standardmodels import StandardSynapseType
from ..connectors import Connector
from .populations import BasePopulation, Assembly
logger = logging.getLogger("PyNN")
deprecated = core.deprecated
class Projection(object):
"""
A container for all the connections of a given type (same synapse type and
plasticity mechanisms) between two populations, together with methods to
set the parameters of those connections, including the parameters of
plasticity mechanisms.
Arguments:
`presynaptic_neurons` and `postsynaptic_neurons`:
Population, PopulationView or Assembly objects.
`source`:
string specifying which attribute of the presynaptic cell signals
action potentials. This is only needed for multicompartmental cells
with branching axons or dendrodendritic synapses. All standard cells
have a single source, and this is the default.
`receptor_type`:
string specifying which synaptic receptor_type type on the postsynaptic cell to connect
to. For standard cells, this can be 'excitatory' or 'inhibitory'.
For non-standard cells, it could be 'NMDA', etc. If receptor_type is not
given, the default values of 'excitatory' is used.
`connector`:
a Connector object, encapsulating the algorithm to use for
connecting the neurons.
`synapse_type`:
a SynapseType object specifying which synaptic connection
mechanisms to use.
`space`:
TO DOCUMENT
"""
_nProj = 0
MULTI_SYNAPSE_OPERATIONS = {
'last': lambda a, b: b,
'first': lambda a, b: a,
'sum': operator.iadd,
'min': min,
'max': max
}
def __init__(self, presynaptic_neurons, postsynaptic_neurons, connector,
synapse_type=None, source=None, receptor_type=None,
space=Space(), label=None):
"""
Create a new projection, connecting the pre- and post-synaptic neurons.
"""
if not hasattr(self, "_simulator"):
err_msg = "`common.Projection` should not be instantiated directly. " \
"You should import Projection from a PyNN backend module, " \
"e.g. pyNN.nest or pyNN.neuron"
raise Exception(err_msg)
for prefix, pop in zip(("pre", "post"),
(presynaptic_neurons, postsynaptic_neurons)):
if not isinstance(pop, (BasePopulation, Assembly)):
raise errors.ConnectionError(
f"{prefix}synaptic_neurons must be a Population, PopulationView or Assembly, "
f"not a {type(pop)}")
if isinstance(postsynaptic_neurons, Assembly):
if not postsynaptic_neurons._homogeneous_synapses:
raise errors.ConnectionError(
"Projection to an Assembly object can be made only "
"with homogeneous synapses types")
self.pre = presynaptic_neurons # } these really
self.source = source # } should be
self.post = postsynaptic_neurons # } read-only
self.label = label
self.space = space
if not isinstance(connector, Connector):
raise TypeError(
"`connector` should be an instance of a subclass of Connector. "
f"The argument provided was of type '{type(connector).__name__}'."
)
self._connector = connector
self.synapse_type = synapse_type or self._static_synapse_class()
if not isinstance(self.synapse_type, models.BaseSynapseType):
raise TypeError(
"`synapse_type` should be an instance of a subclass of BaseSynapseType. "
f"The argument provided was of type '{type(synapse_type).__name__}'"
)
self.receptor_type = receptor_type
if self.receptor_type in ("default", None):
self._guess_receptor_type()
if self.receptor_type not in postsynaptic_neurons.receptor_types:
valid_types = postsynaptic_neurons.receptor_types
assert len(valid_types) > 0
err_msg = "User gave receptor_types=%s, receptor_types must be one of: '%s'"
raise errors.ConnectionError(err_msg % (self.receptor_type, "', '".join(valid_types)))
if label is None:
if self.pre.label and self.post.label:
self.label = u"%s→%s" % (self.pre.label, self.post.label)
self.initial_values = {}
self.annotations = {}
Projection._nProj += 1
def _guess_receptor_type(self):
"""
If the receptor_type is not specified, we follow the convention that the first element
in the list of available post-synaptic receptor types is the default for excitatory
synapses and the second element is the default for inhibitory synapses.
"""
if len(self.post.receptor_types) > 1:
ps = deepcopy(self.synapse_type.parameter_space)
ps = self._handle_distance_expressions(ps)
weights = ps["weight"]
if weights.shape is None:
weights.shape = self.shape
try:
wl = weights[self.pre.size - 1, self.post.size - 1]
if wl >= 0:
self.receptor_type = self.post.receptor_types[0]
else:
self.receptor_type = self.post.receptor_types[1]
except TypeError: # for example, if using a native RNG with no Python interface
warn("Unable to guess receptor type")
self.receptor_type = self.post.receptor_types[0]
else:
self.receptor_type = self.post.receptor_types[0]
def __len__(self):
"""Return the total number of local connections."""
raise NotImplementedError
[docs]
def size(self, gather=True):
"""
Return the total number of connections.
- only local connections, if gather is False,
- all connections, if gather is True (default)
"""
if gather and self._simulator.state.num_processes > 1:
n = len(self)
return recording.mpi_sum(n)
else:
return len(self)
@property
def shape(self):
return (self.pre.size, self.post.size)
def __repr__(self):
return 'Projection("%s")' % self.label
def __getitem__(self, i):
"""Return the /i/th connection within the Projection."""
raise NotImplementedError
[docs]
def __iter__(self):
"""Return an iterator over all connections on the local MPI node."""
for i in range(len(self)):
yield self[i]
# --- Methods for setting connection parameters ---------------------------
[docs]
def set(self, **attributes):
"""
Set connection attributes for all connections on the local MPI node.
Attribute names may be 'weight', 'delay', or the name of any parameter
of a synapse dynamics model (e.g. 'U' for TsodyksMarkramSynapse).
Each attribute value may be:
(1) a single number
(2) a RandomDistribution object
(3) a 2D array with the same dimensions as the connectivity matrix
(as returned by `get(format='array')`
(4) a mapping function, which accepts a single float argument (the
distance between pre- and post-synaptic cells) and returns a single value.
Weights should be in nA for current-based and µS for conductance-based
synapses. Delays should be in milliseconds.
Note that where a projection contains multiple connections between a given pair
of neurons, all these connections will be set to the same value.
"""
# should perhaps add a "distribute" argument, for symmetry with "gather" in get()
# Note: we have removed the option:
# "a list/1D array of the same length as the number of local connections"
# because it was proving tricky to implement and was holding up the release.
# The plan is to add this option back at a later date.
attributes = self._value_list_to_array(attributes)
parameter_space = ParameterSpace(attributes,
self.synapse_type.get_schema(),
(self.pre.size, self.post.size))
parameter_space = self._handle_distance_expressions(parameter_space)
if isinstance(self.synapse_type, StandardSynapseType):
parameter_space = self.synapse_type.translate(parameter_space)
self._set_attributes(parameter_space)
[docs]
def initialize(self, **initial_values):
"""
Set initial values of state variables of synaptic plasticity models.
Values passed to initialize() may be:
(1) single numeric values (all neurons set to the same value)
(2) RandomDistribution objects
(3) a 2D array with the same dimensions as the connectivity matrix
(as returned by `get(format='array')`
(4) a mapping function, which accepts a single float argument (the
distance between pre- and post-synaptic cells) and returns a single value.
Values should be expressed in the standard PyNN units (i.e. millivolts,
nanoamps, milliseconds, microsiemens, nanofarads, event per second).
Example::
prj.initialize(u=-70.0)
"""
for variable, value in initial_values.items():
logger.debug("In Projection '%s', initialising %s to %s" %
(self.label, variable, value))
initial_value = LazyArray(value, shape=(self.size,), dtype=float)
self._set_initial_value_array(variable, initial_value)
self.initial_values[variable] = initial_value
def _value_list_to_array(self, attributes):
"""Convert a list of connection parameters/attributes to a 2D array."""
connection_mask = ~np.isnan(self.get('weight', format='array', gather='all'))
for name, value in attributes.items():
if isinstance(value, list) or (isinstance(value, np.ndarray) and value.ndim == 1):
array_value = np.nan * np.ones(self.shape)
array_value[connection_mask] = value
attributes[name] = array_value
return attributes
def _handle_distance_expressions(self, parameter_space):
# also index-based expressions
for name, map in parameter_space.items():
if callable(map.base_value):
if isinstance(map.base_value, core.IndexBasedExpression):
map.base_value.projection = self
parameter_space[name] = map
else:
# Assumes map is a function of distance
position_generators = (self.pre.position_generator,
self.post.position_generator)
distance_map = LazyArray(self.space.distance_generator(*position_generators),
shape=self.shape)
parameter_space[name] = map(distance_map)
return parameter_space
@deprecated("set(weight=w)")
def setWeights(self, w):
self.set(weight=w)
@deprecated("set(weight=rand_distr)")
def randomizeWeights(self, rand_distr):
self.set(weight=rand_distr)
@deprecated("set(delay=d)")
def setDelays(self, d):
self.set(delay=d)
@deprecated("set(delay=rand_distr)")
def randomizeDelays(self, rand_distr):
self.set(delay=rand_distr)
@deprecated("set(parameter_name=value)")
def setSynapseDynamics(self, parameter_name, value):
self.set(parameter_name=value)
@deprecated("set(name=rand_distr)")
def randomizeSynapseDynamics(self, parameter_name, rand_distr):
self.set(parameter_name=rand_distr)
# --- Methods for writing/reading information to/from file. ---------------
[docs]
def get(self, attribute_names, format, gather=True, with_address=True,
multiple_synapses='sum'):
"""
Get the values of a given attribute (weight or delay) for all
connections in this Projection.
`attribute_names`:
name of the attributes whose values are wanted, or a list of such
names.
`format`:
"list" or "array".
`gather`:
If True, node 0 gets connection information from all MPI nodes,
other nodes get information only from connections that exist in this node.
If 'all', all nodes will receive connection information from all other nodes.
If False, all nodes get only information about local connections.
With list format, returns a list of tuples. By default, each tuple
contains the indices of the pre- and post-synaptic cell followed by
the attribute values in the order given in `attribute_names`.
Example::
>>> prj.get(["weight", "delay"], format="list")[:5]
[(0.0, 0.0, 0.3401892507507171, 0.1),
(0.0, 1.0, 0.7990713166233654, 0.30000000000000004),
(0.0, 2.0, 0.6180841812877726, 0.5),
(0.0, 3.0, 0.6758149775627305, 0.7000000000000001),
(0.0, 4.0, 0.7166906726862953, 0.9)]
If `with_address` is set to False, then the tuples will contain only the
attribute values, not the cell indices.
With array format, returns a tuple of 2D NumPy arrays, one for each
name in `attribute_names`. The array element X_ij contains the
attribute value for the connection from the ith neuron in the pre-
synaptic Population to the jth neuron in the post-synaptic Population,
if a single such connection exists. If there are no such connections,
X_ij will be NaN. Example::
>>> weights, delays = prj.get(["weight", "delay"], format="array")
>>> weights
array([[ 0.66210438, nan, 0.10744555, 0.54557088],
[ 0.3676134 , nan, 0.41463193, nan],
[ 0.57434871, 0.4329354 , 0.58482943, 0.42863916]])
If there are multiple such connections, the action to take is
controlled by the `multiple_synapses` argument, which must be one of
{'last', 'first', 'sum', 'min', 'max'}.
Values will be expressed in the standard PyNN units (i.e. millivolts,
nanoamps, milliseconds, microsiemens, nanofarads, event per second).
"""
if isinstance(attribute_names, str):
attribute_names = (attribute_names,)
return_single = True
else:
return_single = False
if isinstance(self.synapse_type, StandardSynapseType):
attribute_names = self.synapse_type.get_native_names(*attribute_names)
if format == 'list':
names = list(attribute_names)
if with_address:
names = ["presynaptic_index", "postsynaptic_index"] + names
values = self._get_attributes_as_list(names)
if gather and self._simulator.state.num_processes > 1:
all_values = {self._simulator.state.mpi_rank: values}
all_values = recording.gather_dict(all_values, all=(gather == 'all'))
if gather == 'all' or self._simulator.state.mpi_rank == 0:
values = reduce(operator.add, all_values.values())
if not with_address and return_single:
values = [val[0] for val in values]
return values
elif format == 'array':
if multiple_synapses not in Projection.MULTI_SYNAPSE_OPERATIONS:
raise ValueError("`multiple_synapses` argument must be one of {}".format(
list(Projection.MULTI_SYNAPSE_OPERATIONS)))
if gather and self._simulator.state.num_processes > 1:
# Node 0 is the only one creating a full connection matrix, and returning it
# (saving memory)
# Slaves nodes are returning list of connections, so this may be inconsistent...
names = list(attribute_names)
names = ["presynaptic_index", "postsynaptic_index"] + names
values = self._get_attributes_as_list(names)
all_values = {self._simulator.state.mpi_rank: values}
all_values = recording.gather_dict(all_values, all=(gather == 'all'))
if gather == 'all' or self._simulator.state.mpi_rank == 0:
tmp_values = reduce(operator.add, all_values.values())
values = self._get_attributes_as_arrays(attribute_names,
multiple_synapses=multiple_synapses)
tmp_values = np.array(tmp_values)
for i in range(len(values)):
values[i][tmp_values[:, 0].astype(
int), tmp_values[:, 1].astype(int)] = tmp_values[:, 2 + i]
else:
values = self._get_attributes_as_arrays(attribute_names,
multiple_synapses=multiple_synapses)
if return_single:
if gather == 'all' or self._simulator.state.mpi_rank == 0:
assert len(values) == 1, values
return values[0]
else:
return values
else:
raise Exception("format must be 'list' or 'array'")
def _get_attributes_as_list(self, names):
return [c.as_tuple(*names) for c in self.connections]
def _get_attributes_as_arrays(self, names, multiple_synapses='sum'):
multi_synapse_operation = Projection.MULTI_SYNAPSE_OPERATIONS[multiple_synapses]
all_values = []
for attribute_name in names:
values = np.nan * np.ones((self.pre.size, self.post.size))
if attribute_name[-1] == "s": # weights --> weight, delays --> delay
attribute_name = attribute_name[:-1]
for c in self.connections:
value = getattr(c, attribute_name)
addr = (c.presynaptic_index, c.postsynaptic_index)
if np.isnan(values[addr]):
values[addr] = value
else:
values[addr] = multi_synapse_operation(values[addr], value)
all_values.append(values)
return all_values
@deprecated("get('weight', format, gather)")
def getWeights(self, format='list', gather=True):
return self.get('weight', format, gather, with_address=False)
@deprecated("get('delay', format, gather)")
def getDelays(self, format='list', gather=True):
return self.get('delay', format, gather, with_address=False)
@deprecated("get(parameter_name, format, gather)")
def getSynapseDynamics(self, parameter_name, format='list', gather=True):
return self.get(parameter_name, format, gather, with_address=False)
[docs]
def save(self, attribute_names, file, format='list', gather=True, with_address=True):
"""
Print synaptic attributes (weights, delays, etc.) to file. In the array
format, zeros are printed for non-existent connections.
Values will be expressed in the standard PyNN units (i.e. millivolts,
nanoamps, milliseconds, microsiemens, nanofarads, event per second).
"""
if attribute_names in ('all', 'connections'):
attribute_names = self.synapse_type.get_parameter_names()
if isinstance(file, str):
file = recording.files.StandardTextFile(file, mode='wb')
all_values = self.get(attribute_names, format=format,
gather=gather, with_address=with_address)
if format == 'array':
all_values = [np.where(np.isnan(values), 0.0, values)
for values in all_values]
if self._simulator.state.mpi_rank == 0:
metadata = {"columns": attribute_names}
if with_address:
metadata["columns"] = ["i", "j"] + list(metadata["columns"])
file.write(all_values, metadata)
file.close()
@deprecated("save('all', file, format='list', gather=gather)")
def saveConnections(self, file, gather=True, compatible_output=True):
self.save('all', file, format='list', gather=gather)
@deprecated("save('weight', file, format, gather)")
def printWeights(self, file, format='list', gather=True):
self.save('weight', file, format, gather)
@deprecated("save('delay', file, format, gather)")
def printDelays(self, file, format='list', gather=True):
"""
Print synaptic weights to file. In the array format, zeros are printed
for non-existent connections.
"""
self.save('delay', file, format, gather)
@deprecated("np.histogram()")
def weightHistogram(self, min=None, max=None, nbins=10):
"""
Return a histogram of synaptic weights.
If min and max are not given, the minimum and maximum weights are
calculated automatically.
"""
weights = np.array(self.get('weight', format='list', gather=True, with_address=False))
if min is None:
min = weights.min()
if max is None:
max = weights.max()
bins = np.linspace(min, max, nbins + 1)
return np.histogram(weights, bins) # returns n, bins
[docs]
def annotate(self, **annotations):
self.annotations.update(annotations)
[docs]
def describe(self, template='projection_default.txt', engine='default'):
"""
Returns a human-readable description of the projection.
The output may be customized by specifying a different template
togther with an associated template engine (see ``pyNN.descriptions``).
If template is None, then a dictionary containing the template context
will be returned.
"""
context = {
"label": self.label,
"pre": self.pre.describe(template=None),
"post": self.post.describe(template=None),
"source": self.source,
"receptor_type": self.receptor_type,
"size_local": len(self),
"size": self.size(gather=True),
"connector": self._connector.describe(template=None),
"plasticity": None,
}
if self.synapse_type:
context.update(plasticity=self.synapse_type.describe(template=None))
return descriptions.render(engine, template, context)
class Connection(object):
"""
Store an individual plastic connection and information about it. Provide an
interface that allows access to the connection's weight, delay and other
attributes.
"""
pass
