strict bytes

spinn_front_end_common/interface/buffer_management/buffer_manager.py

@@ -19,6 +19,8 @@
 import logging
 from typing import (
     Dict, Iterable, List, Optional, Set, Tuple, cast, TYPE_CHECKING)
+from typing_extensions import Never
+
 from spinn_utilities.config_holder import get_config_bool
 from spinn_utilities.log import FormatAdapter
 from spinn_utilities.ordered_set import OrderedSet
@@ -132,7 +134,7 @@ def __init__(self) -> None:
 
     def _request_data(
             self, placement_x: int, placement_y: int, address: int,
-            length: int) -> bytes:
+            length: int) -> bytearray:
         """
         Uses the extra monitor cores for data extraction.
 
@@ -176,7 +178,7 @@ def _request_data(
             return extra_mon_data
 
     @staticmethod
-    def _verify_data(extra_mon_data: bytes, txrx_data: bytes) -> None:
+    def _verify_data(extra_mon_data: bytearray, txrx_data: bytearray) -> None:
         sm = difflib.SequenceMatcher(a=extra_mon_data, b=txrx_data)
         failed_index = -1
         for (tag, i1, i2, j1, j2) in sm.get_opcodes():
@@ -406,8 +408,9 @@ def __python_extract_no_monitors(
         for placement in progress.over(recording_placements):
             self._retreive_by_placement(placement)
 
-    def get_data_by_placement(self, placement: Placement,
-                              recording_region_id: int) -> Tuple[bytes, bool]:
+    def get_data_by_placement(
+            self, placement: Placement,
+            recording_region_id: int) -> Tuple[memoryview, bool]:
         """
         Deprecated use get_recording or get_download
 
@@ -432,7 +435,7 @@ def get_data_by_placement(self, placement: Placement,
                 f"Unable to get data for vertex {placement.vertex}")
 
     def get_recording(self, placement: Placement,
-                      recording_region_id: int) -> Tuple[bytes, bool]:
+                      recording_region_id: int) -> Tuple[memoryview, bool]:
         """
         Get the data container for the data retrieved
         during the simulation from a specific region area of a core.
@@ -457,7 +460,7 @@ def get_recording(self, placement: Placement,
                 placement, recording_region_id, lookup_error)
 
     def get_download(self, placement: Placement,
-                     recording_region_id: int) -> Tuple[bytes, bool]:
+                     recording_region_id: int) -> Tuple[memoryview, bool]:
         """
         Get the data container for the data retrieved
         during the simulation from a specific region area of a core.
@@ -479,11 +482,11 @@ def get_download(self, placement: Placement,
                     placement.x, placement.y, placement.p,
                     recording_region_id, -1)
         except LookupError as lookup_error:
-            return self._raise_error(
+            self._raise_error(
                 placement, recording_region_id, lookup_error)
 
     def _raise_error(self, placement: Placement, recording_region_id: int,
-                     lookup_error: LookupError) -> Tuple[bytes, bool]:
+                     lookup_error: LookupError) -> Never:
         """
         Raises the correct exception-
         """

spinn_front_end_common/interface/buffer_management/storage_objects/buffer_database.py

@@ -338,7 +338,7 @@ def get_last_extraction_id(self) -> int:
         raise LookupError("No Extraction id found")
 
     def store_recording(self, x: int, y: int, p: int, region: int,
-                        missing: bool, data: bytes) -> None:
+                        missing: bool, data: bytearray | bytes) -> None:
         """
         Store some information in the corresponding buffer for a
         specific chip, core and recording region.
@@ -368,7 +368,7 @@ def store_recording(self, x: int, y: int, p: int, region: int,
 
     def store_download(
             self, x: int, y: int, p: int, region: int, missing: bool,
-            data: bytes) -> None:
+            data: bytearray | bytes) -> None:
         """
         Store some information in the corresponding buffer for a
         specific chip, core and recording region.

spinn_front_end_common/interface/ds/ds_sqllite_database.py

@@ -287,7 +287,7 @@ def get_double_region(self) -> Iterable[Tuple[int, int, int, int]]:
             yield (row["x"], row["y"], row["p"], row["region_num"])
 
     def set_region_content(
-            self, x: int, y: int, p: int, region_num: int, content: bytes,
+            self, x: int, y: int, p: int, region_num: int, content: bytearray,
             content_debug: Optional[str]) -> None:
         """
         Sets the content for this region
@@ -458,7 +458,7 @@ def set_region_pointer(self, x: int, y: int, p: int, region_num: int,
 
     def get_region_pointers_and_content(
             self, x: int, y: int, p: int) -> Iterable[Tuple[
-                int, int, Optional[bytes]]]:
+                int, int, Optional[bytearray]]]:
         """
         Yields the number, pointers and content for each reserved region
 
@@ -485,8 +485,8 @@ def get_region_pointers_and_content(
                 content = None
             yield row["region_num"], row["pointer"], content
 
-    def get_regions_content(
-            self, x: int, y: int, p: int) -> Iterable[Tuple[int, int, bytes]]:
+    def get_regions_content(self, x: int, y: int,
+                            p: int) -> Iterable[Tuple[int, int, bytearray]]:
         """
         Yields the number, pointers and content for each region
 

spinn_front_end_common/interface/interface_functions/host_no_bitfield_router_compression.py

@@ -228,4 +228,4 @@ def _build_data(self, table: AbstractMulticastRoutingTable) -> bytes:
                 entry.key, entry.mask,
                 Router.convert_routing_table_entry_to_spinnaker_route(entry),
                 get_defaultable_source_id(entry))
-        return bytearray(data)
+        return data

spinn_front_end_common/interface/interface_functions/load_data_specification.py

@@ -34,7 +34,7 @@
 from spinn_front_end_common.interface.ds import DsSqlliteDatabase
 
 logger = FormatAdapter(logging.getLogger(__name__))
-_Writer: TypeAlias = Callable[[int, int, int, bytes], Any]
+_Writer: TypeAlias = Callable[[int, int, int, bytearray | bytes], Any]
 
 MONITOR_CUTOFF = 12800  # 50 packets of 256 bytes
 

spinn_front_end_common/interface/profiling/profile_data.py

@@ -58,7 +58,7 @@ def __init__(self, tag_labels: Mapping[int, str]):
         self._tags: Dict[str, Tuple[numpy.ndarray, numpy.ndarray]] = dict()
         self._max_time: float = 0.0
 
-    def add_data(self, data: bytes) -> None:
+    def add_data(self, data: bytearray) -> None:
         """
         Add profiling data read from the profile section.
 

spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/set_reinjection_packet_types_message.py

@@ -53,7 +53,7 @@ def __init__(self, x: int, y: int, p: int, multicast: bool,
             argument_1=int(bool(multicast)),
             argument_2=int(bool(point_to_point)),
             argument_3=int(bool(fixed_route)),
-            data=bytearray(struct.pack("<B", nearest_neighbour)))
+            data=struct.pack("<B", nearest_neighbour))
 
     @overrides(AbstractSCPRequest.get_scp_response)
     def get_scp_response(self) -> CheckOKResponse:

spinn_front_end_common/utility_models/data_speed_up_packet_gatherer_machine_vertex.py

@@ -19,8 +19,7 @@
 import struct
 from enum import Enum, IntEnum
 from typing import (
-    Any, BinaryIO, Iterable, List, Optional, Set, Tuple, Union,
-    TYPE_CHECKING)
+    Any, BinaryIO, Iterable, List, Optional, Set, Tuple, TYPE_CHECKING)
 
 from spinn_utilities.config_holder import get_config_bool, get_report_path
 from spinn_utilities.overrides import overrides
@@ -489,7 +488,7 @@ def _generate_data_in_report(
 
     def send_data_into_spinnaker(
             self, x: int, y: int, base_address: int,
-            data: Union[BinaryIO, bytes, str, int], *,
+            data: BinaryIO | bytearray | bytes | str | int, *,
             n_bytes: Optional[int] = None, offset: int = 0) -> None:
         """
         Sends a block of data into SpiNNaker to a given chip.
@@ -606,7 +605,7 @@ def __open_connection(self) -> SCAMPConnection:
 
     def _send_data_via_extra_monitors(
             self, destination_chip: Chip, start_address: int,
-            data_to_write: bytes) -> None:
+            data_to_write: bytearray | bytes) -> None:
         """
         Sends data using the extra monitor cores.
 
@@ -739,7 +738,7 @@ def _read_in_missing_seq_nums(
         return seen_last, seen_all
 
     def _outgoing_retransmit_missing_seq_nums(
-            self, data_to_write: bytes, missing: Set[int],
+            self, data_to_write: bytearray | bytes, missing: Set[int],
             connection: SCAMPConnection) -> None:
         """
         Transmits back into SpiNNaker the missing data based off missing
@@ -772,7 +771,7 @@ def __position_from_seq_number(seq_num: int) -> int:
         return BYTES_IN_FULL_PACKET_WITH_KEY * seq_num
 
     def __make_data_in_stream_message(
-            self, data_to_write: bytes, seq_num: int,
+            self, data_to_write: bytearray | bytes, seq_num: int,
             position: Optional[int]) -> Tuple[SDPMessage, int]:
         """
         Determine the data needed to be sent to the SpiNNaker machine
@@ -833,7 +832,7 @@ def __send_tell_flag(self, connection: SCAMPConnection) -> None:
                 _DataInCommands.SEND_TELL, self._transaction_id)))
 
     def _send_all_data_based_packets(
-            self, data_to_write: bytes, start_address: int,
+            self, data_to_write: bytearray | bytes, start_address: int,
             connection: SCAMPConnection) -> None:
         """
         Send all the data as one block.
@@ -994,7 +993,7 @@ def unset_cores_for_data_streaming(self) -> None:
     def get_data(
             self, extra_monitor: ExtraMonitorSupportMachineVertex,
             placement: Placement, memory_address: int,
-            length_in_bytes: int) -> bytes:
+            length_in_bytes: int) -> bytearray:
         """
         Gets data from a given core and memory address.
 
@@ -1243,7 +1242,7 @@ def _determine_and_retransmit_missing_seq_nums(
 
             # build SDP message and send it to the core
             connection.send_sdp_message(self.__make_data_out_message(
-                placement, data))
+                placement, bytes(data)))
 
             # sleep for ensuring core doesn't lose packets
             time.sleep(self._TIMEOUT_FOR_SENDING_IN_SECONDS)

unittests/interface/interface_functions/test_front_end_common_dsg_region_reloader.py

@@ -111,7 +111,7 @@ def __init__(self) -> None:
     @overrides(MockableTransceiver.write_memory)
     def write_memory(
             self, x: int, y: int, base_address: int,
-            data: Union[BinaryIO, bytes, int, str], *,
+            data:  Union[BinaryIO, bytearray, bytes, int, str], *,
             n_bytes: Optional[int] = None, offset: int = 0, cpu: int = 0,
             get_sum: bool = False) -> Tuple[int, int]:
         self._regions_rewritten.append((base_address, data))

unittests/interface/interface_functions/test_load_data_specification.py

@@ -68,7 +68,7 @@ def malloc_sdram(
     @overrides(Version5Transceiver.write_memory)
     def write_memory(
             self, x: int, y: int, base_address: int,
-            data: Union[BinaryIO, bytes, int, str], *,
+            data:  Union[BinaryIO, bytearray, bytes, int, str], *,
             n_bytes: Optional[int] = None, offset: int = 0, cpu: int = 0,
             get_sum: bool = False) -> Tuple[int, int]:
         if isinstance(data, int):