6574804013

Committed 19 Oct 2023 12:47PM UTC coverage: 51.937% (+1.2%) from 50.777%

Build # 6574804013

Build Type

Pull #327

github

Committed by

Christian-B

Commit Message

typing changes

Pull Request Pull Request #327: Type Annotations and Checking

Run Details

105 of 1288 branches covered (0.0%)

Branch coverage included in aggregate %.

2375 of 2375 new or added lines in 180 files covered. (100.0%)

4775 of 8108 relevant lines covered (58.89%)

0.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.82

/spinnman/transceiver/mockable_transceiver.py

# Copyright (c) 2014 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.

# pylint: disable=too-many-arguments

from typing import (
    BinaryIO, Collection, Dict, FrozenSet, Iterable,
    List, Optional, Tuple, Union)
from spinn_utilities.overrides import overrides
from spinn_utilities.progress_bar import ProgressBar
from spinn_utilities.typing.coords import XY
from spinn_machine import (
    CoreSubsets, FixedRouteEntry, Machine, MulticastRoutingEntry)
from spinn_machine.tags import AbstractTag, IPTag, ReverseIPTag
from spinnman.data import SpiNNManDataView
from spinnman.connections.udp_packet_connections import BMPConnection
from spinnman.connections.udp_packet_connections import (
    SCAMPConnection, SDPConnection)
from spinnman.processes import ConnectionSelector, FixedConnectionSelector
from spinnman.messages.scp.enums import Signal
from spinnman.messages.sdp import SDPMessage
from spinnman.model import (
    CPUInfos, DiagnosticFilter, IOBuffer, RouterDiagnostics,
    VersionInfo)
from spinnman.model.enums import CPUState, UserRegister
from spinnman.transceiver.transceiver import Transceiver
from spinnman.transceiver.extendable_transceiver import ExtendableTransceiver
from spinnman.processes import MostDirectConnectionSelector


class MockableTransceiver(ExtendableTransceiver):
    """
    A based for Mock Transceivers
    """
    __slots__ = ["written_memory"]

    def __init__(self) -> None:
        super().__init__()
        self.written_memory: List[
            Tuple[int, int, int, Union[BinaryIO, bytes, int, str],
                  Optional[int], int, int]] = []

    @overrides(Transceiver.send_sdp_message)
    def send_sdp_message(self, message: SDPMessage,
                         connection: Optional[SDPConnection] = None):
        pass

    @overrides(Transceiver.discover_scamp_connections)
    def discover_scamp_connections(self) -> None:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.add_scamp_connections)
    def add_scamp_connections(self, connections: Dict[XY, str]):
        pass

    @overrides(Transceiver.get_machine_details)
    def get_machine_details(self) -> Machine:
        return SpiNNManDataView.get_machine()

    @overrides(Transceiver.get_connections)
    def get_connections(self):
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_cpu_infos)
    def get_cpu_infos(
            self, core_subsets: Optional[CoreSubsets] = None,
            states: Union[CPUState, Iterable[CPUState], None] = None,
            include: bool = True) -> CPUInfos:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_clock_drift)
    def get_clock_drift(self, x, y):
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.read_user)
    def read_user(self, x: int, y: int, p: int, user: UserRegister):
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.add_cpu_information_from_core)
    def add_cpu_information_from_core(
            self, cpu_infos: CPUInfos, x: int, y: int, p: int,
            states: Iterable[CPUState]):
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_region_base_address)
    def get_region_base_address(self, x: int, y: int, p: int):
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_iobuf)
    def get_iobuf(self, core_subsets: Optional[CoreSubsets] = None
                  ) -> Iterable[IOBuffer]:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_core_state_count)
    def get_core_state_count(
            self, app_id: int, state: CPUState,
            xys: Optional[Iterable[Tuple[int, int]]] = None) -> int:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.execute_flood)
    def execute_flood(
            self, core_subsets: CoreSubsets,
            executable: Union[BinaryIO, bytes, str], app_id: int, *,
            n_bytes: Optional[int] = None, wait: bool = False):
        pass

    @overrides(Transceiver.read_fpga_register)
    def read_fpga_register(
            self, fpga_num: int, register: int, board: int = 0) -> int:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.write_fpga_register)
    def write_fpga_register(
            self, fpga_num: int, register: int, value: int, board: int = 0):
        pass

    @overrides(Transceiver.read_bmp_version)
    def read_bmp_version(self, board: int) -> VersionInfo:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.write_memory)
    def write_memory(
            self, x: int, y: int, base_address: int,
            data: Union[BinaryIO, bytes, int, str], *,
            n_bytes: Optional[int] = None, offset: int = 0, cpu: int = 0,
            get_sum: bool = False) -> Tuple[int, int]:
        print("Doing write to", x, y)
        self.written_memory.append(
            (x, y, base_address, data, n_bytes, offset, cpu))
        # Hope the return is never used as it will be wrong
        return (-1, -1)

    @overrides(Transceiver.write_user)
    def write_user(
            self, x: int, y: int, p: int, user: UserRegister, value: int):
        pass

    @overrides(Transceiver.read_memory)
    def read_memory(
            self, x: int, y: int, base_address: int, length: int,
            cpu: int = 0) -> bytes:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.read_word)
    def read_word(
            self, x: int, y: int, base_address: int, cpu: int = 0) -> int:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.stop_application)
    def stop_application(self, app_id: int):
        pass

    @overrides(Transceiver.wait_for_cores_to_be_in_state)
    def wait_for_cores_to_be_in_state(
            self, all_core_subsets: CoreSubsets, app_id: int,
            cpu_states: Union[CPUState, Iterable[CPUState]], *,
            timeout: Optional[float] = None,
            time_between_polls: float = 0.1,
            error_states: FrozenSet[CPUState] = frozenset((
                CPUState.RUN_TIME_EXCEPTION, CPUState.WATCHDOG)),
            counts_between_full_check: int = 100,
            progress_bar: Optional[ProgressBar] = None):
        pass

    @overrides(Transceiver.send_signal)
    def send_signal(self, app_id: int, signal: Signal):
        pass

    @overrides(Transceiver.set_ip_tag)
    def set_ip_tag(self, ip_tag: IPTag, use_sender: bool = False):
        pass

    @overrides(Transceiver.set_reverse_ip_tag)
    def set_reverse_ip_tag(self, reverse_ip_tag: ReverseIPTag):
        pass

    @overrides(Transceiver.clear_ip_tag)
    def clear_ip_tag(self, tag: int, board_address: Optional[str] = None):
        pass

    @overrides(Transceiver.get_tags)
    def get_tags(self, connection: Optional[SCAMPConnection] = None
                 ) -> Iterable[AbstractTag]:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.malloc_sdram)
    def malloc_sdram(
            self, x: int, y: int, size: int, app_id: int, tag=0) -> int:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.load_multicast_routes)
    def load_multicast_routes(
            self, x: int, y: int, routes: Collection[MulticastRoutingEntry],
            app_id: int):
        pass

    @overrides(Transceiver.load_fixed_route)
    def load_fixed_route(
            self, x: int, y: int, fixed_route: FixedRouteEntry, app_id: int):
        pass

    @overrides(Transceiver.read_fixed_route)
    def read_fixed_route(self, x: int, y: int, app_id: int) -> FixedRouteEntry:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_multicast_routes)
    def get_multicast_routes(
            self, x: int, y: int,
            app_id: Optional[int] = None) -> List[MulticastRoutingEntry]:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.clear_multicast_routes)
    def clear_multicast_routes(self, x: int, y: int):
        pass

    @overrides(Transceiver.get_router_diagnostics)
    def get_router_diagnostics(self, x: int, y: int) -> RouterDiagnostics:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.get_scamp_connection_selector)
    def get_scamp_connection_selector(self) -> MostDirectConnectionSelector:
        raise NotImplementedError("Needs to be mocked")

    @overrides(Transceiver.set_router_diagnostic_filter)
    def set_router_diagnostic_filter(
            self, x: int, y: int, position: int,
            diagnostic_filter: DiagnosticFilter):
        pass

    @overrides(Transceiver.clear_router_diagnostic_counters)
    def clear_router_diagnostic_counters(self, x: int, y: int):
        pass

    @overrides(Transceiver.close)
    def close(self) -> None:
        pass

    @overrides(Transceiver.control_sync)
    def control_sync(self, do_sync: bool):
        pass

    @overrides(Transceiver.update_provenance_and_exit)
    def update_provenance_and_exit(self, x: int, y: int, p: int):
        pass

    @overrides(Transceiver.send_chip_update_provenance_and_exit)
    def send_chip_update_provenance_and_exit(self, x: int, y: int, p: int):
        pass

    @property
    @overrides(ExtendableTransceiver.bmp_selector)
    def bmp_selector(self) -> Optional[FixedConnectionSelector[BMPConnection]]:
        raise NotImplementedError("Needs to be mocked")

    @property
    @overrides(ExtendableTransceiver.scamp_connection_selector)
    def scamp_connection_selector(self) -> ConnectionSelector:
        raise NotImplementedError("Needs to be mocked")

1	# Copyright (c) 2014 The University of Manchester
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# https://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	# pylint: disable=too-many-arguments
16
17	from typing import (	1✔
18	BinaryIO, Collection, Dict, FrozenSet, Iterable,
19	List, Optional, Tuple, Union)
20	from spinn_utilities.overrides import overrides	1✔
21	from spinn_utilities.progress_bar import ProgressBar	1✔
22	from spinn_utilities.typing.coords import XY	1✔
23	from spinn_machine import (	1✔
24	CoreSubsets, FixedRouteEntry, Machine, MulticastRoutingEntry)
25	from spinn_machine.tags import AbstractTag, IPTag, ReverseIPTag	1✔
26	from spinnman.data import SpiNNManDataView	1✔
27	from spinnman.connections.udp_packet_connections import BMPConnection	1✔
28	from spinnman.connections.udp_packet_connections import (	1✔
29	SCAMPConnection, SDPConnection)
30	from spinnman.processes import ConnectionSelector, FixedConnectionSelector	1✔
31	from spinnman.messages.scp.enums import Signal	1✔
32	from spinnman.messages.sdp import SDPMessage	1✔
33	from spinnman.model import (	1✔
34	CPUInfos, DiagnosticFilter, IOBuffer, RouterDiagnostics,
35	VersionInfo)
36	from spinnman.model.enums import CPUState, UserRegister	1✔
37	from spinnman.transceiver.transceiver import Transceiver	1✔
38	from spinnman.transceiver.extendable_transceiver import ExtendableTransceiver	1✔
39	from spinnman.processes import MostDirectConnectionSelector	1✔
40
41
42	class MockableTransceiver(ExtendableTransceiver):	1✔
43	"""
44	A based for Mock Transceivers
45	"""
46	__slots__ = ["written_memory"]	1✔
47
48	def __init__(self) -> None:	1✔
49	super().__init__()	1✔
50	self.written_memory: List[	1✔
51	Tuple[int, int, int, Union[BinaryIO, bytes, int, str],
52	Optional[int], int, int]] = []
53
54	@overrides(Transceiver.send_sdp_message)	1✔
55	def send_sdp_message(self, message: SDPMessage,	1✔
56	connection: Optional[SDPConnection] = None):
57	pass	×
58
59	@overrides(Transceiver.discover_scamp_connections)	1✔
60	def discover_scamp_connections(self) -> None:	1✔
61	raise NotImplementedError("Needs to be mocked")
62
63	@overrides(Transceiver.add_scamp_connections)	1✔
64	def add_scamp_connections(self, connections: Dict[XY, str]):	1✔
65	pass	×
66
67	@overrides(Transceiver.get_machine_details)	1✔
68	def get_machine_details(self) -> Machine:	1✔
69	return SpiNNManDataView.get_machine()	1✔
70
71	@overrides(Transceiver.get_connections)	1✔
72	def get_connections(self):	1✔
73	raise NotImplementedError("Needs to be mocked")
74
75	@overrides(Transceiver.get_cpu_infos)	1✔
76	def get_cpu_infos(	1✔
77	self, core_subsets: Optional[CoreSubsets] = None,
78	states: Union[CPUState, Iterable[CPUState], None] = None,
79	include: bool = True) -> CPUInfos:
80	raise NotImplementedError("Needs to be mocked")
81
82	@overrides(Transceiver.get_clock_drift)	1✔
83	def get_clock_drift(self, x, y):	1✔
84	raise NotImplementedError("Needs to be mocked")
85
86	@overrides(Transceiver.read_user)	1✔
87	def read_user(self, x: int, y: int, p: int, user: UserRegister):	1✔
88	raise NotImplementedError("Needs to be mocked")
89
90	@overrides(Transceiver.add_cpu_information_from_core)	1✔
91	def add_cpu_information_from_core(	1✔
92	self, cpu_infos: CPUInfos, x: int, y: int, p: int,
93	states: Iterable[CPUState]):
94	raise NotImplementedError("Needs to be mocked")
95
96	@overrides(Transceiver.get_region_base_address)	1✔
97	def get_region_base_address(self, x: int, y: int, p: int):	1✔
98	raise NotImplementedError("Needs to be mocked")
99
100	@overrides(Transceiver.get_iobuf)	1✔
101	def get_iobuf(self, core_subsets: Optional[CoreSubsets] = None	1✔
102	) -> Iterable[IOBuffer]:
103	raise NotImplementedError("Needs to be mocked")
104
105	@overrides(Transceiver.get_core_state_count)	1✔
106	def get_core_state_count(	1✔
107	self, app_id: int, state: CPUState,
108	xys: Optional[Iterable[Tuple[int, int]]] = None) -> int:
109	raise NotImplementedError("Needs to be mocked")
110
111	@overrides(Transceiver.execute_flood)	1✔
112	def execute_flood(	1✔
113	self, core_subsets: CoreSubsets,
114	executable: Union[BinaryIO, bytes, str], app_id: int, *,
115	n_bytes: Optional[int] = None, wait: bool = False):
116	pass	×
117
118	@overrides(Transceiver.read_fpga_register)	1✔
119	def read_fpga_register(	1✔
120	self, fpga_num: int, register: int, board: int = 0) -> int:
121	raise NotImplementedError("Needs to be mocked")
122
123	@overrides(Transceiver.write_fpga_register)	1✔
124	def write_fpga_register(	1✔
125	self, fpga_num: int, register: int, value: int, board: int = 0):
126	pass	×
127
128	@overrides(Transceiver.read_bmp_version)	1✔
129	def read_bmp_version(self, board: int) -> VersionInfo:	1✔
130	raise NotImplementedError("Needs to be mocked")
131
132	@overrides(Transceiver.write_memory)	1✔
133	def write_memory(	1✔
134	self, x: int, y: int, base_address: int,
135	data: Union[BinaryIO, bytes, int, str], *,
136	n_bytes: Optional[int] = None, offset: int = 0, cpu: int = 0,
137	get_sum: bool = False) -> Tuple[int, int]:
138	print("Doing write to", x, y)	1✔
139	self.written_memory.append(	1✔
140	(x, y, base_address, data, n_bytes, offset, cpu))
141	# Hope the return is never used as it will be wrong
142	return (-1, -1)	1✔
143
144	@overrides(Transceiver.write_user)	1✔
145	def write_user(	1✔
146	self, x: int, y: int, p: int, user: UserRegister, value: int):
147	pass	×
148
149	@overrides(Transceiver.read_memory)	1✔
150	def read_memory(	1✔
151	self, x: int, y: int, base_address: int, length: int,
152	cpu: int = 0) -> bytes:
153	raise NotImplementedError("Needs to be mocked")
154
155	@overrides(Transceiver.read_word)	1✔
156	def read_word(	1✔
157	self, x: int, y: int, base_address: int, cpu: int = 0) -> int:
158	raise NotImplementedError("Needs to be mocked")
159
160	@overrides(Transceiver.stop_application)	1✔
161	def stop_application(self, app_id: int):	1✔
162	pass	×
163
164	@overrides(Transceiver.wait_for_cores_to_be_in_state)	1✔
165	def wait_for_cores_to_be_in_state(	1✔
166	self, all_core_subsets: CoreSubsets, app_id: int,
167	cpu_states: Union[CPUState, Iterable[CPUState]], *,
168	timeout: Optional[float] = None,
169	time_between_polls: float = 0.1,
170	error_states: FrozenSet[CPUState] = frozenset((
171	CPUState.RUN_TIME_EXCEPTION, CPUState.WATCHDOG)),
172	counts_between_full_check: int = 100,
173	progress_bar: Optional[ProgressBar] = None):
174	pass	×
175
176	@overrides(Transceiver.send_signal)	1✔
177	def send_signal(self, app_id: int, signal: Signal):	1✔
178	pass	×
179
180	@overrides(Transceiver.set_ip_tag)	1✔
181	def set_ip_tag(self, ip_tag: IPTag, use_sender: bool = False):	1✔
182	pass	×
183
184	@overrides(Transceiver.set_reverse_ip_tag)	1✔
185	def set_reverse_ip_tag(self, reverse_ip_tag: ReverseIPTag):	1✔
186	pass	×
187
188	@overrides(Transceiver.clear_ip_tag)	1✔
189	def clear_ip_tag(self, tag: int, board_address: Optional[str] = None):	1✔
190	pass	×
191
192	@overrides(Transceiver.get_tags)	1✔
193	def get_tags(self, connection: Optional[SCAMPConnection] = None	1✔
194	) -> Iterable[AbstractTag]:
195	raise NotImplementedError("Needs to be mocked")
196
197	@overrides(Transceiver.malloc_sdram)	1✔
198	def malloc_sdram(	1✔
199	self, x: int, y: int, size: int, app_id: int, tag=0) -> int:
200	raise NotImplementedError("Needs to be mocked")
201
202	@overrides(Transceiver.load_multicast_routes)	1✔
203	def load_multicast_routes(	1✔
204	self, x: int, y: int, routes: Collection[MulticastRoutingEntry],
205	app_id: int):
206	pass	×
207
208	@overrides(Transceiver.load_fixed_route)	1✔
209	def load_fixed_route(	1✔
210	self, x: int, y: int, fixed_route: FixedRouteEntry, app_id: int):
211	pass	×
212
213	@overrides(Transceiver.read_fixed_route)	1✔
214	def read_fixed_route(self, x: int, y: int, app_id: int) -> FixedRouteEntry:	1✔
215	raise NotImplementedError("Needs to be mocked")
216
217	@overrides(Transceiver.get_multicast_routes)	1✔
218	def get_multicast_routes(	1✔
219	self, x: int, y: int,
220	app_id: Optional[int] = None) -> List[MulticastRoutingEntry]:
221	raise NotImplementedError("Needs to be mocked")
222
223	@overrides(Transceiver.clear_multicast_routes)	1✔
224	def clear_multicast_routes(self, x: int, y: int):	1✔
225	pass	×
226
227	@overrides(Transceiver.get_router_diagnostics)	1✔
228	def get_router_diagnostics(self, x: int, y: int) -> RouterDiagnostics:	1✔
229	raise NotImplementedError("Needs to be mocked")
230
231	@overrides(Transceiver.get_scamp_connection_selector)	1✔
232	def get_scamp_connection_selector(self) -> MostDirectConnectionSelector:	1✔
233	raise NotImplementedError("Needs to be mocked")
234
235	@overrides(Transceiver.set_router_diagnostic_filter)	1✔
236	def set_router_diagnostic_filter(	1✔
237	self, x: int, y: int, position: int,
238	diagnostic_filter: DiagnosticFilter):
239	pass	×
240
241	@overrides(Transceiver.clear_router_diagnostic_counters)	1✔
242	def clear_router_diagnostic_counters(self, x: int, y: int):	1✔
243	pass	×
244
245	@overrides(Transceiver.close)	1✔
246	def close(self) -> None:	1✔
247	pass	1✔
248
249	@overrides(Transceiver.control_sync)	1✔
250	def control_sync(self, do_sync: bool):	1✔
251	pass	×
252
253	@overrides(Transceiver.update_provenance_and_exit)	1✔
254	def update_provenance_and_exit(self, x: int, y: int, p: int):	1✔
255	pass	×
256
257	@overrides(Transceiver.send_chip_update_provenance_and_exit)	1✔
258	def send_chip_update_provenance_and_exit(self, x: int, y: int, p: int):	1✔
259	pass	×
260
261	@property	1✔
262	@overrides(ExtendableTransceiver.bmp_selector)	1✔
263	def bmp_selector(self) -> Optional[FixedConnectionSelector[BMPConnection]]:	1✔
264	raise NotImplementedError("Needs to be mocked")
265
266	@property	1✔
267	@overrides(ExtendableTransceiver.scamp_connection_selector)	1✔
268	def scamp_connection_selector(self) -> ConnectionSelector:	1✔
269	raise NotImplementedError("Needs to be mocked")

SpiNNakerManchester / SpiNNMan / 6574804013

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous