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

# 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 numpy
from pyNN.standardmodels.synapses import StaticSynapse
from spinn_utilities.overrides import overrides
from spynnaker.pyNN.exceptions import SynapticConfigurationException
from spynnaker.pyNN.utilities.utility_calls import create_mars_kiss_seeds
from .abstract_synapse_dynamics_structural import (
from .synapse_dynamics_stdp import SynapseDynamicsSTDP
from .synapse_dynamics_structural_common import (
    DEFAULT_S_MAX, SynapseDynamicsStructuralCommon)
from .synapse_dynamics_neuromodulation import SynapseDynamicsNeuromodulation
from spynnaker.pyNN.utilities.constants import SPIKE_PARTITION_ID

class SynapseDynamicsStructuralSTDP(
        SynapseDynamicsSTDP, SynapseDynamicsStructuralCommon):
    Class that enables synaptic rewiring in the presence of STDP.

    It acts as a wrapper around SynapseDynamicsSTDP, meaning rewiring can
    operate in parallel with STDP synapses.

    Written by Petrut Bogdan.
    __slots__ = [
        # Frequency of rewiring (Hz)
        # Initial weight assigned to a newly formed connection
        # Delay assigned to a newly formed connection
        # Maximum fan-in per target layer neuron
        # The seed
        # Holds initial connectivity as defined via connector
        # The actual type of weights: static through the simulation or those
        # that can be change through STDP
        # Shared RNG seed to be written on all cores
        # The RNG used with the seed that is passed in
        # The partner selection rule
        # The formation rule
        # The elimination rule

    def __init__(
            self, partner_selection, formation, elimination,
            timing_dependence=None, weight_dependence=None,
            voltage_dependence=None, dendritic_delay_fraction=1.0,
            f_rew=DEFAULT_F_REW, initial_weight=DEFAULT_INITIAL_WEIGHT,
            initial_delay=DEFAULT_INITIAL_DELAY, s_max=DEFAULT_S_MAX,
            with_replacement=True, seed=None,
            weight=StaticSynapse.default_parameters['weight'], delay=None,
        :param AbstractPartnerSelection partner_selection:
            The partner selection rule
        :param AbstractFormation formation: The formation rule
        :param AbstractElimination elimination: The elimination rule
        :param AbstractTimingDependence timing_dependence:
            The STDP timing dependence rule
        :param AbstractWeightDependence weight_dependence:
            The STDP weight dependence rule
        :param None voltage_dependence:
            The STDP voltage dependence (unsupported)
        :param float dendritic_delay_fraction:
            The STDP dendritic delay fraction
        :param float f_rew:
            How many rewiring attempts will be done per second.
        :param float initial_weight:
            Weight assigned to a newly formed connection
        :param initial_delay:
            Delay assigned to a newly formed connection; a single value means
            a fixed delay value, or a tuple of two values means the delay will
            be chosen at random from a uniform distribution between the given
        :type initial_delay: float or tuple(float, float)
        :param int s_max: Maximum fan-in per target layer neuron
        :param bool with_replacement:
            If set to True (default), a new synapse can be formed in a
            location where a connection already exists; if False, then it must
            form where no connection already exists
        :param seed: seed for the random number generators
        :type seed: int or None
        :param float weight: The weight of connections formed by the connector
        :param delay: The delay of connections formed by the connector
            Use ``None`` to get the simulator default minimum delay.
        :type delay: float or None
        :param bool backprop_delay: Whether back-propagated delays are used
            timing_dependence, weight_dependence, voltage_dependence,
            dendritic_delay_fraction, weight, delay, pad_to_length=s_max,
        self.__partner_selection = partner_selection
        self.__formation = formation
        self.__elimination = elimination
        self.__f_rew = float(f_rew)
        self.__initial_weight = initial_weight
        self.__initial_delay = initial_delay
        self.__s_max = s_max
        self.__with_replacement = with_replacement
        self.__seed = seed
        self.__connections = dict()

        self.__rng = numpy.random.RandomState(seed)
        self.__seeds = dict()

[docs] @overrides(SynapseDynamicsSTDP.merge) def merge(self, synapse_dynamics): # If dynamics is Neuromodulation, merge with other neuromodulation, # and then return ourselves, as neuromodulation can't be used by # itself if isinstance(synapse_dynamics, SynapseDynamicsNeuromodulation): super().merge_neuromodulation(synapse_dynamics) return self # If other is structural, check structural matches if isinstance(synapse_dynamics, AbstractSynapseDynamicsStructural): if not SynapseDynamicsStructuralCommon.is_same_as( self, synapse_dynamics): raise SynapticConfigurationException( "Synapse dynamics must match exactly when using multiple" " edges to the same population") # If other is STDP, check STDP matches if isinstance(synapse_dynamics, SynapseDynamicsSTDP): if not SynapseDynamicsSTDP.is_same_as(self, synapse_dynamics): raise SynapticConfigurationException( "Synapse dynamics must match exactly when using multiple" " edges to the same population") # If everything matches, return ourselves as supreme! return self
[docs] def set_projection_parameter(self, param, value): """ :param str param: :param value: """ for item in (self.partner_selection, self.__formation, self.__elimination): if hasattr(item, param): setattr(item, param, value) break else: raise ValueError(f"Unknown parameter {param}")
[docs] @overrides(SynapseDynamicsSTDP.is_same_as) def is_same_as(self, synapse_dynamics): if (isinstance(synapse_dynamics, SynapseDynamicsSTDP) and not super().is_same_as(synapse_dynamics)): return False return SynapseDynamicsStructuralCommon.is_same_as( self, synapse_dynamics)
[docs] @overrides(SynapseDynamicsSTDP.get_vertex_executable_suffix) def get_vertex_executable_suffix(self): return (super().get_vertex_executable_suffix() + SynapseDynamicsStructuralCommon.get_vertex_executable_suffix( self))
[docs] @overrides(AbstractSynapseDynamicsStructural.set_connections) def set_connections( self, connections, post_vertex_slice, app_edge, synapse_info): if not isinstance(synapse_info.synapse_dynamics, AbstractSynapseDynamicsStructural): return collector = self.__connections.setdefault( (app_edge.post_vertex, post_vertex_slice.lo_atom), []) collector.append((connections, app_edge, synapse_info))
[docs] @overrides(SynapseDynamicsSTDP.get_parameter_names) def get_parameter_names(self): names = super().get_parameter_names() names.extend(SynapseDynamicsStructuralCommon.get_parameter_names(self)) return names
@property @overrides(AbstractSynapseDynamicsStructural.f_rew) def f_rew(self): return self.__f_rew @property @overrides(AbstractSynapseDynamicsStructural.s_max) def s_max(self): return self.__s_max @property @overrides(AbstractSynapseDynamicsStructural.with_replacement) def with_replacement(self): return self.__with_replacement @property @overrides(AbstractSynapseDynamicsStructural.seed) def seed(self): return self.__seed @property @overrides(AbstractSynapseDynamicsStructural.initial_weight) def initial_weight(self): return self.__initial_weight @property @overrides(AbstractSynapseDynamicsStructural.initial_delay) def initial_delay(self): return self.__initial_delay @property @overrides(AbstractSynapseDynamicsStructural.partner_selection) def partner_selection(self): return self.__partner_selection @property @overrides(AbstractSynapseDynamicsStructural.formation) def formation(self): return self.__formation @property @overrides(AbstractSynapseDynamicsStructural.elimination) def elimination(self): return self.__elimination @property @overrides(SynapseDynamicsStructuralCommon.connections) def connections(self): return self.__connections
[docs] @overrides(SynapseDynamicsSTDP.get_weight_mean) def get_weight_mean(self, connector, synapse_info): return self.get_weight_maximum(connector, synapse_info)
[docs] @overrides(SynapseDynamicsSTDP.get_weight_maximum) def get_weight_maximum(self, connector, synapse_info): w_max = super().get_weight_maximum(connector, synapse_info) return max(w_max, self.__initial_weight)
[docs] @overrides(SynapseDynamicsSTDP.get_delay_maximum) def get_delay_maximum(self, connector, synapse_info): d_m = super().get_delay_maximum(connector, synapse_info) return max(d_m, self.__initial_delay)
[docs] @overrides(SynapseDynamicsSTDP.get_delay_minimum) def get_delay_minimum(self, connector, synapse_info): d_m = super().get_delay_minimum(connector, synapse_info) return min(d_m, self.__initial_delay)
[docs] @overrides(SynapseDynamicsSTDP.get_delay_variance) def get_delay_variance(self, connector, delays, synapse_info): return 0.0
[docs] @overrides(SynapseDynamicsStructuralCommon.get_seeds) def get_seeds(self, app_vertex=None): if app_vertex: if app_vertex not in self.__seeds.keys(): self.__seeds[app_vertex] = ( create_mars_kiss_seeds(self.__rng)) return self.__seeds[app_vertex] else: return create_mars_kiss_seeds(self.__rng)
[docs] @overrides(SynapseDynamicsSTDP.generate_on_machine) def generate_on_machine(self): # Never generate structural connections on the machine return False
[docs] @overrides(SynapseDynamicsSTDP.get_connected_vertices) def get_connected_vertices(self, s_info, source_vertex, target_vertex): # Things change, so assume all connected return [(m_vertex, [source_vertex]) for m_vertex in target_vertex.splitter.get_in_coming_vertices( SPIKE_PARTITION_ID)]
@property @overrides(SynapseDynamicsSTDP.is_combined_core_capable) def is_combined_core_capable(self): return False