typing

spinnman/connections/abstract_classes/abstract_scp_connection.py

@@ -26,7 +26,7 @@ class AbstractSCPConnection(Connection, metaclass=AbstractBase):
     __slots__ = ()
 
     @abstractmethod
-    def is_ready_to_receive(self, timeout: float = 0):
+    def is_ready_to_receive(self, timeout: float = 0) -> bool:
         """
         Determines if there is an SCP packet to be read without blocking.
 

spinnman/connections/connection_listener.py

@@ -58,7 +58,7 @@ def __init__(self, connection: Listenable[T],
         self.__done = False
         self.__callbacks: List[Callable[[T], None]] = []
 
-    def __run_step(self, handler: Callable[[], T]):
+    def __run_step(self, handler: Callable[[], T]) -> None:
         """
         :param ~collections.abc.Callable handler:
         """
@@ -69,7 +69,7 @@ def __run_step(self, handler: Callable[[], T]):
                     callback, message)
                 future.add_done_callback(self.__done_callback)
 
-    def __done_callback(self, future: Future[None]):
+    def __done_callback(self, future: Future[None]) -> None:
         """
         :param ~concurrent.futures.Future future:
         """
@@ -94,7 +94,7 @@ def run(self) -> None:
                         logger.warning("problem when dispatching message",
                                        exc_info=True)
 
-    def add_callback(self, callback: Callable[[T], None]):
+    def add_callback(self, callback: Callable[[T], None]) -> None:
         """
         Add a callback to be called when a message is received.
 

spinnman/connections/scp_request_pipeline.py

@@ -16,7 +16,8 @@
 from threading import RLock
 import time
 from types import TracebackType
-from typing import Callable, Dict, Generic, List, Optional, TypeVar, cast
+from typing import (Callable, Dict, Generic, List, Optional, Set, TypeVar,
+                    cast)
 from typing_extensions import TypeAlias
 from spinnman.messages.scp.enums import SCPResult
 from spinnman.exceptions import SpinnmanTimeoutException, SpinnmanIOException
@@ -77,10 +78,11 @@ class SCPRequestPipeLine(Generic[R]):
         "_retries",
         "_send_time")
 
-    def __init__(self, connection: SCAMPConnection, n_channels=1,
-                 intermediate_channel_waits=0,
-                 n_retries=N_RETRIES, packet_timeout=SCP_TIMEOUT,
-                 non_fail_retry_codes=None):
+    def __init__(self, connection: SCAMPConnection, n_channels: int = 1,
+                 intermediate_channel_waits: int = 0,
+                 n_retries: int = N_RETRIES,
+                 packet_timeout: float = SCP_TIMEOUT,
+                 non_fail_retry_codes: Optional[Set[SCPResult]] = None):
         """
         :param SCAMPConnection connection:
             The connection over which the communication is to take place
@@ -134,10 +136,11 @@ def __init__(self, connection: SCAMPConnection, n_channels=1,
         # The number of packets that have been resent
         self._n_resent = 0
         self._n_retry_code_resent = 0
-        self._non_fail_retry_codes = non_fail_retry_codes
-        if self._non_fail_retry_codes is None:
+        self._non_fail_retry_codes: Set[SCPResult]
+        if non_fail_retry_codes is None:
             self._non_fail_retry_codes = set()
-
+        else:
+            self._non_fail_retry_codes = non_fail_retry_codes
         # self._token_bucket = TokenBucket(43750, 4375000)
         # self._token_bucket = TokenBucket(3408, 700000)
 
@@ -159,7 +162,7 @@ def __get_next_sequence_number() -> int:
 
     def send_request(
             self, request: AbstractSCPRequest[R], callback: Optional[CB],
-            error_callback: ECB):
+            error_callback: ECB) -> None:
         """
         Add an SCP request to the set to be sent.
 
@@ -260,7 +263,7 @@ def _remove_record(self, seq: int) -> None:
         del self._error_callbacks[seq]
         del self._retry_reason[seq]
 
-    def _single_retrieve(self, timeout: float):
+    def _single_retrieve(self, timeout: float) -> None:
         # Receive the next response
         result, seq, raw_data, offset = \
             self._connection.receive_scp_response(timeout)
@@ -320,7 +323,8 @@ def _handle_receive_timeout(self) -> None:
         for seq in to_remove:
             self._remove_record(seq)
 
-    def _resend(self, seq: int, request_sent, reason: str):
+    def _resend(self, seq: int, request_sent: AbstractSCPRequest,
+                reason: str) -> None:
         if self._retries[seq] <= 0:
             # Report timeouts as timeout exception
 
@@ -346,7 +350,7 @@ def _resend(self, seq: int, request_sent, reason: str):
         self._connection.send(self._request_data[seq])
         self._n_resent += 1
 
-    def _do_retrieve(self, n_packets: int, timeout: float):
+    def _do_retrieve(self, n_packets: int, timeout: float) -> None:
         """
         Receives responses until there are only n_packets responses left.
 

spinnman/connections/udp_packet_connections/bmp_connection.py

@@ -83,7 +83,7 @@ def receive_scp_response(self, timeout: Optional[float] = 1.0) -> Tuple[
         result, sequence = _TWO_SHORTS.unpack_from(data, 10)
         return SCPResult(result), sequence, data, 2
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return (
             f"BMPConnection("
             f"boards={self._boards}, local_host={self.local_ip_address}, "

spinnman/connections/udp_packet_connections/boot_connection.py

@@ -44,7 +44,7 @@ def __init__(self, remote_host: Optional[str] = None):
         super().__init__(remote_host=remote_host,
                          remote_port=UDP_BOOT_CONNECTION_DEFAULT_PORT)
 
-    def send_boot_message(self, boot_message: SpinnakerBootMessage):
+    def send_boot_message(self, boot_message: SpinnakerBootMessage) -> None:
         """
         Sends a SpiNNaker boot message using this connection.
 
@@ -80,7 +80,7 @@ def receive_boot_message(
         data = self.receive(timeout)
         return SpinnakerBootMessage.from_bytestring(data, 0)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return self._REPR_TEMPLATE.format(
             self.local_ip_address, self.local_port,
             self.remote_ip_address, self.remote_port)

spinnman/connections/udp_packet_connections/eieio_connection.py

@@ -61,7 +61,7 @@ def receive_eieio_message(
             return read_eieio_command_message(data, 0)
         return read_eieio_data_message(data, 0)
 
-    def send_eieio_message(self, eieio_message: AbstractEIEIOMessage):
+    def send_eieio_message(self, eieio_message: AbstractEIEIOMessage) -> None:
         """
         Sends an EIEIO message down this connection.
 
@@ -74,7 +74,7 @@ def send_eieio_message(self, eieio_message: AbstractEIEIOMessage):
 
     def send_eieio_message_to(
             self, eieio_message: AbstractEIEIOMessage,
-            ip_address: str, port: int):
+            ip_address: str, port: int) -> None:
         """
 
         :param AbstractEIEIOMessage eieio_message:

spinnman/connections/udp_packet_connections/ip_address_connection.py

@@ -26,8 +26,8 @@ class IPAddressesConnection(UDPConnection):
     """
     __slots__ = ()
 
-    def __init__(self, local_host=None,
-                 local_port=UDP_BOOT_CONNECTION_DEFAULT_PORT):
+    def __init__(self, local_host: Optional[str] = None,
+                 local_port: int = UDP_BOOT_CONNECTION_DEFAULT_PORT):
         super().__init__(local_host=local_host, local_port=local_port)
 
     def receive_ip_address(self, timeout:  Optional[float] = None
@@ -44,6 +44,6 @@ def receive_ip_address(self, timeout:  Optional[float] = None
                 return ip_address
         return None
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"IPAddressesConnection(local_host={self.local_ip_address}," \
                f" local_port={self.local_port})"

spinnman/connections/udp_packet_connections/scamp_connection.py

@@ -68,7 +68,7 @@ def chip_x(self) -> int:
     def chip_y(self) -> int:
         return self._chip_y
 
-    def update_chip_coordinates(self, x: int, y: int):
+    def update_chip_coordinates(self, x: int, y: int) -> None:
         """
         Sets the coordinates without checking they are valid.
 

spinnman/connections/udp_packet_connections/sdp_connection.py

@@ -84,7 +84,7 @@ def receive_sdp_message(
         data = self.receive(timeout)
         return SDPMessage.from_bytestring(data, 2)
 
-    def send_sdp_message(self, sdp_message: SDPMessage):
+    def send_sdp_message(self, sdp_message: SDPMessage) -> None:
         """
         Sends an SDP message down this connection.
 

spinnman/connections/udp_packet_connections/udp_connection.py

@@ -219,7 +219,7 @@ def receive_with_address(self, timeout: Optional[float] = None) -> Tuple[
             raise SpinnmanEOFException()
         return receive_message_and_address(self._socket, timeout, _MSG_MAX)
 
-    def send(self, data: bytes):
+    def send(self, data: bytes) -> None:
         """
         Send data down this connection.
 
@@ -237,7 +237,7 @@ def send(self, data: bytes):
             if self.__is_closed:
                 raise SpinnmanEOFException()
 
-    def send_to(self, data: bytes, address: Tuple[str, int]):
+    def send_to(self, data: bytes, address: Tuple[str, int]) -> None:
         """
         Send data down this connection.
 
@@ -261,7 +261,7 @@ def close(self) -> None:
             self._socket.shutdown(socket.SHUT_WR)
         self._socket.close()
 
-    def is_ready_to_receive(self, timeout: float = 0):
+    def is_ready_to_receive(self, timeout: float = 0) -> bool:
         if self.__is_closed:
             return True
         return len(select.select([self._socket], [], [], timeout)[0]) == 1

spinnman/exceptions.py

@@ -14,7 +14,7 @@
 from __future__ import annotations
 import traceback
 from types import TracebackType
-from typing import List, Optional, FrozenSet, Union, TYPE_CHECKING
+from typing import Any, List, Optional, FrozenSet, Union, TYPE_CHECKING
 if TYPE_CHECKING:
     from spinnman.messages.scp.enums import SCPResult
     from spinnman.model.enums import CPUState
@@ -192,11 +192,11 @@ class SpinnmanTimeoutException(SpinnmanException):
     could finish.
     """
 
-    def __init__(self, operation: str, timeout: Optional[float],
+    def __init__(self, operation: Any, timeout: Optional[float],
                  msg: Optional[str] = None):
         """
-        :param str operation: The operation being performed
-        :param float timeout: The timeout value in seconds
+        :param operation: The operation being performed
+        :param timeout: The timeout value in seconds
         """
         if msg is None:
             msg = f"Operation {operation} timed out after {timeout} seconds"

spinnman/extended/de_alloc_sdram_process.py

@@ -13,7 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from typing import Optional
-from spinnman.messages.scp.impl import SDRAMDeAlloc
+from spinnman.messages.scp.impl.sdram_de_alloc import (
+    SDRAMDeAlloc, _SCPSDRAMDeAllocResponse)
 from spinnman.processes import AbstractMultiConnectionProcess
 from spinnman.processes import ConnectionSelector
 
@@ -27,14 +28,14 @@ class DeAllocSDRAMProcess(AbstractMultiConnectionProcess):
     """
     __slots__ = ("_no_blocks_freed", )
 
-    def __init__(self, connection_selector: ConnectionSelector):
+    def __init__(self, connection_selector: ConnectionSelector) -> None:
         """
         :param ConnectionSelector connection_selector:
         """
         super().__init__(connection_selector)
         self._no_blocks_freed: Optional[int] = None
 
-    def de_alloc_all_app_sdram(self, x: int, y: int, app_id: int):
+    def de_alloc_all_app_sdram(self, x: int, y: int, app_id: int) -> None:
         """
         :param int x:
         :param int y:
@@ -47,7 +48,7 @@ def de_alloc_all_app_sdram(self, x: int, y: int, app_id: int):
             self._send_request(SDRAMDeAlloc(x, y, app_id=app_id),
                                callback=self.__handle_sdram_alloc_response)
 
-    def de_alloc_sdram(self, x: int, y: int, base_address: int):
+    def de_alloc_sdram(self, x: int, y: int, base_address: int) -> None:
         """
         :param int x:
         :param int y:
@@ -57,7 +58,8 @@ def de_alloc_sdram(self, x: int, y: int, base_address: int):
             self._send_request(SDRAMDeAlloc(x, y, base_address=base_address),
                                callback=None)
 
-    def __handle_sdram_alloc_response(self, response):
+    def __handle_sdram_alloc_response(
+            self, response: _SCPSDRAMDeAllocResponse) -> None:
         self._no_blocks_freed = response.number_of_blocks_freed
 
     @property

spinnman/extended/read_adc.py

@@ -62,7 +62,7 @@ def _parse_payload(self, data: bytes, offset: int) -> ADCInfo:
         return ADCInfo(data, offset)
 
     @property
-    def adc_info(self):
+    def adc_info(self) -> ADCInfo:
         """
         The ADC information.
         """

spinnman/extended/write_memory_flood_process.py

@@ -32,20 +32,21 @@ def __init__(self, next_connection_selector: ConnectionSelector):
             self, next_connection_selector, n_channels=3,
             intermediate_channel_waits=2)
 
-    def _start_flood_fill(self, n_bytes: int, nearest_neighbour_id: int):
+    def _start_flood_fill(
+            self, n_bytes: int, nearest_neighbour_id: int) -> None:
         n_blocks = int(math.ceil(math.ceil(n_bytes / 4.0) /
                                  UDP_MESSAGE_MAX_SIZE))
         with self._collect_responses():
             self._send_request(
                 FloodFillStart(nearest_neighbour_id, n_blocks))
 
-    def _end_flood_fill(self, nearest_neighbour_id: int):
+    def _end_flood_fill(self, nearest_neighbour_id: int) -> None:
         with self._collect_responses():
             self._send_request(FloodFillEnd(nearest_neighbour_id))
 
     def write_memory_from_bytearray(
             self, nearest_neighbour_id: int, base_address: int,
-            data: bytes, offset: int, n_bytes: Optional[int] = None):
+            data: bytes, offset: int, n_bytes: Optional[int] = None) -> None:
         """
         :param int nearest_neighbour_id:
         :param int base_address:
@@ -79,7 +80,7 @@ def write_memory_from_bytearray(
 
     def write_memory_from_reader(
             self, nearest_neighbour_id: int, base_address: int,
-            reader: BinaryIO, n_bytes: int):
+            reader: BinaryIO, n_bytes: int) -> None:
         """
         :param int nearest_neighbour_id:
         :param int base_address:

spinnman/messages/scp/impl/app_copy_run.py

@@ -70,7 +70,7 @@ def __init__(self, x: int, y: int, link: int, size: int, app_id: int,
             SCPRequestHeader(command=SCPCommand.CMD_APP_COPY_RUN),
             argument_1=arg1, argument_2=size, argument_3=processor_mask)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"{super(AppCopyRun, self).__repr__()} (Link {self.__link})"
 
     @overrides(AbstractSCPRequest.get_scp_response)

spinnman/messages/scp/impl/app_stop.py

@@ -22,7 +22,7 @@
 _APP_MASK = 0xFF
 
 
-def _get_data(app_id, signal):
+def _get_data(app_id: int, signal: Signal) -> int:
     data = (_APP_MASK << 8) | app_id
     data += signal.value << 16
     return data

spinnman/messages/scp/impl/check_ok_response.py

@@ -12,6 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from enum import Enum
+from typing import Union
+
 from spinn_utilities.overrides import overrides
 from spinnman.messages.scp.abstract_messages import AbstractSCPResponse
 from spinnman.messages.scp.enums import SCPResult
@@ -26,7 +29,7 @@ class CheckOKResponse(AbstractSCPResponse):
         "_command",
         "_operation")
 
-    def __init__(self, operation: str, command):
+    def __init__(self, operation: str, command: Union[Enum, int, str]):
         """
         :param str operation: The operation being performed
         :param command: The command that was sent
@@ -37,7 +40,7 @@ def __init__(self, operation: str, command):
         self._command = command
 
     @overrides(AbstractSCPResponse.read_data_bytestring)
-    def read_data_bytestring(self, data: bytes, offset: int):
+    def read_data_bytestring(self, data: bytes, offset: int) -> None:
         result = self.scp_response_header.result
         if result != SCPResult.RC_OK:
             raise SpinnmanUnexpectedResponseCodeException(