Typing for Spynnaker

mypyd.bash

@@ -0,0 +1,29 @@
+#!/bin/bash
+
+# Copyright (c) 2024 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.
+
+# This bash assumes that other repositories are installed in paralled
+
+# requires the latest mypy
+# pip install --upgrade mypy
+
+utils="../SpiNNUtils/spinn_utilities"
+machine="../SpiNNMachine/spinn_machine"
+man="../SpiNNMan/spinnman"
+pacman="../PACMAN/pacman"
+spalloc="../spalloc/spalloc_client"
+fec="../SpiNNFrontEndCommon/spinn_front_end_common"
+
+mypy --python-version 3.8 --disallow-untyped-defs $utils $machine $man $pacman $spalloc $fec spynnaker

spynnaker/gsyn_tools.py

@@ -15,6 +15,9 @@
 import os
 import spynnaker.pyNN.utilities.utility_calls as utility_calls
 
+# TODO No use case found for this code
+# mypy: disable-error-code="no-untyped-def"
+
 
 def check_gsyn(gsyn1, gsyn2):
     """

spynnaker/pyNN/__init__.py

@@ -225,7 +225,8 @@ class __PynnOperations(TypedDict, total=False):
 def distance(src_cell: IDMixin, tgt_cell: IDMixin,
              mask: Optional[NDArray] = None,
              scale_factor: float = 1.0, offset: float = 0.0,
-             periodic_boundaries=None) -> float:
+             periodic_boundaries: Optional[Tuple[
+                 Optional[Tuple[int, int]]]] = None) -> float:
     """
     Return the Euclidean distance between two cells.
 
@@ -254,7 +255,8 @@ def setup(timestep: Optional[Union[float, Literal["auto"]]] = None,
           database_socket_addresses: Optional[Iterable[SocketAddress]] = None,
           time_scale_factor: Optional[int] = None,
           n_chips_required: Optional[int] = None,
-          n_boards_required: Optional[int] = None, **extra_params) -> int:
+          n_boards_required: Optional[int] = None,
+          **extra_params: Any) -> int:
     """
     The main method needed to be called to make the PyNN 0.8 setup.
     Needs to be called before any other function
@@ -391,7 +393,7 @@ def Projection(
         partition_id=partition_id)
 
 
-def _create_overloaded_functions(spinnaker_simulator: SpiNNaker):
+def _create_overloaded_functions(spinnaker_simulator: SpiNNaker) -> None:
     """
     Creates functions that the main PyNN interface supports
     (given from PyNN)
@@ -417,7 +419,7 @@ def _create_overloaded_functions(spinnaker_simulator: SpiNNaker):
     __pynn["record"] = pynn_common.build_record(spinnaker_simulator)
 
 
-def end(_=True) -> None:
+def end(_: Any = True) -> None:
     """
     Cleans up the SpiNNaker machine and software
 
@@ -456,7 +458,7 @@ def list_standard_models() -> List[str]:
 
 def set_number_of_neurons_per_core(
         neuron_type: Type,
-        max_permitted: Optional[Union[int, Tuple[int, ...]]]):
+        max_permitted: Optional[Union[int, Tuple[int, ...]]]) -> None:
     """
     Sets a ceiling on the number of neurons of a given model that can be
     placed on a single core.
@@ -495,7 +497,7 @@ def set_number_of_neurons_per_core(
 
 def connect(pre: Population, post: Population, weight: float = 0.0,
             delay: Optional[float] = None, receptor_type: Optional[str] = None,
-            p: int = 1, rng: Optional[NumpyRNG] = None):
+            p: int = 1, rng: Optional[NumpyRNG] = None) -> None:
     """
     Builds a projection.
 
@@ -586,7 +588,7 @@ def get_time_step() -> float:
     return float(__pynn["get_time_step"]())
 
 
-def initialize(cells: PopulationBase, **initial_values):
+def initialize(cells: PopulationBase, **initial_values: Any) -> None:
     """
     Sets cells to be initialised to the given values.
 
@@ -653,7 +655,7 @@ def record(variables: Union[str, Sequence[str]], source: PopulationBase,
                             annotations)
 
 
-def reset(annotations: Optional[Dict[str, Any]] = None):
+def reset(annotations: Optional[Dict[str, Any]] = None) -> None:
     """
     Resets the simulation to t = 0.
 
@@ -666,7 +668,7 @@ def reset(annotations: Optional[Dict[str, Any]] = None):
     __pynn["reset"](annotations)
 
 
-def run(simtime: float, callbacks=None) -> float:
+def run(simtime: float, callbacks: Optional[Callable] = None) -> float:
     """
     The run() function advances the simulation for a given number of
     milliseconds.

spynnaker/pyNN/connections/ethernet_command_connection.py

@@ -61,7 +61,8 @@ def __init__(
         self.__translator = translator
 
     def add_command_container(
-            self, command_container: AbstractSendMeMulticastCommandsVertex):
+            self,
+            command_container: AbstractSendMeMulticastCommandsVertex) -> None:
         """
         Add a command container.
 

spynnaker/pyNN/connections/ethernet_control_connection.py

@@ -47,7 +47,7 @@ def __init__(
             label, self._translate, translate_key=False)
 
     def add_translator(
-            self, label: str, translator: AbstractEthernetTranslator):
+            self, label: str, translator: AbstractEthernetTranslator) -> None:
         """
         Add another translator that routes via the LPG.
 
@@ -60,7 +60,8 @@ def add_translator(
         self.add_receive_no_time_callback(
             label, self._translate, translate_key=False)
 
-    def _translate(self, label: str, key: int, payload: Optional[int] = None):
+    def _translate(self, label: str, key: int,
+                   payload: Optional[int] = None) -> None:
         translator = self.__translators[label]
         if payload is None:
             translator.translate_control_packet(MultiCastCommand(key))

spynnaker/pyNN/data/spynnaker_data_writer.py

@@ -58,7 +58,7 @@ def _soft_reset(self) -> None:
     def set_up_timings_and_delay(
             self, simulation_time_step_us: Optional[int],
             time_scale_factor: Optional[float],
-            min_delay: Optional[Union[int, float]]):
+            min_delay: Optional[Union[int, float]]) -> None:
         """
         :param simulation_time_step_us:
             An explicitly specified time step for the simulation in
@@ -80,7 +80,7 @@ def set_up_timings_and_delay(
             self.__spy_data._min_delay = None
             raise
 
-    def _set_min_delay(self, min_delay: Optional[Union[int, float]]):
+    def _set_min_delay(self, min_delay: Optional[Union[int, float]]) -> None:
         """
         Sets a min delay or accepts `None` to use simulation_time_step_ms.
 

spynnaker/pyNN/external_devices/__init__.py

@@ -118,7 +118,7 @@
 __simulator: Optional[SpiNNaker] = None
 
 
-def run_forever(sync_time: float = 0.0):
+def run_forever(sync_time: float = 0.0) -> None:
     """
     Supports running forever in PyNN 0.8/0.9 format.
 
@@ -134,7 +134,7 @@ def run_forever(sync_time: float = 0.0):
     __simulator.run(None, sync_time)
 
 
-def run_sync(run_time: float, sync_time: float):
+def run_sync(run_time: float, sync_time: float) -> None:
     """
     Run in steps of the given number of milliseconds pausing between
     for a signal to be sent from the host.
@@ -167,7 +167,7 @@ def request_stop() -> None:
 
 
 def register_database_notification_request(
-        hostname: str, notify_port: int, ack_port: int):
+        hostname: str, notify_port: int, ack_port: int) -> None:
     """
     Adds a socket system which is registered with the notification protocol.
 
@@ -360,7 +360,7 @@ def SpikeInjector(
     return ExternalDeviceSpikeInjector()
 
 
-def _set_simulator(simulator: SpiNNaker):
+def _set_simulator(simulator: SpiNNaker) -> None:
     """
     Should only be called by pyNN/__init__py setup method.
 

spynnaker/pyNN/external_devices_models/abstract_ethernet_translator.py

@@ -25,7 +25,8 @@ class AbstractEthernetTranslator(object, metaclass=AbstractBase):
     __slots__ = ()
 
     @abstractmethod
-    def translate_control_packet(self, multicast_packet: MultiCastCommand):
+    def translate_control_packet(
+            self, multicast_packet: MultiCastCommand) -> None:
         """
         Translate a multicast packet received over Ethernet and send
         appropriate messages to the external device.

spynnaker/pyNN/external_devices_models/abstract_multicast_controllable_device.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from enum import Enum
+from typing import Optional
 from spinn_utilities.abstract_base import AbstractBase, abstractmethod
 
 
@@ -95,7 +96,7 @@ def device_control_max_value(self) -> float:
 
     @property
     @abstractmethod
-    def device_control_timesteps_between_sending(self) -> int:
+    def device_control_timesteps_between_sending(self) -> Optional[int]:
         """
         The number of timesteps between sending commands to the device.
         This defines the "sampling interval" for the device.
@@ -124,10 +125,8 @@ def device_control_scaling_factor(self) -> int:  # pragma: no cover
         return 1
 
     @property
-    def device_control_first_send_timestep(self) -> int:
+    def device_control_first_send_timestep(self) -> Optional[int]:
         """
         The first timestep that the device should send in (0 by default).
-
-        :rtype: int
         """
         return 0

spynnaker/pyNN/external_devices_models/external_device_lif_control.py

@@ -26,6 +26,9 @@
 from spynnaker.pyNN.models.neuron.neuron_models import (
     NeuronModelLeakyIntegrateAndFire)
 from spynnaker.pyNN.models.neuron.synapse_types import SynapseTypeExponential
+from .abstract_ethernet_translator import AbstractEthernetTranslator
+from .abstract_multicast_controllable_device import (
+    AbstractMulticastControllableDevice)
 from .external_device_lif_control_vertex import ExternalDeviceLifControlVertex
 from .threshold_type_multicast_device_control import (
     ThresholdTypeMulticastDeviceControl)
@@ -46,12 +49,15 @@ class ExternalDeviceLifControl(AbstractPyNNNeuronModelStandard):
         "tau_m", "cm", "v_rest", "v_reset", "tau_syn_E", "tau_syn_I",
         "tau_refrac", "i_offset"})
     def __init__(
-            self, devices, create_edges, translator=None,
-
+            self, devices:  List[AbstractMulticastControllableDevice],
+            create_edges: bool,
+            translator: Optional[AbstractEthernetTranslator] = None,
             # default params for the neuron model type
-            tau_m=20.0, cm=1.0, v_rest=0.0, v_reset=0.0, tau_syn_E=5.0,
-            tau_syn_I=5.0, tau_refrac=0.1, i_offset=0.0, v=0.0,
-            isyn_exc=0.0, isyn_inh=0.0):
+            tau_m: float = 20.0, cm: float = 1.0, v_rest: float = 0.0,
+            v_reset: float = 0.0, tau_syn_E: float = 5.0,
+            tau_syn_I: float = 5.0, tau_refrac: float = 0.1,
+            i_offset: float = 0.0, v: float = 0.0, isyn_exc: float = 0.0,
+            isyn_inh: float = 0.0):
         """
         :param list(AbstractMulticastControllableDevice) devices:
             The AbstractMulticastControllableDevice instances to be controlled

spynnaker/pyNN/external_devices_models/machine_munich_motor_device.py

@@ -12,13 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Sequence
+from typing import Optional, Sequence
+
 from spinn_utilities.overrides import overrides
+
 from spinnman.model.enums import ExecutableType
+
+from pacman.model.graphs.application import ApplicationVertex
 from pacman.model.graphs.common import Slice
 from pacman.model.graphs.machine import MachineVertex
 from pacman.model.placements import Placement
 from pacman.model.resources import ConstantSDRAM
+
 from spinn_front_end_common.abstract_models import (
     AbstractHasAssociatedBinary)
 from spinn_front_end_common.abstract_models import (
@@ -29,6 +34,7 @@
 from spinn_front_end_common.interface.simulation import simulation_utilities
 from spinn_front_end_common.utilities.constants import (
     SYSTEM_BYTES_REQUIREMENT, SIMULATION_N_BYTES, BYTES_PER_WORD)
+
 from spynnaker.pyNN.data import SpynnakerDataView
 
 
@@ -68,17 +74,18 @@ class MachineMunichMotorDevice(
     INPUT_BUFFER_FULL_NAME = "Times_the_input_buffer_lost_packets"
 
     def __init__(
-            self, speed, sample_time, update_time, delay_time,
-            delta_threshold, continue_if_not_different,
-            label=None, app_vertex=None):
+            self, speed: int, sample_time: int, update_time: int,
+            delay_time: int, delta_threshold: int,
+            continue_if_not_different: bool, label: Optional[str] = None,
+            app_vertex: Optional[ApplicationVertex] = None):
         """
         :param int speed:
         :param int sample_time:
         :param int update_time:
         :param int delay_time:
         :param int delta_threshold:
         :param bool continue_if_not_different:
-        :param str label:
+        :param label:
         :param ~pacman.model.graphs.application.ApplicationVertex app_vertex:
         """
         super().__init__(
@@ -120,7 +127,7 @@ def _n_additional_data_items(self) -> int:
         ProvidesProvenanceDataFromMachineImpl.parse_extra_provenance_items)
     def parse_extra_provenance_items(
             self, label: str, x: int, y: int, p: int,
-            provenance_data: Sequence[int]):
+            provenance_data: Sequence[int]) -> None:
         n_buffer_overflows, = provenance_data
 
         with ProvenanceWriter() as db:
@@ -169,7 +176,7 @@ def generate_data_specification(self, spec: DataSpecificationGenerator,
         # End-of-Spec:
         spec.end_specification()
 
-    def reserve_memory_regions(self, spec):
+    def reserve_memory_regions(self, spec: DataSpecificationGenerator) -> None:
         """
         Reserve SDRAM space for memory areas:
 

spynnaker/pyNN/external_devices_models/munich_spinnaker_link_motor_device.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Iterable
+from typing import Iterable, Optional
 from spinn_utilities.overrides import overrides
 from pacman.model.graphs.application import (
     ApplicationSpiNNakerLinkVertex)
@@ -21,25 +21,26 @@
     AbstractOneAppOneMachineVertex)
 from spinn_front_end_common.abstract_models import (
     AbstractVertexWithEdgeToDependentVertices)
-from spynnaker.pyNN.models.defaults import defaults
+from spynnaker.pyNN.models.defaults import AbstractProvidesDefaults
 from spynnaker.pyNN.models.common import PopulationApplicationVertex
 from .machine_munich_motor_device import MachineMunichMotorDevice
 
 
 class _MunichMotorDevice(ApplicationSpiNNakerLinkVertex):
     __slots__ = ()
 
-    def __init__(self, spinnaker_link_id, board_address=None):
+    def __init__(self, spinnaker_link_id: int,
+                 board_address: Optional[str] = None):
         super().__init__(
             n_atoms=6, spinnaker_link_id=spinnaker_link_id,
             label="External Munich Motor", board_address=board_address)
 
 
-@defaults
 class MunichMotorDevice(
         AbstractOneAppOneMachineVertex,
         AbstractVertexWithEdgeToDependentVertices,
-        PopulationApplicationVertex):
+        PopulationApplicationVertex,
+        AbstractProvidesDefaults):
     """
     An Omnibot motor control device. This has a real vertex and an
     external device vertex.
@@ -48,9 +49,11 @@ class MunichMotorDevice(
     __slots__ = ("__dependent_vertices", )
 
     def __init__(
-            self, spinnaker_link_id, board_address=None, speed=30,
-            sample_time=4096, update_time=512, delay_time=5,
-            delta_threshold=23, continue_if_not_different=True, label=None):
+            self, spinnaker_link_id: int, board_address: Optional[str] = None,
+            speed: int = 30, sample_time: int = 4096, update_time: int = 512,
+            delay_time: int = 5, delta_threshold: int = 23,
+            continue_if_not_different: bool = True,
+            label: Optional[str] = None):
         """
         :param int spinnaker_link_id:
             The SpiNNaker link to which the motor is connected