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

56.04

/spinnman/messages/scp/impl/iptag_get.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.

from spinn_utilities.overrides import overrides
from spinnman.messages.scp import SCPRequestHeader
from spinnman.messages.scp.abstract_messages import (
    AbstractSCPRequest, AbstractSCPResponse)
from spinnman.messages.scp.enums import SCPCommand, SCPResult
from spinnman.messages.sdp import SDPFlag, SDPHeader
from spinnman.exceptions import SpinnmanUnexpectedResponseCodeException

import struct


_IPTAG_GET = 2

_IPTAG_FORMAT = struct.Struct("<4s6s3HIH3B")


class IPTagGetResponse(AbstractSCPResponse):
    """
    An SCP response to a request for an IP tags.
    """
    __slots__ = [
        "_count",
        "_flags",
        "_ip_address",
        "_mac_address",
        "_port",
        "_rx_port",
        "_spin_chip_x",
        "_spin_chip_y",
        "_spin_cpu",
        "_spin_port",
        "_timeout"]

    def __init__(self) -> None:
        super().__init__()
        self._ip_address = b'\0\0\0\0'
        self._mac_address = b'\0\0\0\0\0\0'
        self._port = 0
        self._timeout = 0
        self._flags = 0
        self._count = 0
        self._rx_port = 0
        self._spin_chip_y = 0
        self._spin_chip_x = 0
        self._spin_cpu = 0
        self._spin_port = 0

    @overrides(AbstractSCPResponse.read_data_bytestring)
    def read_data_bytestring(self, data: bytes, offset: int):
        result = self.scp_response_header.result
        if result != SCPResult.RC_OK:
            raise SpinnmanUnexpectedResponseCodeException(
                "Get IP Tag Info", "CMD_IPTAG", result.name)
        (ip_address, mac_address, self._port, self._timeout,
         self._flags, self._count, self._rx_port, self._spin_chip_y,
         self._spin_chip_x, processor_and_port) = _IPTAG_FORMAT.unpack_from(
            data, offset)
        self._ip_address = bytearray(ip_address)
        self._mac_address = bytearray(mac_address)
        self._spin_port = (processor_and_port >> 5) & 0x7
        self._spin_cpu = processor_and_port & 0x1F

    @property
    def ip_address(self) -> bytes:
        """
        The IP address of the tag, as a bytearray of 4 bytes.

        :rtype: bytearray
        """
        return self._ip_address

    @property
    def mac_address(self) -> bytes:
        """
        The MAC address of the tag, as a bytearray of 6 bytes.

        :rtype: bytearray
        """
        return self._mac_address

    @property
    def port(self) -> int:
        """
        The port of the tag.

        :rtype: int
        """
        return self._port

    @property
    def timeout(self) -> int:
        """
        The timeout of the tag.

        :rtype: int
        """
        return self._timeout

    @property
    def flags(self) -> int:
        """
        The flags of the tag.

        :rtype: int
        """
        return self._flags

    @property
    def in_use(self) -> bool:
        """
        Whether the tag is marked as being in use.

        :rtype: bool
        """
        return (self._flags & 0x8000) > 0

    @property
    def is_temporary(self) -> bool:
        """
        Whether the tag is temporary.

        :rtype: bool
        """
        return (self._flags & 0x4000) > 0

    @property
    def is_arp(self) -> bool:
        """
        Whether the tag is in the Address Resolution Protocol state
        (where the MAC address is being looked up).

        .. note::
            This is a transient state; it is unlikely to be observed.

        :rtype: bool
        """
        return (self._flags & 0x2000) > 0

    @property
    def is_reverse(self) -> bool:
        """
        Whether the tag is a reverse tag.

        :rtype: bool
        """
        return (self._flags & 0x0200) > 0

    @property
    def strip_sdp(self) -> bool:
        """
        Whether the tag is to strip the SDP header.

        :rtype: bool
        """
        return (self._flags & 0x0100) > 0

    @property
    def count(self) -> int:
        """
        The count of the number of packets that have been sent with the tag.

        :rtype: int
        """
        return self._count

    @property
    def rx_port(self) -> int:
        """
        The receive port of the tag.

        :rtype: int
        """
        return self._rx_port

    @property
    def spin_chip_x(self) -> int:
        """
        The X-coordinate of the chip on which the tag is defined.

        :rtype: int
        """
        return self._spin_chip_x

    @property
    def spin_chip_y(self) -> int:
        """
        The Y-coordinate of the chip on which the tag is defined.

        :rtype: int
        """
        return self._spin_chip_y

    @property
    def spin_port(self) -> int:
        """
        The spin-port of the IP tag.

        :rtype: int
        """
        return self._spin_port

    @property
    def spin_cpu(self) -> int:
        """
        The CPU ID of the IP tag.

        :rtype: int
        """
        return self._spin_cpu


class IPTagGet(AbstractSCPRequest[IPTagGetResponse]):
    """
    An SCP Request to get an IP tag.
    """
    __slots__ = ()

    def __init__(self, x: int, y: int, tag: int):
        """
        :param int x: The x-coordinate of a chip, between 0 and 255
        :param int y: The y-coordinate of a chip, between 0 and 255
        :param int tag: The tag to get details of, between 0 and 7
        """
        super().__init__(
            SDPHeader(
                flags=SDPFlag.REPLY_EXPECTED, destination_port=0,
                destination_cpu=0, destination_chip_x=x,
                destination_chip_y=y),
            SCPRequestHeader(command=SCPCommand.CMD_IPTAG),
            argument_1=(_IPTAG_GET << 16) | tag,
            argument_2=1)

    @overrides(AbstractSCPRequest.get_scp_response)
    def get_scp_response(self) -> IPTagGetResponse:
        return IPTagGetResponse()

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	from spinn_utilities.overrides import overrides	1✔
16	from spinnman.messages.scp import SCPRequestHeader	1✔
17	from spinnman.messages.scp.abstract_messages import (	1✔
18	AbstractSCPRequest, AbstractSCPResponse)
19	from spinnman.messages.scp.enums import SCPCommand, SCPResult	1✔
20	from spinnman.messages.sdp import SDPFlag, SDPHeader	1✔
21	from spinnman.exceptions import SpinnmanUnexpectedResponseCodeException	1✔
22
23	import struct	1✔
24
25
26	_IPTAG_GET = 2	1✔
27
28	_IPTAG_FORMAT = struct.Struct("<4s6s3HIH3B")	1✔
29
30
31	class IPTagGetResponse(AbstractSCPResponse):	1✔
32	"""
33	An SCP response to a request for an IP tags.
34	"""
35	__slots__ = [	1✔
36	"_count",
37	"_flags",
38	"_ip_address",
39	"_mac_address",
40	"_port",
41	"_rx_port",
42	"_spin_chip_x",
43	"_spin_chip_y",
44	"_spin_cpu",
45	"_spin_port",
46	"_timeout"]
47
48	def __init__(self) -> None:	1✔
49	super().__init__()	×
50	self._ip_address = b'\0\0\0\0'	×
51	self._mac_address = b'\0\0\0\0\0\0'	×
52	self._port = 0	×
53	self._timeout = 0	×
54	self._flags = 0	×
55	self._count = 0	×
56	self._rx_port = 0	×
57	self._spin_chip_y = 0	×
58	self._spin_chip_x = 0	×
59	self._spin_cpu = 0	×
60	self._spin_port = 0	×
61
62	@overrides(AbstractSCPResponse.read_data_bytestring)	1✔
63	def read_data_bytestring(self, data: bytes, offset: int):	1✔
64	result = self.scp_response_header.result	×
65	if result != SCPResult.RC_OK:	×
66	raise SpinnmanUnexpectedResponseCodeException(	×
67	"Get IP Tag Info", "CMD_IPTAG", result.name)
68	(ip_address, mac_address, self._port, self._timeout,	×
69	self._flags, self._count, self._rx_port, self._spin_chip_y,
70	self._spin_chip_x, processor_and_port) = _IPTAG_FORMAT.unpack_from(
71	data, offset)
72	self._ip_address = bytearray(ip_address)	×
73	self._mac_address = bytearray(mac_address)	×
74	self._spin_port = (processor_and_port >> 5) & 0x7	×
75	self._spin_cpu = processor_and_port & 0x1F	×
76
77	@property	1✔
78	def ip_address(self) -> bytes:	1✔
79	"""
80	The IP address of the tag, as a bytearray of 4 bytes.
81
82	:rtype: bytearray
83	"""
84	return self._ip_address	×
85
86	@property	1✔
87	def mac_address(self) -> bytes:	1✔
88	"""
89	The MAC address of the tag, as a bytearray of 6 bytes.
90
91	:rtype: bytearray
92	"""
93	return self._mac_address	×
94
95	@property	1✔
96	def port(self) -> int:	1✔
97	"""
98	The port of the tag.
99
100	:rtype: int
101	"""
102	return self._port	×
103
104	@property	1✔
105	def timeout(self) -> int:	1✔
106	"""
107	The timeout of the tag.
108
109	:rtype: int
110	"""
111	return self._timeout	×
112
113	@property	1✔
114	def flags(self) -> int:	1✔
115	"""
116	The flags of the tag.
117
118	:rtype: int
119	"""
120	return self._flags	×
121
122	@property	1✔
123	def in_use(self) -> bool:	1✔
124	"""
125	Whether the tag is marked as being in use.
126
127	:rtype: bool
128	"""
129	return (self._flags & 0x8000) > 0	×
130
131	@property	1✔
132	def is_temporary(self) -> bool:	1✔
133	"""
134	Whether the tag is temporary.
135
136	:rtype: bool
137	"""
138	return (self._flags & 0x4000) > 0	×
139
140	@property	1✔
141	def is_arp(self) -> bool:	1✔
142	"""
143	Whether the tag is in the Address Resolution Protocol state
144	(where the MAC address is being looked up).
145
146	.. note::
147	This is a transient state; it is unlikely to be observed.
148
149	:rtype: bool
150	"""
151	return (self._flags & 0x2000) > 0	×
152
153	@property	1✔
154	def is_reverse(self) -> bool:	1✔
155	"""
156	Whether the tag is a reverse tag.
157
158	:rtype: bool
159	"""
160	return (self._flags & 0x0200) > 0	×
161
162	@property	1✔
163	def strip_sdp(self) -> bool:	1✔
164	"""
165	Whether the tag is to strip the SDP header.
166
167	:rtype: bool
168	"""
169	return (self._flags & 0x0100) > 0	×
170
171	@property	1✔
172	def count(self) -> int:	1✔
173	"""
174	The count of the number of packets that have been sent with the tag.
175
176	:rtype: int
177	"""
178	return self._count	×
179
180	@property	1✔
181	def rx_port(self) -> int:	1✔
182	"""
183	The receive port of the tag.
184
185	:rtype: int
186	"""
187	return self._rx_port	×
188
189	@property	1✔
190	def spin_chip_x(self) -> int:	1✔
191	"""
192	The X-coordinate of the chip on which the tag is defined.
193
194	:rtype: int
195	"""
196	return self._spin_chip_x	×
197
198	@property	1✔
199	def spin_chip_y(self) -> int:	1✔
200	"""
201	The Y-coordinate of the chip on which the tag is defined.
202
203	:rtype: int
204	"""
205	return self._spin_chip_y	×
206
207	@property	1✔
208	def spin_port(self) -> int:	1✔
209	"""
210	The spin-port of the IP tag.
211
212	:rtype: int
213	"""
214	return self._spin_port	×
215
216	@property	1✔
217	def spin_cpu(self) -> int:	1✔
218	"""
219	The CPU ID of the IP tag.
220
221	:rtype: int
222	"""
223	return self._spin_cpu	×
224
225
226	class IPTagGet(AbstractSCPRequest[IPTagGetResponse]):	1✔
227	"""
228	An SCP Request to get an IP tag.
229	"""
230	__slots__ = ()	1✔
231
232	def __init__(self, x: int, y: int, tag: int):	1✔
233	"""
234	:param int x: The x-coordinate of a chip, between 0 and 255
235	:param int y: The y-coordinate of a chip, between 0 and 255
236	:param int tag: The tag to get details of, between 0 and 7
237	"""
238	super().__init__(	×
239	SDPHeader(
240	flags=SDPFlag.REPLY_EXPECTED, destination_port=0,
241	destination_cpu=0, destination_chip_x=x,
242	destination_chip_y=y),
243	SCPRequestHeader(command=SCPCommand.CMD_IPTAG),
244	argument_1=(_IPTAG_GET << 16) \| tag,
245	argument_2=1)
246
247	@overrides(AbstractSCPRequest.get_scp_response)	1✔
248	def get_scp_response(self) -> IPTagGetResponse:	1✔
249	return IPTagGetResponse()	×

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