Merge branch 'master' into pylint

Author: Christian-B

spynnaker/pyNN/connections/spif_live_spikes_connection.py

@@ -15,7 +15,7 @@
 import logging
 import struct
 from threading import Thread
-from typing import Callable, Dict, Iterable, List, Optional, Set, Tuple
+from typing import Callable, Dict, Final, Iterable, List, Optional, Set, Tuple
 from typing_extensions import TypeAlias
 from spinn_utilities.log import FormatAdapter
 from spinnman.connections import ConnectionListener
@@ -24,9 +24,10 @@
 from spinn_front_end_common.utilities.database import DatabaseConnection
 from spinn_front_end_common.utilities.database import DatabaseReader
 
-_EventCB: TypeAlias = Callable[[str, List[int]], None]
-_InitCB: TypeAlias = Callable[[str, int, float, float], None]
-_StartStopCB: TypeAlias = Callable[[str, 'SPIFLiveSpikesConnection'], None]
+_EVENT: Final['TypeAlias'] = Callable[[str, List[int]], None]
+_INIT: Final['TypeAlias'] = Callable[[str, int, float, float], None]
+_START_STOP: Final['TypeAlias'] = Callable[
+    [str, 'SPIFLiveSpikesConnection'], None]
 logger = FormatAdapter(logging.getLogger(__name__))
 
 _TWO_SKIP = struct.Struct("<2x")
@@ -118,10 +119,10 @@ def __init__(self, receive_labels: Optional[Iterable[str]],
         self.__spif_packet_size = events_per_packet * BYTES_PER_WORD
         self.__spif_packet_time_us = time_per_packet
         self.__key_to_atom_id_and_label: Dict[int, Tuple[int, int]] = dict()
-        self.__live_event_callbacks: List[List[Tuple[_EventCB, bool]]] = list()
-        self.__start_resume_callbacks: Dict[str, List[_StartStopCB]] = dict()
-        self.__pause_stop_callbacks: Dict[str, List[_StartStopCB]] = dict()
-        self.__init_callbacks: Dict[str, List[_InitCB]] = dict()
+        self.__live_event_callbacks: List[List[Tuple[_EVENT, bool]]] = list()
+        self.__start_resume_callbacks: Dict[str, List[_START_STOP]] = dict()
+        self.__pause_stop_callbacks: Dict[str, List[_START_STOP]] = dict()
+        self.__init_callbacks: Dict[str, List[_INIT]] = dict()
         if receive_labels is not None:
             for label in receive_labels:
                 self.__live_event_callbacks.append(list())
@@ -144,7 +145,7 @@ def add_receive_label(self, label: str) -> None:
             self.__pause_stop_callbacks[label] = list()
             self.__init_callbacks[label] = list()
 
-    def add_init_callback(self, label: str, init_callback: _InitCB) -> None:
+    def add_init_callback(self, label: str, init_callback: _INIT) -> None:
         """
         Add a callback to be called to initialise a vertex.
 
@@ -161,7 +162,7 @@ def add_init_callback(self, label: str, init_callback: _InitCB) -> None:
         self.__init_callbacks[label].append(init_callback)
 
     def add_receive_callback(
-            self, label: str, live_event_callback: _EventCB,
+            self, label: str, live_event_callback: _EVENT,
             translate_key: bool = True) -> None:
         """
         Add a callback for the reception of live events from a vertex.
@@ -184,7 +185,7 @@ def add_receive_callback(
             (live_event_callback, translate_key))
 
     def add_start_resume_callback(
-            self, label: str, start_resume_callback: _StartStopCB) -> None:
+            self, label: str, start_resume_callback: _START_STOP) -> None:
         """
         Add a callback for the start and resume state of the simulation.
 
@@ -200,7 +201,7 @@ def add_start_resume_callback(
         self.__start_resume_callbacks[label].append(start_resume_callback)
 
     def add_pause_stop_callback(
-            self, label: str, pause_stop_callback: _StartStopCB) -> None:
+            self, label: str, pause_stop_callback: _START_STOP) -> None:
         """
         Add a callback for the pause and stop state of the simulation.
 
@@ -267,7 +268,7 @@ def __handle_possible_rerun_state(self) -> None:
             self.__receiver_connection = None
 
     def __launch_thread(
-            self, kind: str, label: str, callback: _StartStopCB) -> None:
+            self, kind: str, label: str, callback: _START_STOP) -> None:
         thread = Thread(target=callback, args=(label, self), name=(
             f"{kind} callback thread for live_event_connection "
             f"{self._local_port}:{self._local_ip_address}"))

spynnaker/pyNN/extra_algorithms/splitter_components/splitter_utils.py

@@ -21,13 +21,13 @@
 from spynnaker.pyNN.models.neuron.synapse_dynamics import (
     AbstractSynapseDynamics, SynapseDynamicsStatic)
 from spynnaker.pyNN.models.spike_source import SpikeSourcePoissonVertex
-from spynnaker.pyNN.types import Delay_Types
+from spynnaker.pyNN.types import DELAYS
 
 
 def is_direct_poisson_source(
         post_vertex: ApplicationVertex, pre_vertex: ApplicationVertex,
         connector: AbstractConnector, dynamics: AbstractSynapseDynamics,
-        delay: Delay_Types) -> bool:
+        delay: DELAYS) -> bool:
     """
     Determine if a given Poisson source can be created by this splitter.
 

spynnaker/pyNN/models/neural_projections/connectors/abstract_connector.py

@@ -40,7 +40,7 @@
 
 from spynnaker.pyNN.data import SpynnakerDataView
 from spynnaker.pyNN.types import (
-    Delay_Types, is_scalar, Weight_Delay_Types, Weight_Types)
+    DELAYS, is_scalar, WEIGHTS_DELAYS, WEIGHTS)
 from spynnaker.pyNN.utilities import utility_calls
 from spynnaker.pyNN.exceptions import SpynnakerException
 
@@ -122,7 +122,7 @@ def set_projection_information(
         self.__min_delay = SpynnakerDataView.get_simulation_time_step_ms()
 
     def _get_delay_minimum(
-            self, delays: Delay_Types, n_connections: int,
+            self, delays: DELAYS, n_connections: int,
             synapse_info: SynapseInformation) -> float:
         """
         Get the minimum delay given a float or RandomDistribution.
@@ -148,7 +148,7 @@ def _get_delay_minimum(
         raise self.delay_type_exception(delays)
 
     def _get_delay_maximum(
-            self, delays: Delay_Types, n_connections: int,
+            self, delays: DELAYS, n_connections: int,
             synapse_info: SynapseInformation) -> float:
         """
         Get the maximum delay given a float or RandomDistribution.
@@ -195,7 +195,7 @@ def get_delay_minimum(
         """
         raise NotImplementedError
 
-    def get_delay_variance(self, delays: Delay_Types,
+    def get_delay_variance(self, delays: DELAYS,
                            synapse_info: SynapseInformation) -> float:
         """
         Get the variance of the delays.
@@ -214,7 +214,7 @@ def get_delay_variance(self, delays: Delay_Types,
         raise self.delay_type_exception(delays)
 
     def _get_n_connections_from_pre_vertex_with_delay_maximum(
-            self, delays: Delay_Types, n_total_connections: int,
+            self, delays: DELAYS, n_total_connections: int,
             n_connections: int, min_delay: float, max_delay: float,
             synapse_info: SynapseInformation) -> int:
         """
@@ -294,7 +294,7 @@ def get_n_connections_to_post_vertex_maximum(
         """
         raise NotImplementedError
 
-    def get_weight_mean(self, weights: Weight_Types,
+    def get_weight_mean(self, weights: WEIGHTS,
                         synapse_info: SynapseInformation) -> float:
         """
         Get the mean of the weights.
@@ -313,7 +313,7 @@ def get_weight_mean(self, weights: Weight_Types,
         raise self.weight_type_exception(synapse_info)
 
     def _get_weight_maximum(
-            self, weights: Weight_Types, n_connections: int,
+            self, weights: WEIGHTS, n_connections: int,
             synapse_info: SynapseInformation) -> float:
         """
         Get the maximum of the weights.
@@ -356,7 +356,7 @@ def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:
         """
         raise NotImplementedError
 
-    def get_weight_variance(self, weights: Weight_Types,
+    def get_weight_variance(self, weights: WEIGHTS,
                             synapse_info: SynapseInformation) -> float:
         """
         Get the variance of the weights.
@@ -422,7 +422,7 @@ def _generate_random_values(
         return copy_rd.next(n_connections)
 
     def _no_space_exception(
-            self, values: Weight_Delay_Types,
+            self, values: WEIGHTS_DELAYS,
             synapse_info: SynapseInformation) -> SpynnakerException:
         """
         Returns a SpynnakerException about there being no space defined
@@ -438,7 +438,7 @@ def _no_space_exception(
             f"{synapse_info.post_population}")
 
     def weight_type_exception(
-            self, weights: Weight_Types) -> SpynnakerException:
+            self, weights: WEIGHTS) -> SpynnakerException:
         """
         Returns an Exception explaining incorrect weight or delay type
 
@@ -463,7 +463,7 @@ def weight_type_exception(
         else:
             return SpynnakerException(f"Unrecognised weight {weights}")
 
-    def delay_type_exception(self, delays: Delay_Types) -> SpynnakerException:
+    def delay_type_exception(self, delays: DELAYS) -> SpynnakerException:
         """
         Returns an Exception explaining incorrect delay type
 
@@ -485,7 +485,7 @@ def delay_type_exception(self, delays: Delay_Types) -> SpynnakerException:
             return SpynnakerException(f"Unrecognised delay {delays}")
 
     def _generate_values(
-            self, values: Weight_Delay_Types, sources: numpy.ndarray,
+            self, values: WEIGHTS_DELAYS, sources: numpy.ndarray,
             targets: numpy.ndarray, n_connections: int, post_slice: Slice,
             synapse_info: SynapseInformation,
             weights: bool) -> NDArray[float64]:

spynnaker/pyNN/models/neural_projections/connectors/abstract_generate_connector_on_machine.py

@@ -31,7 +31,7 @@
 from spynnaker.pyNN.models.common.param_generator_data import (
     param_generator_params, param_generator_params_size_in_bytes,
     param_generator_id, is_param_generatable)
-from spynnaker.pyNN.types import (Delay_Types, Weight_Types)
+from spynnaker.pyNN.types import (DELAYS, WEIGHTS)
 from spynnaker.pyNN.utilities.utility_calls import check_rng
 
 from .abstract_generate_connector_on_host import (
@@ -96,7 +96,7 @@ def generate_on_machine(self, synapse_info: SynapseInformation) -> bool:
             check_rng(synapse_info.delays.rng, "RandomDistribution in delay")
         return True
 
-    def gen_weights_id(self, weights: Weight_Types) -> int:
+    def gen_weights_id(self, weights: WEIGHTS) -> int:
         """
         Get the id of the weight generator on the machine.
 
@@ -106,7 +106,7 @@ def gen_weights_id(self, weights: Weight_Types) -> int:
         """
         return param_generator_id(weights)
 
-    def gen_weights_params(self, weights: Weight_Types) -> NDArray[uint32]:
+    def gen_weights_params(self, weights: WEIGHTS) -> NDArray[uint32]:
         """
         Get the parameters of the weight generator on the machine.
 
@@ -116,7 +116,7 @@ def gen_weights_params(self, weights: Weight_Types) -> NDArray[uint32]:
         """
         return param_generator_params(weights)
 
-    def gen_weight_params_size_in_bytes(self, weights:  Weight_Types) -> int:
+    def gen_weight_params_size_in_bytes(self, weights:  WEIGHTS) -> int:
         """
         The size of the weight parameters in bytes.
 
@@ -126,7 +126,7 @@ def gen_weight_params_size_in_bytes(self, weights:  Weight_Types) -> int:
         """
         return param_generator_params_size_in_bytes(weights)
 
-    def gen_delays_id(self, delays: Delay_Types) -> int:
+    def gen_delays_id(self, delays: DELAYS) -> int:
         """
         Get the id of the delay generator on the machine.
 
@@ -136,7 +136,7 @@ def gen_delays_id(self, delays: Delay_Types) -> int:
         """
         return param_generator_id(delays)
 
-    def gen_delay_params(self, delays: Delay_Types) -> NDArray[uint32]:
+    def gen_delay_params(self, delays: DELAYS) -> NDArray[uint32]:
         """
         Get the parameters of the delay generator on the machine.
 
@@ -146,7 +146,7 @@ def gen_delay_params(self, delays: Delay_Types) -> NDArray[uint32]:
         """
         return param_generator_params(delays)
 
-    def gen_delay_params_size_in_bytes(self, delays: Delay_Types) -> int:
+    def gen_delay_params_size_in_bytes(self, delays: DELAYS) -> int:
         """
         The size of the delay parameters in bytes.
 

spynnaker/pyNN/models/neural_projections/connectors/all_but_me_connector.py

@@ -24,7 +24,7 @@
 
 from spinn_front_end_common.utilities.constants import BYTES_PER_WORD
 from spinn_front_end_common.interface.ds import DataType
-from spynnaker.pyNN.types import Weight_Types
+from spynnaker.pyNN.types import WEIGHTS
 
 from .abstract_connector import AbstractConnector
 from .abstract_generate_connector_on_machine import (
@@ -168,7 +168,7 @@ def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:
             synapse_info)
 
     @overrides(AbstractConnector.get_weight_mean)
-    def get_weight_mean(self, weights: Weight_Types,
+    def get_weight_mean(self, weights: WEIGHTS,
                         synapse_info: SynapseInformation) -> float:
         if self.__weights is None:
             return AbstractConnector.get_weight_mean(
@@ -177,7 +177,7 @@ def get_weight_mean(self, weights: Weight_Types,
             return float(numpy.mean(numpy.abs(self.__weights)))
 
     @overrides(AbstractConnector.get_weight_variance)
-    def get_weight_variance(self, weights: Weight_Types,
+    def get_weight_variance(self, weights: WEIGHTS,
                             synapse_info: SynapseInformation) -> float:
         if self.__weights is None:
             return AbstractConnector.get_weight_variance(

spynnaker/pyNN/models/neural_projections/connectors/from_list_connector.py

@@ -33,8 +33,7 @@
 
 from spynnaker.pyNN.data import SpynnakerDataView
 from spynnaker.pyNN.exceptions import InvalidParameterType
-from spynnaker.pyNN.types import Delay_Types, Weight_Delay_Types, Weight_Types
-
+from spynnaker.pyNN.types import DELAYS, WEIGHTS_DELAYS, WEIGHTS
 from .abstract_connector import AbstractConnector
 from .abstract_generate_connector_on_host import (
     AbstractGenerateConnectorOnHost)
@@ -53,7 +52,7 @@
 _FIRST_PARAM = 2
 
 
-def _is_sequential(value: Weight_Delay_Types
+def _is_sequential(value: WEIGHTS_DELAYS
                    ) -> TypeGuard[NDArray[numpy.float64]]:
     return isinstance(value, numpy.ndarray)
 
@@ -154,7 +153,7 @@ def get_delay_minimum(self, synapse_info: SynapseInformation) -> float:
             return numpy.min(self.__delays)
 
     @overrides(AbstractConnector.get_delay_variance)
-    def get_delay_variance(self, delays: Delay_Types,
+    def get_delay_variance(self, delays: DELAYS,
                            synapse_info: SynapseInformation) -> float:
         if self.__delays is None:
             if _is_sequential(synapse_info.delays):
@@ -297,7 +296,7 @@ def get_n_connections_to_post_vertex_maximum(
             self.__targets.astype(int64, copy=False))))
 
     @overrides(AbstractConnector.get_weight_mean)
-    def get_weight_mean(self, weights: Weight_Types,
+    def get_weight_mean(self, weights: WEIGHTS,
                         synapse_info: SynapseInformation) -> float:
         if self.__weights is None:
             if _is_sequential(synapse_info.weights):
@@ -318,7 +317,7 @@ def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:
             return float(numpy.amax(numpy.abs(self.__weights)))
 
     @overrides(AbstractConnector.get_weight_variance)
-    def get_weight_variance(self, weights: Weight_Types,
+    def get_weight_variance(self, weights: WEIGHTS,
                             synapse_info: SynapseInformation) -> float:
         if self.__weights is None:
             if _is_sequential(synapse_info.weights):