Source code for spynnaker.pyNN.models.neuron.synapse_dynamics.synapse_dynamics_structural_common

# Copyright (c) 2016 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.

from import Iterable
import numpy
from spinn_utilities.abstract_base import (
    AbstractBase, abstractmethod, abstractproperty)
from spinn_utilities.overrides import overrides
from spinn_front_end_common.interface.ds import DataType
from spinn_front_end_common.utilities.constants import (
from import SpynnakerDataView
from .abstract_synapse_dynamics_structural import (
from spynnaker.pyNN.exceptions import SynapticConfigurationException
from spynnaker.pyNN.utilities.constants import SPIKE_PARTITION_ID

#: Default value for frequency of rewiring
DEFAULT_F_REW = 10 ** 4.0
#: Default value for initial weight on connection formation
#: Default value for initial delay on connection formation
#: Default value for maximum fan-in per target layer neuron

class SynapseDynamicsStructuralCommon(
        AbstractSynapseDynamicsStructural, metaclass=AbstractBase):

    # 8 32-bit numbers (fast; p_rew; s_max; app_no_atoms; machine_no_atoms;
    # low_atom; high_atom; with_replacement) + 2 4-word RNG seeds (shared_seed;
    # local_seed) + 1 32-bit number (no_pre_pops)
        (8 * BYTES_PER_WORD) + (2 * 4 * BYTES_PER_WORD) + BYTES_PER_WORD)

    # Size excluding key_atom_info (as variable length)
    # 4 16-bit numbers (no_pre_vertices; sp_control; delay_lo; delay_hi)
    # + 3 32-bit numbers (weight; connection_type; total_no_atoms)

    # 6 32-bit numbers (key; mask; n_atoms; n_colour_bits; lo_atom;
    # m_pop_index)

    # 1 16-bit number (neuron_index)
    # + 2 8-bit numbers (sub_pop_index; pop_index)

    PAIR_ERROR = (
        "Only one Projection between each pair of Populations can use "
        "structural plasticity")

[docs] def get_parameter_names(self): """ :rtype: list(str) """ names = ['initial_weight', 'initial_delay', 'f_rew', 'p_rew', 's_max', 'with_replacement'] # pylint: disable=no-member names.extend(self.partner_selection.get_parameter_names()) names.extend(self.formation.get_parameter_names()) names.extend(self.elimination.get_parameter_names()) return names
@property def p_rew(self): """ The period of rewiring. :rtype: float """ return 1. / self.f_rew
[docs] @overrides(AbstractSynapseDynamicsStructural.write_structural_parameters) def write_structural_parameters( self, spec, region, weight_scales, app_vertex, vertex_slice, synaptic_matrices): spec.comment("Writing structural plasticity parameters") spec.switch_write_focus(region) # Get relevant edges structural_projections = self.__get_structural_projections( app_vertex.incoming_projections) # Write the common part of the rewiring data self.__write_common_rewiring_data( spec, app_vertex, vertex_slice, len(structural_projections)) # Write the pre-population info pop_index, subpop_index, lo_atom_index = \ self.__write_prepopulation_info( spec, app_vertex, structural_projections, weight_scales, synaptic_matrices) # Write the post-to-pre table self.__write_post_to_pre_table( spec, pop_index, subpop_index, lo_atom_index, app_vertex, vertex_slice) # Write the component parameters # pylint: disable=no-member, protected-access spec.comment("Writing partner selection parameters") self.partner_selection.write_parameters(spec) for proj in structural_projections: spec.comment(f"Writing formation parameters for {proj.label}") dynamics = proj._synapse_information.synapse_dynamics dynamics.formation.write_parameters(spec) for proj in structural_projections: spec.comment(f"Writing elimination parameters for {proj.label}") dynamics = proj._synapse_information.synapse_dynamics dynamics.elimination.write_parameters( spec, weight_scales[proj._synapse_information.synapse_type])
def __get_structural_projections(self, incoming_projections): """ :param list(Projection) incoming_projections: Projections to filter to structural only :rtype: list(Projection) """ structural_projections = list() seen_app_edges = set() for proj in incoming_projections: # pylint: disable=protected-access app_edge = proj._projection_edge for synapse_info in app_edge.synapse_information: if isinstance(synapse_info.synapse_dynamics, AbstractSynapseDynamicsStructural): if app_edge in seen_app_edges: raise SynapticConfigurationException( self.PAIR_ERROR) else: seen_app_edges.add(app_edge) structural_projections.append(proj) return structural_projections def __write_common_rewiring_data( self, spec, app_vertex, vertex_slice, n_pre_pops): """ Write the non-sub-population synapse parameters to the spec. :param ~data_specification.DataSpecificationGenerator spec: the data spec :param ~pacman.model.graphs.application.ApplicationVertex app_vertex: The application vertex being generated :param ~pacman.model.graphs.common.Slice vertex_slice: The slice of the target vertex to generate for :param int n_pre_pops: the number of pre-populations """ time_step_us = SpynnakerDataView.get_simulation_time_step_us() spec.comment("Writing common rewiring data") if (self.p_rew * MICRO_TO_MILLISECOND_CONVERSION < time_step_us / MICRO_TO_MILLISECOND_CONVERSION): # Fast rewiring spec.write_value(data=1) spec.write_value(data=int( time_step_us / ( self.p_rew * MICRO_TO_SECOND_CONVERSION))) else: # Slow rewiring spec.write_value(data=0) spec.write_value(data=int(( self.p_rew * MICRO_TO_SECOND_CONVERSION) / time_step_us)) # write s_max spec.write_value(data=int(self.s_max)) # write total number of atoms in the application vertex spec.write_value(data=app_vertex.n_atoms) # write local low, high and number of atoms spec.write_value(data=vertex_slice.n_atoms) spec.write_value(data=vertex_slice.lo_atom) spec.write_value(data=vertex_slice.hi_atom) # write with_replacement spec.write_value(data=self.with_replacement) # write app level seeds spec.write_array(self.get_seeds(app_vertex)) # write local seed (4 words), generated randomly! # Note that in case of a reset, these need a key to ensure subsequent # runs match the first run spec.write_array(self.get_seeds(vertex_slice)) # write the number of pre-populations spec.write_value(data=n_pre_pops) def __write_prepopulation_info( self, spec, app_vertex, structural_projections, weight_scales, synaptic_matrices): """ :param ~data_specification.DataSpecificationGenerator spec: :param ~pacman.model.graphs.application.ApplicationVertex app_vertex: the vertex for which data specs are being prepared :param structural_projections: Projections that are structural :type structural_projections: list(tuple(ProjectionApplicationEdge,SynapseInformation)) :param machine_edges_by_app: map of application edge to associated machine edges :type machine_edges_by_app: dict(~pacman.model.graphs.application.ApplicationEdge, list(~pacman.model.graphs.machine.MachineEdge)) :param dict(AbstractSynapseType,float) weight_scales: :param SynapticMatrices synaptic_matrices: :rtype: dict(tuple(AbstractPopulationVertex,SynapseInformation),int) """ spec.comment("Writing pre-population info") pop_index = dict() routing_info = SpynnakerDataView.get_routing_infos() subpop_index = dict() lo_atom_index = dict() index = 0 for proj in structural_projections: spec.comment(f"Writing pre-population info for {proj.label}") # pylint: disable=protected-access app_edge = proj._projection_edge synapse_info = proj._synapse_information pop_index[app_edge.pre_vertex, synapse_info] = index index += 1 dynamics = synapse_info.synapse_dynamics # Number of incoming vertices out_verts = app_edge.pre_vertex.splitter.get_out_going_vertices( SPIKE_PARTITION_ID) spec.write_value(len(out_verts), data_type=DataType.UINT16) # Controls - currently just if this is a self connection or not self_connected = app_vertex == app_edge.pre_vertex spec.write_value(int(self_connected), data_type=DataType.UINT16) # Delay delay_scale = SpynnakerDataView.get_simulation_time_step_per_ms() if isinstance(dynamics.initial_delay, Iterable): spec.write_value(int(dynamics.initial_delay[0] * delay_scale), data_type=DataType.UINT16) spec.write_value(int(dynamics.initial_delay[1] * delay_scale), data_type=DataType.UINT16) else: scaled_delay = dynamics.initial_delay * delay_scale spec.write_value(scaled_delay, data_type=DataType.UINT16) spec.write_value(scaled_delay, data_type=DataType.UINT16) # Weight spec.write_value(round(dynamics.initial_weight * weight_scales[synapse_info.synapse_type])) # Connection type spec.write_value(synapse_info.synapse_type) # Total number of atoms in pre-vertex spec.write_value(app_edge.pre_vertex.n_atoms) # Machine edge information for sub, m_vertex in enumerate(out_verts): r_info = routing_info.get_routing_info_from_pre_vertex( m_vertex, SPIKE_PARTITION_ID) vertex_slice = m_vertex.vertex_slice spec.write_value(r_info.key) spec.write_value(r_info.mask) spec.write_value(m_vertex.app_vertex.n_colour_bits) spec.write_value(vertex_slice.n_atoms) spec.write_value(vertex_slice.lo_atom) spec.write_value(synaptic_matrices.get_index( app_edge, synapse_info)) lo = m_vertex.vertex_slice.lo_atom for i in range(vertex_slice.lo_atom, vertex_slice.hi_atom + 1): subpop_index[app_edge.pre_vertex, synapse_info, i] = sub lo_atom_index[app_edge.pre_vertex, synapse_info, i] = lo return pop_index, subpop_index, lo_atom_index def __write_post_to_pre_table( self, spec, pop_index, subpop_index, lo_atom_index, app_vertex, vertex_slice): """ Post to pre table is basically the transpose of the synaptic matrix. :param ~data_specification.DataSpecificationGenerator spec: :param pop_index: :type pop_index: dict(tuple(AbstractPopulationVertex,SynapseInformation), int) :param ~pacman.model.graphs.application.ApplicationVertex app_vertex: the vertex for which data specs are being prepared :param ~pacman.model.graphs.common.Slice vertex_slice: The target slice """ # pylint: disable=unsubscriptable-object # Get connections for this post slice slice_conns = self.connections[app_vertex, vertex_slice.lo_atom] # Make a single large array of connections connections = numpy.concatenate( [conn for (conn, _, _) in slice_conns]) # Make a single large array of population index conn_lens = [len(conn) for (conn, _, _) in slice_conns] for (_, a_edge, s_info) in slice_conns: if (a_edge.pre_vertex, s_info) not in pop_index: print("Help!") pop_indices = numpy.repeat( [pop_index[a_edge.pre_vertex, s_info] for (_, a_edge, s_info) in slice_conns], conn_lens) # Make a single large array of sub-population index subpop_indices = numpy.array([ subpop_index[a_edge.pre_vertex, s_info, c["source"]] for (conns, a_edge, s_info) in slice_conns for c in conns]) # Get the low atom for each source and subtract lo_atoms = numpy.array([ lo_atom_index[a_edge.pre_vertex, s_info, c["source"]] for (conns, a_edge, s_info) in slice_conns for c in conns]) connections["source"] = connections["source"] - lo_atoms connections["target"] = connections["target"] - vertex_slice.lo_atom # Make an array of all data required conn_data = numpy.dstack( (pop_indices, subpop_indices, connections["source"]))[0] # Break data into rows based on target and strip target out rows = [conn_data[connections["target"] == i] for i in range(0, vertex_slice.n_atoms)] if any(len(row) > self.s_max for row in rows): raise ValueError( "Too many initial connections per incoming neuron") # Make each row the required length through padding with 0xFFFF padded_rows = [numpy.pad(row, [(self.s_max - len(row), 0), (0, 0)], "constant", constant_values=0xFFFF) for row in rows] # Finally make the table and write it out post_to_pre = numpy.core.records.fromarrays( numpy.concatenate(padded_rows).T, formats="u1, u1, u2").view("u4") if len(post_to_pre) != vertex_slice.n_atoms * self.s_max: raise ValueError( f"Wrong size of pre-to-pop tables: {len(post_to_pre)} " f"Found, {vertex_slice.n_atoms * self.s_max} Expected") spec.comment( "Writing post-to-pre table of " f"{vertex_slice.n_atoms * self.s_max} words") spec.write_array(post_to_pre)
[docs] @overrides(AbstractSynapseDynamicsStructural. get_structural_parameters_sdram_usage_in_bytes) def get_structural_parameters_sdram_usage_in_bytes( self, incoming_projections, n_neurons): # Work out how many sub-edges we will end up with, as this is used # for key_atom_info # pylint: disable=no-member param_sizes = ( self.partner_selection.get_parameters_sdram_usage_in_bytes()) n_sub_edges = 0 structural_projections = self.__get_structural_projections( incoming_projections) for proj in structural_projections: # pylint: disable=protected-access dynamics = proj._synapse_information.synapse_dynamics app_edge = proj._projection_edge n_sub_edges += len( app_edge.pre_vertex.splitter.get_out_going_slices()) param_sizes += dynamics.formation\ .get_parameters_sdram_usage_in_bytes() param_sizes += dynamics.elimination\ .get_parameters_sdram_usage_in_bytes() return int( self._REWIRING_DATA_SIZE + (self._PRE_POP_INFO_BASE_SIZE * len(structural_projections)) + (self._KEY_ATOM_INFO_SIZE * n_sub_edges) + (self._POST_TO_PRE_ENTRY_SIZE * n_neurons * self.s_max) + param_sizes)
[docs] def get_vertex_executable_suffix(self): """ :rtype: str """ name = "_structural" # pylint: disable=no-member name += self.partner_selection.vertex_executable_suffix name += self.formation.vertex_executable_suffix name += self.elimination.vertex_executable_suffix return name
[docs] def is_same_as(self, synapse_dynamics): """ :param SynapseDynamicsStructuralCommon synapse_dynamics: :rtype: bool """ # Note noqa:E721 because exact type comparison is required here return ( self.s_max == synapse_dynamics.s_max and self.f_rew == synapse_dynamics.f_rew and self.initial_weight == synapse_dynamics.initial_weight and self.initial_delay == synapse_dynamics.initial_delay and (type(self.partner_selection) == # noqa: E721 type(synapse_dynamics.partner_selection)) and (type(self.formation) == type(synapse_dynamics.formation)) and (type(self.elimination) == type(synapse_dynamics.elimination)))
@abstractproperty def connections(self): """ Initial connectivity as defined via connector. :rtype: dict """
[docs] @abstractmethod def get_seeds(self, app_vertex=None): """ Generate a seed for the RNG on chip that is the same for all of the cores that have perform structural updates. It should be different between application vertices but the same for the same app_vertex. It should be different every time called with `None`. :param app_vertex: :type app_vertex: ~pacman.model.graphs.application.ApplicationVertex or None :return: list of random seed (4 words), generated randomly :rtype: list(int) """
[docs] def check_initial_delay(self, max_delay_ms): """ Check that delays can be done without delay extensions. :param float max_delay_ms: The maximum delay supported, in milliseconds :raises ValueError: if the delay is out of range """ if isinstance(self.initial_delay, Iterable): # pylint: disable=unsubscriptable-object init_del = self.initial_delay if init_del[0] > max_delay_ms or init_del[1] > max_delay_ms: raise ValueError( f"The initial delay {self.initial_delay} has one or more " f"values that are bigger than {max_delay_ms}.") elif self.initial_delay > max_delay_ms: raise ValueError( f"The initial delay {self.initial_delay} " f"is bigger than {max_delay_ms}.")
[docs] def get_max_rewires_per_ts(self): max_rewires_per_ts = 1 if (self.p_rew * MICRO_TO_MILLISECOND_CONVERSION < SpynnakerDataView.get_simulation_time_step_ms()): # fast rewiring, so need to set max_rewires_per_ts max_rewires_per_ts = int( SpynnakerDataView.get_simulation_time_step_us() / ( self.p_rew * MICRO_TO_SECOND_CONVERSION)) return max_rewires_per_ts