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

69.41

/spinnman/data/spinnman_data_view.py

# Copyright (c) 2021 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.
from __future__ import annotations
import logging
from typing import BinaryIO, Optional, Tuple, Union, TYPE_CHECKING
from spinn_utilities.log import FormatAdapter
from spinn_machine.data import MachineDataView
from spinnman.utilities.appid_tracker import AppIdTracker
if TYPE_CHECKING:
    from spinnman.processes import MostDirectConnectionSelector
    from spinnman.transceiver import Transceiver

logger = FormatAdapter(logging.getLogger(__name__))
# pylint: disable=protected-access


class _SpiNNManDataModel(object):
    """
    Singleton data model.

    This class should not be accessed directly please use the DataView and
    DataWriter classes.
    Accessing or editing the data held here directly is *not supported!*

    There may be other DataModel classes which sit next to this one and hold
    additional data. The DataView and DataWriter classes will combine these
    as needed.

    What data is held where and how can change without notice.
    """

    __singleton = None

    __slots__ = [
        # Data values cached
        "_app_id",
        "_app_id_tracker",
        "_scamp_connection_selector",
        "_transceiver",
    ]

    def __new__(cls) -> _SpiNNManDataModel:
        if cls.__singleton:
            return cls.__singleton
        obj = object.__new__(cls)
        cls.__singleton = obj
        obj._transceiver = None
        obj._clear()
        return obj

    def _clear(self) -> None:
        """
        Clears out all data.
        """
        self._hard_reset()

    def _hard_reset(self) -> None:
        """
        Clears out all data that should change after a reset and graph change.
        """
        self._app_id: Optional[int] = None
        self._app_id_tracker: Optional[AppIdTracker] = None
        self._soft_reset()
        self._scamp_connection_selector: Optional[
            MostDirectConnectionSelector] = None
        if self._transceiver:
            try:
                self._transceiver.close()
            except Exception as ex:  # pylint: disable=broad-except
                logger.exception(
                    f"Error {ex} when closing the transceiver ignored")
        self._transceiver: Optional[Transceiver] = None

    def _soft_reset(self) -> None:
        """
        Clears timing and other data that should changed every reset.
        """
        # Holder for any later additions


class SpiNNManDataView(MachineDataView):
    """
    Adds the extra Methods to the View for SpiNNMan level.

    See :py:class:`~spinn_utilities.data.UtilsDataView` for a more detailed
    description.

    This class is designed to only be used directly within the SpiNNMan
    repository as all methods are available to subclasses
    """

    __data = _SpiNNManDataModel()
    __slots__ = ()

    # transceiver methods

    @classmethod
    def has_transceiver(cls) -> bool:
        """
        Reports if a transceiver is currently set.

        :rtype: bool
        """
        return cls.__data._transceiver is not None

    @classmethod
    def get_transceiver(cls) -> Transceiver:
        """
        The transceiver description.

        :rtype: ~spinnman.transceiver.Transceiver
        :raises ~spinn_utilities.exceptions.SpiNNUtilsException:
            If the transceiver is currently unavailable
        """
        if cls.__data._transceiver is None:
            raise cls._exception("transceiver")
        return cls.__data._transceiver

    @classmethod
    def read_memory(
            cls, x: int, y: int, base_address: int, length: int, *,
            cpu: int = 0) -> bytes:
        """
        Read some areas of memory (usually SDRAM) from the board.

        Syntactic sugar for `get_transceiver().read_memory()`.

        :param int x:
            The x-coordinate of the chip where the memory is to be read from
        :param int y:
            The y-coordinate of the chip where the memory is to be read from
        :param int base_address:
            The address in SDRAM where the region of memory to be read starts
        :param int length: The length of the data to be read in bytes
        :param int cpu:
            the core ID used to read the memory of; should usually be 0 when
            reading from SDRAM, but may be other values when reading from DTCM.
        :return: A bytearray of data read
        :rtype: bytes
        :raises ~spinn_utilities.exceptions.SpiNNUtilsException:
            If the transceiver is currently unavailable
        :raise SpinnmanIOException:
            If there is an error communicating with the board
        :raise SpinnmanInvalidPacketException:
            If a packet is received that is not in the valid format
        :raise SpinnmanInvalidParameterException:
            * If one of `x`, `y`, `cpu`, `base_address` or `length` is invalid
            * If a packet is received that has invalid parameters
        :raise SpinnmanUnexpectedResponseCodeException:
            If a response indicates an error during the exchange
        """
        try:
            return cls.get_transceiver().read_memory(
                x, y, base_address, length, cpu=cpu)
        except AttributeError as ex:
            raise cls._exception("transceiver") from ex

    @classmethod
    def write_memory(
            cls, x: int, y: int, base_address: int, data: Union[
                BinaryIO, bytes, bytearray, int, str], *,
            n_bytes: Optional[int] = None, offset: int = 0,
            cpu: int = 0) -> Tuple[int, int]:
        """
        Write to the SDRAM on the board.

        Syntactic sugar for `get_transceiver().write_memory()`.

        :param int x:
            The x-coordinate of the chip where the memory is to be written to
        :param int y:
            The y-coordinate of the chip where the memory is to be written to
        :param int base_address:
            The address in SDRAM where the region of memory is to be written
        :param data: The data to write.  Should be one of the following:

            * An instance of RawIOBase
            * A bytearray/bytes
            * A single integer - will be written in little-endian byte order
            * A filename of a data file
        :type data:
            ~io.RawIOBase or bytes or bytearray or int or str
        :param int n_bytes:
            The amount of data to be written in bytes.  If not specified:

            * If `data` is an RawIOBase, an error is raised
            * If `data` is a byte string (bytearray or bytes), the length of\
              the byte string will be used
            * If `data` is an int, 4 will be used
            * If `data` is a str, the length of the file will be used
        :param int offset: The offset from which the valid data begins
        :param int cpu: The optional CPU to write to
        :raises ~spinn_utilities.exceptions.SpiNNUtilsException:
            If the transceiver is currently unavailable
        :raise SpinnmanIOException:
            * If there is an error communicating with the board
            * If there is an error reading the data
        :raise SpinnmanInvalidPacketException:
            If a packet is received that is not in the valid format
        :raise SpinnmanInvalidParameterException:
            * If `x, y` does not lead to a valid chip
            * If a packet is received that has invalid parameters
            * If `base_address` is not a positive integer
            * If `data` is an RawIOBase but `n_bytes` is not specified
            * If `data` is an int and `n_bytes` is more than 4
            * If `n_bytes` is less than 0
        :raise SpinnmanUnexpectedResponseCodeException:
            If a response indicates an error during the exchange
        """
        return cls.get_transceiver().write_memory(
            x, y, base_address, data,
            n_bytes=n_bytes, offset=offset, cpu=cpu)

    # app_id methods

    @classmethod
    def get_app_id(cls) -> int:
        """
        Gets the main app_id used by the transceiver.

        This method will create a new app_id if one has not yet been created.

        :rtype: int
        """
        if cls.__data._app_id is None:
            cls.__data._app_id = cls.get_new_id()
        return cls.__data._app_id

    @classmethod
    def get_new_id(cls) -> int:
        """
        Gets a new id from the current `app_id_tracker`

        previously `get_transceiver().app_id_tracker().get_new_id()`

        :rtype: AppIdTracker
        """
        if cls.__data._app_id_tracker is None:
            cls.__data._app_id_tracker = AppIdTracker()
        return cls.__data._app_id_tracker.get_new_id()

    @classmethod
    def free_id(cls, app_id: int):
        """
        Frees up an app_id.

        previously `get_transceiver().app_id_tracker().free_id(app_id)`

        :param int app_id:
        """
        if cls.__data._app_id_tracker:
            cls.__data._app_id_tracker.free_id(app_id)

    @classmethod
    def get_scamp_connection_selector(cls) -> MostDirectConnectionSelector:
        """
        Gets the SCAMP connection selector from the transceiver.

        Syntactic sugar for `get_transceiver().get_scamp_connection_selector()`

        :rtype: MostDirectConnectionSelector
        :raises ~spinn_utilities.exceptions.SpiNNUtilsException:
            If the transceiver is currently unavailable
        """
        if not cls.__data._scamp_connection_selector:
            cls.__data._scamp_connection_selector =\
                cls. get_transceiver().get_scamp_connection_selector()
        return cls.__data._scamp_connection_selector

1	# Copyright (c) 2021 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	from __future__ import annotations	1✔
15	import logging	1✔
16	from typing import BinaryIO, Optional, Tuple, Union, TYPE_CHECKING	1✔
17	from spinn_utilities.log import FormatAdapter	1✔
18	from spinn_machine.data import MachineDataView	1✔
19	from spinnman.utilities.appid_tracker import AppIdTracker	1✔
20	if TYPE_CHECKING:
21	from spinnman.processes import MostDirectConnectionSelector
22	from spinnman.transceiver import Transceiver
23
24	logger = FormatAdapter(logging.getLogger(__name__))	1✔
25	# pylint: disable=protected-access
26
27
28	class _SpiNNManDataModel(object):	1✔
29	"""
30	Singleton data model.
31
32	This class should not be accessed directly please use the DataView and
33	DataWriter classes.
34	Accessing or editing the data held here directly is not supported!
35
36	There may be other DataModel classes which sit next to this one and hold
37	additional data. The DataView and DataWriter classes will combine these
38	as needed.
39
40	What data is held where and how can change without notice.
41	"""
42
43	__singleton = None	1✔
44
45	__slots__ = [	1✔
46	# Data values cached
47	"_app_id",
48	"_app_id_tracker",
49	"_scamp_connection_selector",
50	"_transceiver",
51	]
52
53	def __new__(cls) -> _SpiNNManDataModel:	1✔
54	if cls.__singleton:	1✔
55	return cls.__singleton	1✔
56	obj = object.__new__(cls)	1✔
57	cls.__singleton = obj	1✔
58	obj._transceiver = None	1✔
59	obj._clear()	1✔
60	return obj	1✔
61
62	def _clear(self) -> None:	1✔
63	"""
64	Clears out all data.
65	"""
66	self._hard_reset()	1✔
67
68	def _hard_reset(self) -> None:	1✔
69	"""
70	Clears out all data that should change after a reset and graph change.
71	"""
72	self._app_id: Optional[int] = None	1✔
73	self._app_id_tracker: Optional[AppIdTracker] = None	1✔
74	self._soft_reset()	1✔
75	self._scamp_connection_selector: Optional[	1✔
76	MostDirectConnectionSelector] = None
77	if self._transceiver:	1✔
78	try:	1✔
79	self._transceiver.close()	1✔
80	except Exception as ex: # pylint: disable=broad-except	×
81	logger.exception(	×
82	f"Error {ex} when closing the transceiver ignored")
83	self._transceiver: Optional[Transceiver] = None	1✔
84
85	def _soft_reset(self) -> None:	1✔
86	"""
87	Clears timing and other data that should changed every reset.
88	"""
89	# Holder for any later additions
90
91
92	class SpiNNManDataView(MachineDataView):	1✔
93	"""
94	Adds the extra Methods to the View for SpiNNMan level.
95
96	See :py:class:`~spinn_utilities.data.UtilsDataView` for a more detailed
97	description.
98
99	This class is designed to only be used directly within the SpiNNMan
100	repository as all methods are available to subclasses
101	"""
102
103	__data = _SpiNNManDataModel()	1✔
104	__slots__ = ()	1✔
105
106	# transceiver methods
107
108	@classmethod	1✔
109	def has_transceiver(cls) -> bool:	1✔
110	"""
111	Reports if a transceiver is currently set.
112
113	:rtype: bool
114	"""
115	return cls.__data._transceiver is not None	1✔
116
117	@classmethod	1✔
118	def get_transceiver(cls) -> Transceiver:	1✔
119	"""
120	The transceiver description.
121
122	:rtype: ~spinnman.transceiver.Transceiver
123	:raises ~spinn_utilities.exceptions.SpiNNUtilsException:
124	If the transceiver is currently unavailable
125	"""
126	if cls.__data._transceiver is None:	1✔
127	raise cls._exception("transceiver")	1✔
128	return cls.__data._transceiver	1✔
129
130	@classmethod	1✔
131	def read_memory(	1✔
132	cls, x: int, y: int, base_address: int, length: int, *,
133	cpu: int = 0) -> bytes:
134	"""
135	Read some areas of memory (usually SDRAM) from the board.
136
137	Syntactic sugar for `get_transceiver().read_memory()`.
138
139	:param int x:
140	The x-coordinate of the chip where the memory is to be read from
141	:param int y:
142	The y-coordinate of the chip where the memory is to be read from
143	:param int base_address:
144	The address in SDRAM where the region of memory to be read starts
145	:param int length: The length of the data to be read in bytes
146	:param int cpu:
147	the core ID used to read the memory of; should usually be 0 when
148	reading from SDRAM, but may be other values when reading from DTCM.
149	:return: A bytearray of data read
150	:rtype: bytes
151	:raises ~spinn_utilities.exceptions.SpiNNUtilsException:
152	If the transceiver is currently unavailable
153	:raise SpinnmanIOException:
154	If there is an error communicating with the board
155	:raise SpinnmanInvalidPacketException:
156	If a packet is received that is not in the valid format
157	:raise SpinnmanInvalidParameterException:
158	* If one of `x`, `y`, `cpu`, `base_address` or `length` is invalid
159	* If a packet is received that has invalid parameters
160	:raise SpinnmanUnexpectedResponseCodeException:
161	If a response indicates an error during the exchange
162	"""
163	try:	×
164	return cls.get_transceiver().read_memory(	×
165	x, y, base_address, length, cpu=cpu)
166	except AttributeError as ex:	×
167	raise cls._exception("transceiver") from ex	×
168
169	@classmethod	1✔
170	def write_memory(	1✔
171	cls, x: int, y: int, base_address: int, data: Union[
172	BinaryIO, bytes, bytearray, int, str], *,
173	n_bytes: Optional[int] = None, offset: int = 0,
174	cpu: int = 0) -> Tuple[int, int]:
175	"""
176	Write to the SDRAM on the board.
177
178	Syntactic sugar for `get_transceiver().write_memory()`.
179
180	:param int x:
181	The x-coordinate of the chip where the memory is to be written to
182	:param int y:
183	The y-coordinate of the chip where the memory is to be written to
184	:param int base_address:
185	The address in SDRAM where the region of memory is to be written
186	:param data: The data to write. Should be one of the following:
187
188	* An instance of RawIOBase
189	* A bytearray/bytes
190	* A single integer - will be written in little-endian byte order
191	* A filename of a data file
192	:type data:
193	~io.RawIOBase or bytes or bytearray or int or str
194	:param int n_bytes:
195	The amount of data to be written in bytes. If not specified:
196
197	* If `data` is an RawIOBase, an error is raised
198	* If `data` is a byte string (bytearray or bytes), the length of\
199	the byte string will be used
200	* If `data` is an int, 4 will be used
201	* If `data` is a str, the length of the file will be used
202	:param int offset: The offset from which the valid data begins
203	:param int cpu: The optional CPU to write to
204	:raises ~spinn_utilities.exceptions.SpiNNUtilsException:
205	If the transceiver is currently unavailable
206	:raise SpinnmanIOException:
207	* If there is an error communicating with the board
208	* If there is an error reading the data
209	:raise SpinnmanInvalidPacketException:
210	If a packet is received that is not in the valid format
211	:raise SpinnmanInvalidParameterException:
212	* If `x, y` does not lead to a valid chip
213	* If a packet is received that has invalid parameters
214	* If `base_address` is not a positive integer
215	* If `data` is an RawIOBase but `n_bytes` is not specified
216	* If `data` is an int and `n_bytes` is more than 4
217	* If `n_bytes` is less than 0
218	:raise SpinnmanUnexpectedResponseCodeException:
219	If a response indicates an error during the exchange
220	"""
221	return cls.get_transceiver().write_memory(	×
222	x, y, base_address, data,
223	n_bytes=n_bytes, offset=offset, cpu=cpu)
224
225	# app_id methods
226
227	@classmethod	1✔
228	def get_app_id(cls) -> int:	1✔
229	"""
230	Gets the main app_id used by the transceiver.
231
232	This method will create a new app_id if one has not yet been created.
233
234	:rtype: int
235	"""
236	if cls.__data._app_id is None:	×
237	cls.__data._app_id = cls.get_new_id()	×
238	return cls.__data._app_id	×
239
240	@classmethod	1✔
241	def get_new_id(cls) -> int:	1✔
242	"""
243	Gets a new id from the current `app_id_tracker`
244
245	previously `get_transceiver().app_id_tracker().get_new_id()`
246
247	:rtype: AppIdTracker
248	"""
249	if cls.__data._app_id_tracker is None:	×
250	cls.__data._app_id_tracker = AppIdTracker()	×
251	return cls.__data._app_id_tracker.get_new_id()	×
252
253	@classmethod	1✔
254	def free_id(cls, app_id: int):	1✔
255	"""
256	Frees up an app_id.
257
258	previously `get_transceiver().app_id_tracker().free_id(app_id)`
259
260	:param int app_id:
261	"""
262	if cls.__data._app_id_tracker:	×
263	cls.__data._app_id_tracker.free_id(app_id)	×
264
265	@classmethod	1✔
266	def get_scamp_connection_selector(cls) -> MostDirectConnectionSelector:	1✔
267	"""
268	Gets the SCAMP connection selector from the transceiver.
269
270	Syntactic sugar for `get_transceiver().get_scamp_connection_selector()`
271
272	:rtype: MostDirectConnectionSelector
273	:raises ~spinn_utilities.exceptions.SpiNNUtilsException:
274	If the transceiver is currently unavailable
275	"""
276	if not cls.__data._scamp_connection_selector:	×
277	cls.__data._scamp_connection_selector =\	×
278	cls. get_transceiver().get_scamp_connection_selector()
279	return cls.__data._scamp_connection_selector	×

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