6944077364

Committed 21 Nov 2023 12:53PM UTC coverage: 63.152% (+1.6%) from 61.529%

Build # 6944077364

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1342 from SpiNNakerManchester/reduce-op-complexity

Type Annotations and Checking

Run Details

1960 of 4482 branches covered (0.0%)

Branch coverage included in aggregate %.

4146 of 5076 new or added lines in 233 files covered. (81.68%)

193 existing lines in 78 files now uncovered.

12724 of 18770 relevant lines covered (67.79%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

19.5

/spynnaker/pyNN/models/neuron/synapse_dynamics/synapse_dynamics_structural_common.py

# 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
#
#     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 numpy
from numpy.typing import NDArray
from typing import (
    Dict, Iterable, List, Sequence, Tuple, Union, TYPE_CHECKING)
from typing_extensions import TypeAlias
from spinn_utilities.abstract_base import AbstractBase, abstractmethod
from spinn_utilities.overrides import overrides
from pacman.model.graphs.application import ApplicationVertex
from pacman.model.graphs.common import Slice
from spinn_front_end_common.interface.ds import (
    DataType, DataSpecificationBase)
from spinn_front_end_common.utilities.constants import (
    MICRO_TO_MILLISECOND_CONVERSION, MICRO_TO_SECOND_CONVERSION,
    BYTES_PER_WORD, BYTES_PER_SHORT)
from spynnaker.pyNN.data import SpynnakerDataView
from .abstract_synapse_dynamics_structural import (
    AbstractSynapseDynamicsStructural)
from spynnaker.pyNN.exceptions import SynapticConfigurationException
from spynnaker.pyNN.models.common import PopulationApplicationVertex
from spynnaker.pyNN.utilities.constants import SPIKE_PARTITION_ID
if TYPE_CHECKING:
    from spynnaker.pyNN.models.projection import Projection
    from spynnaker.pyNN.models.neural_projections import (
        SynapseInformation)
    from spynnaker.pyNN.models.neuron import AbstractPopulationVertex
    from spynnaker.pyNN.models.neuron.synaptic_matrices import SynapticMatrices
    from spynnaker.pyNN.models.neuron.synapse_dynamics.types import (
        ConnectionsArray)
    from spynnaker.pyNN.models.neural_projections import (
        ProjectionApplicationEdge)

    _PopIndexType: TypeAlias = Dict[
        Tuple[PopulationApplicationVertex, SynapseInformation], int]
    _SubpopIndexType: TypeAlias = Dict[
        Tuple[PopulationApplicationVertex, SynapseInformation, int], int]

    #: :meta private:
    ConnectionsInfo: TypeAlias = Dict[
        Tuple[AbstractPopulationVertex, int],
        List[Tuple[ConnectionsArray, ProjectionApplicationEdge,
                   SynapseInformation]]]

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


class SynapseDynamicsStructuralCommon(
        AbstractSynapseDynamicsStructural, metaclass=AbstractBase):
    """
    Common code for structural synapse dynamics.
    """

    # 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)
    _REWIRING_DATA_SIZE = (
        (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)
    _PRE_POP_INFO_BASE_SIZE = (4 * BYTES_PER_SHORT) + (3 * BYTES_PER_WORD)

    # 6 32-bit numbers (key; mask; n_atoms; n_colour_bits; lo_atom;
    # m_pop_index)
    _KEY_ATOM_INFO_SIZE = (6 * BYTES_PER_WORD)

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

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

    __slots__ = ()

    def get_parameter_names(self) -> Iterable[str]:
        """
        :rtype: list(str)
        """
        yield from [
            'initial_weight', 'initial_delay', 'f_rew', 'p_rew', 's_max',
            'with_replacement']
        # pylint: disable=no-member
        yield from self.partner_selection.get_parameter_names()
        yield from self.formation.get_parameter_names()
        yield from self.elimination.get_parameter_names()

    @property
    def p_rew(self) -> float:
        """
        The period of rewiring.

        :rtype: float
        """
        return 1. / self.f_rew

    @overrides(AbstractSynapseDynamicsStructural.write_structural_parameters)
    def write_structural_parameters(
            self, spec: DataSpecificationBase, region: int,
            weight_scales: NDArray[numpy.floating],
            app_vertex: AbstractPopulationVertex, vertex_slice: Slice,
            synaptic_matrices: SynapticMatrices):
        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: Iterable[Projection]
            ) -> List[Projection]:
        """
        :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)
                    seen_app_edges.add(app_edge)
                    structural_projections.append(proj)
        return structural_projections

    def __write_common_rewiring_data(
            self, spec: DataSpecificationBase,
            app_vertex: AbstractPopulationVertex, vertex_slice: Slice,
            n_pre_pops: int):
        """
        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: DataSpecificationBase,
            app_vertex: ApplicationVertex,
            structural_projections: Iterable[Projection],
            weight_scales: NDArray[numpy.floating],
            synaptic_matrices: SynapticMatrices) -> Tuple[
                _PopIndexType, _SubpopIndexType, _SubpopIndexType]:
        """
        :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
        :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(int,float) weight_scales:
        :param SynapticMatrices synaptic_matrices:
        :rtype: dict(tuple(AbstractPopulationVertex,SynapseInformation),int)
        """
        spec.comment("Writing pre-population info")
        pop_index: _PopIndexType = dict()
        routing_info = SpynnakerDataView.get_routing_infos()
        subpop_index: _SubpopIndexType = dict()
        lo_atom_index: _SubpopIndexType = 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, tuple):
                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)
                assert r_info is not None
                vertex_slice = m_vertex.vertex_slice
                spec.write_value(r_info.key)
                spec.write_value(r_info.mask)
                out_app = m_vertex.app_vertex
                assert isinstance(out_app, PopulationApplicationVertex)
                spec.write_value(out_app.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 = vertex_slice.lo_atom
                for i in range(lo, 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: DataSpecificationBase, pop_index: _PopIndexType,
            subpop_index: _SubpopIndexType, lo_atom_index: _SubpopIndexType,
            app_vertex: AbstractPopulationVertex, vertex_slice: 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)

    @overrides(AbstractSynapseDynamicsStructural.
               get_structural_parameters_sdram_usage_in_bytes)
    def get_structural_parameters_sdram_usage_in_bytes(
            self, incoming_projections: Iterable[Projection],
            n_neurons: int) -> int:
        # 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)

    def get_vertex_executable_suffix(self) -> str:
        """
        :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

    def is_same_as(
            self, synapse_dynamics: AbstractSynapseDynamicsStructural) -> bool:
        """
        :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)))

    @property
    @abstractmethod
    def connections(self) -> ConnectionsInfo:
        """
        Initial connectivity as defined via connector.

        :rtype: dict
        """
        raise NotImplementedError

    @abstractmethod
    def _get_seeds(
            self, app_vertex: Union[None, ApplicationVertex, Slice] = None
            ) -> Sequence[int]:
        """
        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)
        """
        raise NotImplementedError

    def check_initial_delay(self, max_delay_ms: float):
        """
        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, tuple):
            # 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}.")

    def get_max_rewires_per_ts(self) -> int:
        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

1	# Copyright (c) 2016 The University of Manchester
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# https://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	from __future__ import annotations	1✔
16	import numpy	1✔
17	from numpy.typing import NDArray	1✔
18	from typing import (	1✔
19	Dict, Iterable, List, Sequence, Tuple, Union, TYPE_CHECKING)
20	from typing_extensions import TypeAlias	1✔
21	from spinn_utilities.abstract_base import AbstractBase, abstractmethod	1✔
22	from spinn_utilities.overrides import overrides	1✔
23	from pacman.model.graphs.application import ApplicationVertex	1✔
24	from pacman.model.graphs.common import Slice	1✔
25	from spinn_front_end_common.interface.ds import (	1✔
26	DataType, DataSpecificationBase)
27	from spinn_front_end_common.utilities.constants import (	1✔
28	MICRO_TO_MILLISECOND_CONVERSION, MICRO_TO_SECOND_CONVERSION,
29	BYTES_PER_WORD, BYTES_PER_SHORT)
30	from spynnaker.pyNN.data import SpynnakerDataView	1✔
31	from .abstract_synapse_dynamics_structural import (	1✔
32	AbstractSynapseDynamicsStructural)
33	from spynnaker.pyNN.exceptions import SynapticConfigurationException	1✔
34	from spynnaker.pyNN.models.common import PopulationApplicationVertex	1✔
35	from spynnaker.pyNN.utilities.constants import SPIKE_PARTITION_ID	1✔
36	if TYPE_CHECKING:
37	from spynnaker.pyNN.models.projection import Projection
38	from spynnaker.pyNN.models.neural_projections import (
39	SynapseInformation)
40	from spynnaker.pyNN.models.neuron import AbstractPopulationVertex
41	from spynnaker.pyNN.models.neuron.synaptic_matrices import SynapticMatrices
42	from spynnaker.pyNN.models.neuron.synapse_dynamics.types import (
43	ConnectionsArray)
44	from spynnaker.pyNN.models.neural_projections import (
45	ProjectionApplicationEdge)
46
47	_PopIndexType: TypeAlias = Dict[
48	Tuple[PopulationApplicationVertex, SynapseInformation], int]
49	_SubpopIndexType: TypeAlias = Dict[
50	Tuple[PopulationApplicationVertex, SynapseInformation, int], int]
51
52	#: :meta private:
53	ConnectionsInfo: TypeAlias = Dict[
54	Tuple[AbstractPopulationVertex, int],
55	List[Tuple[ConnectionsArray, ProjectionApplicationEdge,
56	SynapseInformation]]]
57
58	#: Default value for frequency of rewiring
59	DEFAULT_F_REW = 10 ** 4.0	1✔
60	#: Default value for initial weight on connection formation
61	DEFAULT_INITIAL_WEIGHT = 0.0	1✔
62	#: Default value for initial delay on connection formation
63	DEFAULT_INITIAL_DELAY = 1.0	1✔
64	#: Default value for maximum fan-in per target layer neuron
65	DEFAULT_S_MAX = 32	1✔
66
67
68	class SynapseDynamicsStructuralCommon(	1✔
69	AbstractSynapseDynamicsStructural, metaclass=AbstractBase):
70	"""
71	Common code for structural synapse dynamics.
72	"""
73
74	# 8 32-bit numbers (fast; p_rew; s_max; app_no_atoms; machine_no_atoms;
75	# low_atom; high_atom; with_replacement) + 2 4-word RNG seeds (shared_seed;
76	# local_seed) + 1 32-bit number (no_pre_pops)
77	_REWIRING_DATA_SIZE = (	1✔
78	(8 * BYTES_PER_WORD) + (2 * 4 * BYTES_PER_WORD) + BYTES_PER_WORD)
79
80	# Size excluding key_atom_info (as variable length)
81	# 4 16-bit numbers (no_pre_vertices; sp_control; delay_lo; delay_hi)
82	# + 3 32-bit numbers (weight; connection_type; total_no_atoms)
83	_PRE_POP_INFO_BASE_SIZE = (4 * BYTES_PER_SHORT) + (3 * BYTES_PER_WORD)	1✔
84
85	# 6 32-bit numbers (key; mask; n_atoms; n_colour_bits; lo_atom;
86	# m_pop_index)
87	_KEY_ATOM_INFO_SIZE = (6 * BYTES_PER_WORD)	1✔
88
89	# 1 16-bit number (neuron_index)
90	# + 2 8-bit numbers (sub_pop_index; pop_index)
91	_POST_TO_PRE_ENTRY_SIZE = BYTES_PER_SHORT + (2 * 1)	1✔
92
93	PAIR_ERROR = (	1✔
94	"Only one Projection between each pair of Populations can use "
95	"structural plasticity")
96
97	__slots__ = ()	1✔
98
99	def get_parameter_names(self) -> Iterable[str]:	1✔
100	"""
101	:rtype: list(str)
102	"""
NEW 103	yield from [	×
104	'initial_weight', 'initial_delay', 'f_rew', 'p_rew', 's_max',
105	'with_replacement']
106	# pylint: disable=no-member
NEW 107	yield from self.partner_selection.get_parameter_names()	×
NEW 108	yield from self.formation.get_parameter_names()	×
NEW 109	yield from self.elimination.get_parameter_names()	×
110
111	@property	1✔
112	def p_rew(self) -> float:	1✔
113	"""
114	The period of rewiring.
115
116	:rtype: float
117	"""
118	return 1. / self.f_rew	×
119
120	@overrides(AbstractSynapseDynamicsStructural.write_structural_parameters)	1✔
121	def write_structural_parameters(	1✔
122	self, spec: DataSpecificationBase, region: int,
123	weight_scales: NDArray[numpy.floating],
124	app_vertex: AbstractPopulationVertex, vertex_slice: Slice,
125	synaptic_matrices: SynapticMatrices):
126	spec.comment("Writing structural plasticity parameters")	×
127	spec.switch_write_focus(region)	×
128
129	# Get relevant edges
130	structural_projections = self.__get_structural_projections(	×
131	app_vertex.incoming_projections)
132
133	# Write the common part of the rewiring data
134	self.__write_common_rewiring_data(	×
135	spec, app_vertex, vertex_slice, len(structural_projections))
136
137	# Write the pre-population info
138	pop_index, subpop_index, lo_atom_index = \	×
139	self.__write_prepopulation_info(
140	spec, app_vertex, structural_projections,
141	weight_scales, synaptic_matrices)
142
143	# Write the post-to-pre table
144	self.__write_post_to_pre_table(	×
145	spec, pop_index, subpop_index, lo_atom_index, app_vertex,
146	vertex_slice)
147
148	# Write the component parameters
149	# pylint: disable=no-member, protected-access
150	spec.comment("Writing partner selection parameters")	×
151	self.partner_selection.write_parameters(spec)	×
152	for proj in structural_projections:	×
153	spec.comment(f"Writing formation parameters for {proj.label}")	×
154	dynamics = proj._synapse_information.synapse_dynamics	×
155	dynamics.formation.write_parameters(spec)	×
156	for proj in structural_projections:	×
157	spec.comment(f"Writing elimination parameters for {proj.label}")	×
158	dynamics = proj._synapse_information.synapse_dynamics	×
159	dynamics.elimination.write_parameters(	×
160	spec, weight_scales[proj._synapse_information.synapse_type])
161
162	def __get_structural_projections(	1✔
163	self, incoming_projections: Iterable[Projection]
164	) -> List[Projection]:
165	"""
166	:param list(Projection) incoming_projections:
167	Projections to filter to structural only
168	:rtype: list(Projection)
169	"""
170	structural_projections = list()	×
171	seen_app_edges = set()	×
172	for proj in incoming_projections:	×
173	# pylint: disable=protected-access
174	app_edge = proj._projection_edge	×
175	for synapse_info in app_edge.synapse_information:	×
176	if isinstance(synapse_info.synapse_dynamics,	×
177	AbstractSynapseDynamicsStructural):
178	if app_edge in seen_app_edges:	×
NEW 179	raise SynapticConfigurationException(self.PAIR_ERROR)	×
NEW 180	seen_app_edges.add(app_edge)	×
NEW 181	structural_projections.append(proj)	×
UNCOV 182	return structural_projections	×
183
184	def __write_common_rewiring_data(	1✔
185	self, spec: DataSpecificationBase,
186	app_vertex: AbstractPopulationVertex, vertex_slice: Slice,
187	n_pre_pops: int):
188	"""
189	Write the non-sub-population synapse parameters to the spec.
190
191	:param ~data_specification.DataSpecificationGenerator spec:
192	the data spec
193	:param ~pacman.model.graphs.application.ApplicationVertex app_vertex:
194	The application vertex being generated
195	:param ~pacman.model.graphs.common.Slice vertex_slice:
196	The slice of the target vertex to generate for
197	:param int n_pre_pops: the number of pre-populations
198	"""
199	time_step_us = SpynnakerDataView.get_simulation_time_step_us()	×
200	spec.comment("Writing common rewiring data")	×
201	if (self.p_rew * MICRO_TO_MILLISECOND_CONVERSION <	×
202	time_step_us / MICRO_TO_MILLISECOND_CONVERSION):
203	# Fast rewiring
204	spec.write_value(data=1)	×
205	spec.write_value(data=int(	×
206	time_step_us / (
207	self.p_rew * MICRO_TO_SECOND_CONVERSION)))
208	else:
209	# Slow rewiring
210	spec.write_value(data=0)	×
211	spec.write_value(data=int((	×
212	self.p_rew * MICRO_TO_SECOND_CONVERSION) /
213	time_step_us))
214	# write s_max
215	spec.write_value(data=int(self.s_max))	×
216	# write total number of atoms in the application vertex
217	spec.write_value(data=app_vertex.n_atoms)	×
218	# write local low, high and number of atoms
219	spec.write_value(data=vertex_slice.n_atoms)	×
220	spec.write_value(data=vertex_slice.lo_atom)	×
221	spec.write_value(data=vertex_slice.hi_atom)	×
222	# write with_replacement
223	spec.write_value(data=self.with_replacement)	×
224
225	# write app level seeds
NEW 226	spec.write_array(self._get_seeds(app_vertex))	×
227
228	# write local seed (4 words), generated randomly!
229	# Note that in case of a reset, these need a key to ensure subsequent
230	# runs match the first run
NEW 231	spec.write_array(self._get_seeds(vertex_slice))	×
232
233	# write the number of pre-populations
234	spec.write_value(data=n_pre_pops)	×
235
236	def __write_prepopulation_info(	1✔
237	self, spec: DataSpecificationBase,
238	app_vertex: ApplicationVertex,
239	structural_projections: Iterable[Projection],
240	weight_scales: NDArray[numpy.floating],
241	synaptic_matrices: SynapticMatrices) -> Tuple[
242	_PopIndexType, _SubpopIndexType, _SubpopIndexType]:
243	"""
244	:param ~data_specification.DataSpecificationGenerator spec:
245	:param ~pacman.model.graphs.application.ApplicationVertex app_vertex:
246	the vertex for which data specs are being prepared
247	:param structural_projections: Projections that are structural
248	:param machine_edges_by_app:
249	map of application edge to associated machine edges
250	:type machine_edges_by_app:
251	dict(~pacman.model.graphs.application.ApplicationEdge,
252	list(~pacman.model.graphs.machine.MachineEdge))
253	:param dict(int,float) weight_scales:
254	:param SynapticMatrices synaptic_matrices:
255	:rtype: dict(tuple(AbstractPopulationVertex,SynapseInformation),int)
256	"""
257	spec.comment("Writing pre-population info")	×
NEW 258	pop_index: _PopIndexType = dict()	×
259	routing_info = SpynnakerDataView.get_routing_infos()	×
NEW 260	subpop_index: _SubpopIndexType = dict()	×
NEW 261	lo_atom_index: _SubpopIndexType = dict()	×
262	index = 0	×
263	for proj in structural_projections:	×
264	spec.comment(f"Writing pre-population info for {proj.label}")	×
265	# pylint: disable=protected-access
266	app_edge = proj._projection_edge	×
267	synapse_info = proj._synapse_information	×
268	pop_index[app_edge.pre_vertex, synapse_info] = index	×
269	index += 1	×
270	dynamics = synapse_info.synapse_dynamics	×
271
272	# Number of incoming vertices
273	out_verts = app_edge.pre_vertex.splitter.get_out_going_vertices(	×
274	SPIKE_PARTITION_ID)
275	spec.write_value(len(out_verts), data_type=DataType.UINT16)	×
276
277	# Controls - currently just if this is a self connection or not
278	self_connected = app_vertex == app_edge.pre_vertex	×
279	spec.write_value(int(self_connected), data_type=DataType.UINT16)	×
280	# Delay
281	delay_scale = SpynnakerDataView.get_simulation_time_step_per_ms()	×
NEW 282	if isinstance(dynamics.initial_delay, tuple):	×
283	spec.write_value(int(dynamics.initial_delay[0] * delay_scale),	×
284	data_type=DataType.UINT16)
285	spec.write_value(int(dynamics.initial_delay[1] * delay_scale),	×
286	data_type=DataType.UINT16)
287	else:
288	scaled_delay = dynamics.initial_delay * delay_scale	×
289	spec.write_value(scaled_delay, data_type=DataType.UINT16)	×
290	spec.write_value(scaled_delay, data_type=DataType.UINT16)	×
291
292	# Weight
293	spec.write_value(round(dynamics.initial_weight *	×
294	weight_scales[synapse_info.synapse_type]))
295	# Connection type
296	spec.write_value(synapse_info.synapse_type)	×
297	# Total number of atoms in pre-vertex
298	spec.write_value(app_edge.pre_vertex.n_atoms)	×
299	# Machine edge information
300	for sub, m_vertex in enumerate(out_verts):	×
301	r_info = routing_info.get_routing_info_from_pre_vertex(	×
302	m_vertex, SPIKE_PARTITION_ID)
NEW 303	assert r_info is not None	×
304	vertex_slice = m_vertex.vertex_slice	×
305	spec.write_value(r_info.key)	×
306	spec.write_value(r_info.mask)	×
NEW 307	out_app = m_vertex.app_vertex	×
NEW 308	assert isinstance(out_app, PopulationApplicationVertex)	×
NEW 309	spec.write_value(out_app.n_colour_bits)	×
310	spec.write_value(vertex_slice.n_atoms)	×
311	spec.write_value(vertex_slice.lo_atom)	×
312	spec.write_value(synaptic_matrices.get_index(	×
313	app_edge, synapse_info))
NEW 314	lo = vertex_slice.lo_atom	×
NEW 315	for i in range(lo, vertex_slice.hi_atom + 1):	×
316	subpop_index[app_edge.pre_vertex, synapse_info, i] = sub	×
317	lo_atom_index[app_edge.pre_vertex, synapse_info, i] = lo	×
318	return pop_index, subpop_index, lo_atom_index	×
319
320	def __write_post_to_pre_table(	1✔
321	self, spec: DataSpecificationBase, pop_index: _PopIndexType,
322	subpop_index: _SubpopIndexType, lo_atom_index: _SubpopIndexType,
323	app_vertex: AbstractPopulationVertex, vertex_slice: Slice):
324	"""
325	Post to pre table is basically the transpose of the synaptic matrix.
326
327	:param ~data_specification.DataSpecificationGenerator spec:
328	:param pop_index:
329	:type pop_index:
330	dict(tuple(AbstractPopulationVertex,SynapseInformation), int)
331	:param ~pacman.model.graphs.application.ApplicationVertex app_vertex:
332	the vertex for which data specs are being prepared
333	:param ~pacman.model.graphs.common.Slice vertex_slice:
334	The target slice
335	"""
336	# pylint: disable=unsubscriptable-object
337	# Get connections for this post slice
338	slice_conns = self.connections[app_vertex, vertex_slice.lo_atom]	×
339	# Make a single large array of connections
340	connections = numpy.concatenate(	×
341	[conn for (conn, _, _) in slice_conns])
342	# Make a single large array of population index
343	conn_lens = [len(conn) for (conn, _, _) in slice_conns]	×
344	for (_, a_edge, s_info) in slice_conns:	×
345	if (a_edge.pre_vertex, s_info) not in pop_index:	×
346	print("Help!")	×
347	pop_indices = numpy.repeat(	×
348	[pop_index[a_edge.pre_vertex, s_info]
349	for (_, a_edge, s_info) in slice_conns], conn_lens)
350	# Make a single large array of sub-population index
351	subpop_indices = numpy.array([	×
352	subpop_index[a_edge.pre_vertex, s_info, c["source"]]
353	for (conns, a_edge, s_info) in slice_conns for c in conns])
354	# Get the low atom for each source and subtract
355	lo_atoms = numpy.array([	×
356	lo_atom_index[a_edge.pre_vertex, s_info, c["source"]]
357	for (conns, a_edge, s_info) in slice_conns for c in conns])
358	connections["source"] = connections["source"] - lo_atoms	×
359	connections["target"] = connections["target"] - vertex_slice.lo_atom	×
360
361	# Make an array of all data required
362	conn_data = numpy.dstack(	×
363	(pop_indices, subpop_indices, connections["source"]))[0]
364
365	# Break data into rows based on target and strip target out
366	rows = [conn_data[connections["target"] == i]	×
367	for i in range(0, vertex_slice.n_atoms)]
368
369	if any(len(row) > self.s_max for row in rows):	×
370	raise ValueError(	×
371	"Too many initial connections per incoming neuron")
372
373	# Make each row the required length through padding with 0xFFFF
374	padded_rows = [numpy.pad(row, [(self.s_max - len(row), 0), (0, 0)],	×
375	"constant", constant_values=0xFFFF)
376	for row in rows]
377
378	# Finally make the table and write it out
379	post_to_pre = numpy.core.records.fromarrays(	×
380	numpy.concatenate(padded_rows).T, formats="u1, u1, u2").view("u4")
381	if len(post_to_pre) != vertex_slice.n_atoms * self.s_max:	×
382	raise ValueError(	×
383	f"Wrong size of pre-to-pop tables: {len(post_to_pre)} "
384	f"Found, {vertex_slice.n_atoms * self.s_max} Expected")
385	spec.comment(	×
386	"Writing post-to-pre table of "
387	f"{vertex_slice.n_atoms * self.s_max} words")
388	spec.write_array(post_to_pre)	×
389
390	@overrides(AbstractSynapseDynamicsStructural.	1✔
391	get_structural_parameters_sdram_usage_in_bytes)
392	def get_structural_parameters_sdram_usage_in_bytes(	1✔
393	self, incoming_projections: Iterable[Projection],
394	n_neurons: int) -> int:
395	# Work out how many sub-edges we will end up with, as this is used
396	# for key_atom_info
397	# pylint: disable=no-member
398	param_sizes = (	×
399	self.partner_selection.get_parameters_sdram_usage_in_bytes())
400	n_sub_edges = 0	×
401	structural_projections = self.__get_structural_projections(	×
402	incoming_projections)
403	for proj in structural_projections:	×
404	# pylint: disable=protected-access
405	dynamics = proj._synapse_information.synapse_dynamics	×
406	app_edge = proj._projection_edge	×
407	n_sub_edges += len(	×
408	app_edge.pre_vertex.splitter.get_out_going_slices())
409	param_sizes += dynamics.formation\	×
410	.get_parameters_sdram_usage_in_bytes()
411	param_sizes += dynamics.elimination\	×
412	.get_parameters_sdram_usage_in_bytes()
413
414	return int(	×
415	self._REWIRING_DATA_SIZE +
416	(self._PRE_POP_INFO_BASE_SIZE * len(structural_projections)) +
417	(self._KEY_ATOM_INFO_SIZE * n_sub_edges) +
418	(self._POST_TO_PRE_ENTRY_SIZE * n_neurons * self.s_max) +
419	param_sizes)
420
421	def get_vertex_executable_suffix(self) -> str:	1✔
422	"""
423	:rtype: str
424	"""
425	name = "_structural"	×
426	# pylint: disable=no-member
427	name += self.partner_selection.vertex_executable_suffix	×
428	name += self.formation.vertex_executable_suffix	×
429	name += self.elimination.vertex_executable_suffix	×
430	return name	×
431
432	def is_same_as(	1✔
433	self, synapse_dynamics: AbstractSynapseDynamicsStructural) -> bool:
434	"""
435	:param SynapseDynamicsStructuralCommon synapse_dynamics:
436	:rtype: bool
437	"""
438	# Note noqa:E721 because exact type comparison is required here
439	return (	×
440	self.s_max == synapse_dynamics.s_max and
441	self.f_rew == synapse_dynamics.f_rew and
442	self.initial_weight == synapse_dynamics.initial_weight and
443	self.initial_delay == synapse_dynamics.initial_delay and
444	(type(self.partner_selection) == # noqa: E721
445	type(synapse_dynamics.partner_selection)) and
446	(type(self.formation) ==
447	type(synapse_dynamics.formation)) and
448	(type(self.elimination) ==
449	type(synapse_dynamics.elimination)))
450
451	@property	1✔
452	@abstractmethod	1✔
453	def connections(self) -> ConnectionsInfo:	1✔
454	"""
455	Initial connectivity as defined via connector.
456
457	:rtype: dict
458	"""
459	raise NotImplementedError
460
461	@abstractmethod	1✔
462	def _get_seeds(	1✔
463	self, app_vertex: Union[None, ApplicationVertex, Slice] = None
464	) -> Sequence[int]:
465	"""
466	Generate a seed for the RNG on chip that is the same for all
467	of the cores that have perform structural updates.
468
469	It should be different between application vertices
470	but the same for the same app_vertex.
471	It should be different every time called with `None`.
472
473	:param app_vertex:
474	:type app_vertex:
475	~pacman.model.graphs.application.ApplicationVertex or None
476	:return: list of random seed (4 words), generated randomly
477	:rtype: list(int)
478	"""
479	raise NotImplementedError
480
481	def check_initial_delay(self, max_delay_ms: float):	1✔
482	"""
483	Check that delays can be done without delay extensions.
484
485	:param float max_delay_ms: The maximum delay supported, in milliseconds
486	:raises ValueError: if the delay is out of range
487	"""
NEW 488	if isinstance(self.initial_delay, tuple):	×
489	# pylint: disable=unsubscriptable-object
490	init_del = self.initial_delay	×
491	if init_del[0] > max_delay_ms or init_del[1] > max_delay_ms:	×
492	raise ValueError(	×
493	f"The initial delay {self.initial_delay} has one or more "
494	f"values that are bigger than {max_delay_ms}.")
495	elif self.initial_delay > max_delay_ms:	×
496	raise ValueError(	×
497	f"The initial delay {self.initial_delay} "
498	f"is bigger than {max_delay_ms}.")
499
500	def get_max_rewires_per_ts(self) -> int:	1✔
501	max_rewires_per_ts = 1	×
502	if (self.p_rew * MICRO_TO_MILLISECOND_CONVERSION <	×
503	SpynnakerDataView.get_simulation_time_step_ms()):
504	# fast rewiring, so need to set max_rewires_per_ts
505	max_rewires_per_ts = int(	×
506	SpynnakerDataView.get_simulation_time_step_us() / (
507	self.p_rew * MICRO_TO_SECOND_CONVERSION))
508
509	return max_rewires_per_ts	×

SpiNNakerManchester / sPyNNaker / 6944077364

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous