Source code for spynnaker.pyNN.models.projection

# Copyright (c) 2017 The University of Manchester
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import functools
import logging
import numpy
from spinn_utilities.config_holder import get_config_bool
from spinn_utilities.log import FormatAdapter
from pyNN.recording.files import StandardTextFile
from import Space as PyNNSpace
from spinn_utilities.logger_utils import warn_once
from spinn_front_end_common.utilities.exceptions import ConfigurationException
from import SpynnakerDataView
from spynnaker.pyNN.utilities.constants import SPIKE_PARTITION_ID
from spynnaker.pyNN.models.abstract_models import (
from spynnaker.pyNN.models.neural_projections import (
    SynapseInformation, ProjectionApplicationEdge)
from spynnaker.pyNN.models.neural_projections.connectors import (
from spynnaker.pyNN.models.neuron import ConnectionHolder
from spynnaker.pyNN.models.neuron.synapse_dynamics import (
from spynnaker._version import __version__
from spynnaker.pyNN.models.populations import Population, PopulationView
from spynnaker.pyNN.models.neuron import AbstractPopulationVertex
from spynnaker.pyNN.models.spike_source import SpikeSourcePoissonVertex

logger = FormatAdapter(logging.getLogger(__name__))

def _we_dont_do_this_now(*args):  # pylint: disable=unused-argument
    # pragma: no cover
    raise NotImplementedError("sPyNNaker does not currently do this")

[docs] 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 parameters of those connections, including of plasticity mechanisms. """ # "format" param name defined by PyNN/ # pylint: disable=redefined-builtin __slots__ = [ "__projection_edge", "__synapse_information", "__virtual_connection_list", "__label"] def __init__( self, pre_synaptic_population, post_synaptic_population, connector, synapse_type=None, source=None, receptor_type=None, space=None, label=None): """ :param ~spynnaker.pyNN.models.populations.PopulationBase \ pre_synaptic_population: :param ~spynnaker.pyNN.models.populations.PopulationBase \ post_synaptic_population: :param AbstractConnector connector: :param AbstractSynapseDynamics synapse_type: :param None source: Unsupported; must be `None` :param str receptor_type: :param space: :param str label: """ # pylint: disable=too-many-arguments if source is not None: raise NotImplementedError( f"sPyNNaker {__version__} does not yet support " "multi-compartmental cells.") self.__projection_edge = None self.__label = label pre_is_view = self.__check_population(pre_synaptic_population) post_is_view = self.__check_population(post_synaptic_population) # set default label if label is None: # set the projection's label to a default (maybe non-unique!) self.__label = ( f"from pre {pre_synaptic_population.label} " f"to post {post_synaptic_population.label} " f"with connector {connector}") # give an auto generated label for the underlying edge label = "projection edge {}".format( SpynnakerDataView.get_next_none_labelled_edge_number()) # Handle default synapse type if synapse_type is None: synapse_dynamics = SynapseDynamicsStatic() else: synapse_dynamics = synapse_type # set the space function as required if space is None: space = PyNNSpace() connector.set_space(space) pre_vertex = pre_synaptic_population._vertex post_vertex = post_synaptic_population._vertex if not isinstance(post_vertex, AbstractAcceptsIncomingSynapses): raise ConfigurationException( "postsynaptic population is not designed to receive" " synaptic projections") # sort out synapse type synaptic_type = post_vertex.get_synapse_id_by_target(receptor_type) synapse_type_from_dynamics = False if synaptic_type is None: synaptic_type = synapse_dynamics.get_synapse_id_by_target( receptor_type) synapse_type_from_dynamics = True if synaptic_type is None: raise ConfigurationException( f"Synapse target {receptor_type} not found " f"in {post_synaptic_population.label}") # as a from-list connector can have plastic parameters, grab those ( # if any) and add them to the synapse dynamics object if isinstance(connector, FromListConnector): connector._apply_parameters_to_synapse_type(synapse_dynamics) # set the plasticity dynamics for the post pop (allows plastic stuff # when needed) post_vertex.set_synapse_dynamics(synapse_dynamics) # Set and store synapse information for future processing self.__synapse_information = SynapseInformation( connector, pre_synaptic_population, post_synaptic_population, pre_is_view, post_is_view, synapse_dynamics, synaptic_type, receptor_type, synapse_type_from_dynamics, synapse_dynamics.weight, synapse_dynamics.delay) # Set projection information in connector connector.set_projection_information(self.__synapse_information) # Find out if there is an existing edge between the populations edge_to_merge = self._find_existing_edge(pre_vertex, post_vertex) if edge_to_merge is not None: # If there is an existing edge, add the connector edge_to_merge.add_synapse_information(self.__synapse_information) self.__projection_edge = edge_to_merge else: # If there isn't an existing edge, create a new one and add it self.__projection_edge = ProjectionApplicationEdge( pre_vertex, post_vertex, self.__synapse_information, label=label) SpynnakerDataView.add_edge( self.__projection_edge, SPIKE_PARTITION_ID) # Ensure the connector is happy connector.validate_connection( self.__projection_edge, self.__synapse_information) # add projection to the SpiNNaker control system SpynnakerDataView.add_projection(self) # If there is a virtual board, we need to hold the data in case the # user asks for it self.__virtual_connection_list = None if get_config_bool("Machine", "virtual_board"): self.__virtual_connection_list = list() connection_holder = ConnectionHolder( None, False, pre_vertex.n_atoms, post_vertex.n_atoms, self.__virtual_connection_list) self.__synapse_information.add_pre_run_connection_holder( connection_holder) # If the target is a population, add to the list of incoming # projections if isinstance(post_vertex, AbstractPopulationVertex): post_vertex.add_incoming_projection(self) # If the source is a poisson, add to the list of outgoing projections if isinstance(pre_vertex, SpikeSourcePoissonVertex): pre_vertex.add_outgoing_projection(self) @staticmethod def __check_population(param): """ :param ~spynnaker.pyNN.models.populations.PopulationBase param: :return: Whether the parameter is a view :rtype: bool """ if isinstance(param, Population): # Projections definitely work from Populations return False if not isinstance(param, PopulationView): raise ConfigurationException( f"Unexpected parameter type {type(param)}. " "Expected Population") # Check whether the array is contiguous or not inds = param._indexes # pylint: disable=protected-access if inds != tuple(range(inds[0], inds[-1] + 1)): raise NotImplementedError( "Projections over views only work on contiguous arrays, " "e.g. view = pop[n:m], not view = pop[n,m]") # Projection is compatible with PopulationView return True
[docs] def get(self, attribute_names, format, # @ReservedAssignment gather=True, with_address=True, multiple_synapses='last'): """ Get a parameter/attribute of the projection. .. note:: SpiNNaker always gathers. :param attribute_names: list of attributes to gather :type attribute_names: str or iterable(str) :param str format: ``"list"`` or ``"array"`` :param bool gather: gather over all nodes :param bool with_address: True if the source and target are to be included :param str multiple_synapses: What to do with the data if format="array" and if the multiple source-target pairs with the same values exist. Currently only "last" is supported :return: values selected """ # pylint: disable=too-many-arguments if not gather: logger.warning("sPyNNaker always gathers from every core.") if multiple_synapses != 'last': raise ConfigurationException( "sPyNNaker only recognises multiple_synapses == last") return self.__get_data( attribute_names, format, with_address, notify=None)
[docs] def save( self, attribute_names, file, format='list', # @ReservedAssignment 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). .. note:: SpiNNaker always gathers. :param attribute_names: :type attribute_names: str or list(str) :param file: filename or open handle (which will be closed) :type file: str or pyNN.recording.files.BaseFile :param str format: :param bool gather: Ignored :param bool with_address: """ # pylint: disable=too-many-arguments if not gather: warn_once( logger, "sPyNNaker only supports gather=True. We will run " "as if gather was set to True.") if isinstance(attribute_names, str): attribute_names = [attribute_names] if attribute_names in (['all'], ['connections']): attribute_names = \ self._projection_edge.post_vertex.synapse_dynamics.\ get_parameter_names() metadata = {"columns": attribute_names} if with_address: metadata["columns"] = ["i", "j"] + list(metadata["columns"]) self.__get_data( attribute_names, format, with_address, notify=functools.partial(self.__save_callback, file, metadata))
def __get_data( self, attribute_names, format, # @ReservedAssignment with_address, notify): """ Internal data getter to add notify option. :param attribute_names: list of attributes to gather :type attribute_names: str or iterable(str) :param str format: ``"list"`` or ``"array"`` :param bool with_address: :param callable(ConnectionHolder,None) notify: :return: values selected """ # fix issue with 1 versus many if isinstance(attribute_names, str): attribute_names = [attribute_names] data_items = list() if format != "list": with_address = False if with_address: data_items.append("source") data_items.append("target") if "source" in attribute_names: logger.warning( "Ignoring request to get source as with_address=True. ") attribute_names.remove("source") if "target" in attribute_names: logger.warning( "Ignoring request to get target as with_address=True. ") attribute_names.remove("target") # Split out attributes in to standard versus synapse dynamics data fixed_values = list() for attribute in attribute_names: data_items.append(attribute) if attribute not in {"source", "target", "weight", "delay"}: value = self._synapse_information.synapse_dynamics.get_value( attribute) fixed_values.append((attribute, value)) # Return the connection data return self._get_synaptic_data( format == "list", data_items, fixed_values, notify=notify) @staticmethod def __save_callback(save_file, metadata, data): """ :param save_file: :type save_file: str or pyNN.recording.files.BaseFile :param dict(str,object) metadata: :param data: :type data: ConnectionHolder or numpy.ndarray """ # Convert structured array to normal numpy array if hasattr(data, "dtype") and hasattr(data.dtype, "names"): dtype = [(name, "<f8") for name in data.dtype.names] data = data.astype(dtype) data = numpy.nan_to_num(data) if isinstance(save_file, str): data_file = StandardTextFile(save_file, mode='wb') else: data_file = save_file try: data_file.write(data, metadata) finally: data_file.close() @property def pre(self): """ The pre-population or population view. :rtype: ~spynnaker.pyNN.models.populations.PopulationBase """ return self._synapse_information.pre_population @property def post(self): """ The post-population or population view. :rtype: ~spynnaker.pyNN.models.populations.PopulationBase """ return self._synapse_information.post_population @property def label(self): """ :rtype: str """ return self.__label def __repr__(self): return f"projection {self.__label}" # ----------------------------------------------------------------- @property def _synapse_information(self): """ :rtype: SynapseInformation """ return self.__synapse_information @property def _projection_edge(self): """ :rtype: ProjectionApplicationEdge """ return self.__projection_edge def _find_existing_edge(self, pre_synaptic_vertex, post_synaptic_vertex): """ Searches though the graph's edges to locate any edge which has the same post- and pre- vertex :param pre_synaptic_vertex: the source vertex of the multapse :type pre_synaptic_vertex: ~pacman.model.graphs.application.ApplicationVertex :param post_synaptic_vertex: The destination vertex of the multapse :type post_synaptic_vertex: ~pacman.model.graphs.application.ApplicationVertex :return: `None` or the edge going to these vertices. :rtype: ~.ApplicationEdge """ # Find edges ending at the postsynaptic vertex partitions = ( SpynnakerDataView.get_outgoing_edge_partitions_starting_at_vertex( pre_synaptic_vertex)) # Partitions and Partition.edges will be OrderedSet but may be empty for partition in partitions: for edge in partition.edges: if edge.post_vertex == post_synaptic_vertex: return edge return None def _get_synaptic_data( self, as_list, data_to_get, fixed_values=None, notify=None): """ :param bool as_list: :param list(int) data_to_get: :param list(tuple(str,int)) fixed_values: :param callable(ConnectionHolder,None) notify: :rtype: ConnectionHolder """ # pylint: disable=too-many-arguments post_vertex = self.__projection_edge.post_vertex pre_vertex = self.__projection_edge.pre_vertex # If in virtual board mode, the connection data should be set if self.__virtual_connection_list is not None: connection_holder = ConnectionHolder( data_to_get, as_list, pre_vertex.n_atoms, post_vertex.n_atoms, self.__virtual_connection_list, fixed_values=fixed_values, notify=notify) connection_holder.finish() return connection_holder # if not virtual board, make connection holder to be filled in at # possible later date connection_holder = ConnectionHolder( data_to_get, as_list, pre_vertex.n_atoms, post_vertex.n_atoms, fixed_values=fixed_values, notify=notify) # If we haven't run, add the holder to get connections, and return it # and set up a callback for after run to fill in this connection holder if not SpynnakerDataView.is_ran_ever(): self.__synapse_information.add_pre_run_connection_holder( connection_holder) return connection_holder # Otherwise, get the connections now, as we have ran and therefore can # get them connections = post_vertex.get_connections_from_machine( self.__projection_edge, self.__synapse_information) if connections is not None: connection_holder.add_connections(connections) connection_holder.finish() return connection_holder def _clear_cache(self): post_vertex = self.__projection_edge.post_vertex if isinstance(post_vertex, AbstractAcceptsIncomingSynapses): post_vertex.clear_connection_cache() # -----------------------------------------------------------------
[docs] def set(self, **attributes): # @UnusedVariable # pylint: disable=unused-argument """ .. warning:: Not implemented. """ _we_dont_do_this_now()
[docs] def size(self, gather=True): # @UnusedVariable # pylint: disable=unused-argument """ Return the total number of connections. .. note:: SpiNNaker always gathers. .. warning:: Not implemented. :param bool gather: If False, only get the number of connections locally. """ # TODO _we_dont_do_this_now()