diff --git a/spinn_front_end_common/utility_models/reverse_ip_tag_multi_cast_source.py b/spinn_front_end_common/utility_models/reverse_ip_tag_multi_cast_source.py index a9d20fbe8..37c74fea3 100644 --- a/spinn_front_end_common/utility_models/reverse_ip_tag_multi_cast_source.py +++ b/spinn_front_end_common/utility_models/reverse_ip_tag_multi_cast_source.py @@ -13,7 +13,7 @@ # limitations under the License. import sys -from typing import List, Optional, Tuple, Union +from typing import Optional, Tuple, Union import numpy @@ -39,7 +39,8 @@ is_array_list, ) -_SendBufferTimes = Optional[Union[numpy.ndarray, List[numpy.ndarray]]] +_SendBufferTimesIn = numpy.ndarray | list[numpy.ndarray] | None +_SendBufferTimesOut = numpy.ndarray | None class ReverseIpTagMultiCastSource(ApplicationVertex, LegacyPartitionerAPI): @@ -69,7 +70,7 @@ def __init__( check_keys: bool = False, # Send buffer parameters - send_buffer_times: _SendBufferTimes = None, + send_buffer_times: _SendBufferTimesIn = None, # Extra flag for input without a reserved port reserve_reverse_ip_tag: bool = False, @@ -135,7 +136,8 @@ def __init__( self._is_recording = False def _validate_send_buffer_times( - self, send_buffer_times: _SendBufferTimes) -> _SendBufferTimes: + self, + send_buffer_times: _SendBufferTimesIn) -> _SendBufferTimesOut: if send_buffer_times is None: return None if is_array_list(send_buffer_times): @@ -159,15 +161,26 @@ def get_sdram_used_by_atoms(self, vertex_slice: Slice) -> AbstractSDRAM: vertex_slice.n_atoms) @property - def send_buffer_times(self) -> _SendBufferTimes: + def send_buffer_times(self) -> _SendBufferTimesOut: """ When messages will be sent. """ return self.__send_buffer_times @send_buffer_times.setter - def send_buffer_times(self, send_buffer_times: _SendBufferTimes) -> None: - self.__send_buffer_times = send_buffer_times + def send_buffer_times( + self, send_buffer_times: _SendBufferTimesIn) -> None: + self.set_send_buffer_times(send_buffer_times) + + def set_send_buffer_times( + self, send_buffer_times: _SendBufferTimesIn) -> None: + """ + Sets when messages will be sent. + + :param send_buffer_times: + """ + self.__send_buffer_times = self._validate_send_buffer_times( + send_buffer_times) for vertex in self.machine_vertices: send_buffer_times_to_set = self.__send_buffer_times if is_array_list(self.__send_buffer_times): @@ -201,7 +214,7 @@ def create_machine_vertex( return machine_vertex def _filtered_send_buffer_times( - self, vertex_slice: Slice) -> _SendBufferTimes: + self, vertex_slice: Slice) -> _SendBufferTimesOut: ids = vertex_slice.get_raster_ids() send_buffer_times = self.__send_buffer_times n_buffer_times = 0