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
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
import functools
import logging
from typing import (
    Any, Callable, Dict, List, Optional, Sequence, Tuple, Union,
    cast, TYPE_CHECKING)

import numpy
from numpy import void
from numpy.typing import NDArray
from typing_extensions import Literal, TypeAlias

from pyNN.recording.files import BaseFile
from pyNN.space import Space as PyNNSpace

from spinn_utilities.config_holder import get_config_bool
from spinn_utilities.log import FormatAdapter
from spinn_utilities.logger_utils import warn_once

from pacman.model.graphs.application import ApplicationVertex

from spinn_front_end_common.utilities.exceptions import ConfigurationException

from spynnaker.pyNN.data import SpynnakerDataView
from spynnaker.pyNN.models.abstract_models import (
    AbstractAcceptsIncomingSynapses)
from spynnaker.pyNN.models.neural_projections import (
    SynapseInformation, ProjectionApplicationEdge)
from spynnaker.pyNN.models.neural_projections.connectors import (
    FromListConnector)
from spynnaker.pyNN.models.neuron import (
    AbstractPopulationVertex, ConnectionHolder)
from spynnaker.pyNN.models.populations import Population, PopulationView
from spynnaker.pyNN.models.neuron.synapse_dynamics import (
    SynapseDynamicsStatic, AbstractHasParameterNames)
from spynnaker.pyNN.models.spike_source import SpikeSourcePoissonVertex
from spynnaker._version import __version__

if TYPE_CHECKING:
    from spynnaker.pyNN.models.neural_projections.connectors import (
        AbstractConnector)
    from spynnaker.pyNN.models.neuron.synapse_dynamics import (
        AbstractSynapseDynamics)
    _Pop: TypeAlias = Union[Population, PopulationView]

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: _Pop, post_synaptic_population: _Pop, connector: AbstractConnector, synapse_type: Optional[AbstractSynapseDynamics] = None, source: None = None, receptor_type: str = "excitatory", space: Optional[PyNNSpace] = None, label: Optional[str] = None, download_synapses: bool = False, partition_id: Optional[str] = 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 ~pyNN.space.Space space: :param str label: :param bool download_synapses: """ # pylint: disable=too-many-arguments if source is not None: raise NotImplementedError( f"sPyNNaker {__version__} does not yet support " "multi-compartmental cells.") 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 = f"projection edge " \ f"{SpynnakerDataView.get_next_none_labelled_edge_number()}" else: self.__label = label # Handle default synapse type if synapse_type is None: synapse_dynamics: AbstractSynapseDynamics = 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 = cast(AbstractPopulationVertex, 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 synapse_id = post_vertex.get_synapse_id_by_target(receptor_type) synapse_id_from_dynamics = False if synapse_id is None: synapse_id = synapse_dynamics.get_synapse_id_by_target( receptor_type) synapse_id_from_dynamics = True if synapse_id 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, synapse_id, receptor_type, synapse_id_from_dynamics, synapse_dynamics.weight, synapse_dynamics.delay, download_synapses, partition_id) # 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, self.__synapse_information.partition_id) 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, self.__synapse_information.partition_id) # Ensure the connector is happy synapse_dynamics.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: Optional[List[NDArray[void]]] = None if get_config_bool("Machine", "virtual_board"): self.__virtual_connection_list = list() self.__synapse_information.add_pre_run_connection_holder( ConnectionHolder( None, False, pre_vertex.n_atoms, post_vertex.n_atoms, self.__virtual_connection_list)) # 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: _Pop) -> bool: """ :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 if not param._is_contiguous: # pylint: disable=protected-access 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: Union[str, Sequence[str]], format: str, # @ReservedAssignment gather: Literal[True] = True, with_address: bool = True, multiple_synapses: Literal['last'] = '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") an = [attribute_names] if isinstance(attribute_names, str) else list( attribute_names) return self.__get_data(an, format, with_address, notify=None)
[docs] def save( self, attribute_names: Union[str, Sequence[str]], file: Union[str, BaseFile], format: str = 'list', # @ReservedAssignment gather: Literal[True] = True, with_address: bool = 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] else: attribute_names = list(attribute_names) if len(attribute_names) == 1 and attribute_names[0] in { 'all', 'connections'}: sd = self._projection_edge.post_vertex.synapse_dynamics if isinstance(sd, AbstractHasParameterNames): attribute_names = list(sd.get_parameter_names()) else: attribute_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: List[str], format: str, # @ReservedAssignment with_address: bool, notify: Optional[Callable[[ConnectionHolder], None]]): """ 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[str] = 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[Tuple[str, int]] = 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: Union[str, BaseFile], metadata: Dict[str, Any], data: ConnectionHolder): """ :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) npdata = numpy.nan_to_num(cast(NDArray, data)) if isinstance(save_file, str): data_file = open(save_file, mode='wb') else: data_file = save_file try: header_lines = [ f"# {key} = {value}" for key, value in metadata.items()] header = "\n".join(header_lines) + '\n' data_file.write(header.encode('utf-8')) # write data numpy.savetxt(data_file, npdata, delimiter='\t') data_file.close() finally: data_file.close() @property def pre(self) -> _Pop: """ The pre-population or population view. :rtype: ~spynnaker.pyNN.models.populations.PopulationBase """ return self._synapse_information.pre_population @property def post(self) -> _Pop: """ The post-population or population view. :rtype: ~spynnaker.pyNN.models.populations.PopulationBase """ return self._synapse_information.post_population @property def label(self) -> str: """ :rtype: str """ return self.__label def __repr__(self): return f"projection {self.__label}" # ----------------------------------------------------------------- @property def _synapse_information(self) -> SynapseInformation: """ :rtype: SynapseInformation """ return self.__synapse_information @property def _projection_edge(self) -> ProjectionApplicationEdge: """ :rtype: ProjectionApplicationEdge """ return self.__projection_edge def _find_existing_edge( self, pre_synaptic_vertex: ApplicationVertex, post_synaptic_vertex: ApplicationVertex, partition_id: str) -> Optional[ProjectionApplicationEdge]: """ 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 :param str partition_id: The partition ID of the edge to find :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: if partition.identifier != partition_id: continue for edge in partition.edges: if edge.post_vertex == post_synaptic_vertex: return edge return None def _get_synaptic_data( self, as_list: bool, data_to_get: List[str], fixed_values: List[Tuple[str, int]], notify: Optional[Callable[[ConnectionHolder], None]]): """ :param bool as_list: :param list(str) data_to_get: :param list(tuple(str,int)) fixed_values: :param callable(ConnectionHolder,None) notify: :rtype: ConnectionHolder """ 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) -> None: 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()
[docs] def set_download_synapses(self, download_synapses): """ Set whether synapses should be downloaded when the simulation pauses. :param bool download_synapses: Whether to download synapses or not """ self.__synapse_information.download_on_pause = download_synapses