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

52.78

/spinnman/spalloc/spalloc_eieio_listener.py

# Copyright (c) 2022 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.
"""
API of the client for the Spalloc web service.
"""

import struct
from typing import Optional, Tuple
from spinn_utilities.abstract_base import AbstractBase, abstractmethod
from spinn_utilities.overrides import overrides
from spinn_utilities.typing.coords import XY
from spinnman.connections.udp_packet_connections import EIEIOConnection
from spinnman.constants import SCP_SCAMP_PORT
from spinnman.exceptions import SpinnmanTimeoutException
from spinnman.messages.eieio import (
    AbstractEIEIOMessage,
    read_eieio_command_message, read_eieio_data_message)
from spinnman.messages.sdp import SDPMessage, SDPFlag, SDPHeader
from spinnman.messages.scp.impl import IPTagSet
from .spalloc_proxied_connection import SpallocProxiedConnection
# mypy: disable-error-code=empty-body

_ONE_SHORT = struct.Struct("<H")
_TWO_SHORTS = struct.Struct("<2H")
_TWO_SKIP: bytes = b'\0\0'
_NUM_UPDATE_TAG_TRIES = 3
_UPDATE_TAG_TIMEOUT = 1.0


class SpallocEIEIOListener(
        EIEIOConnection, SpallocProxiedConnection, metaclass=AbstractBase):
    """
    The socket interface supported by proxied EIEIO listener sockets.
    This emulates an :py:class:`EIEOConnection` opened with no address
    specified.
    """
    __slots__ = ()

    @overrides(EIEIOConnection.receive_eieio_message)
    def receive_eieio_message(
            self, timeout: Optional[float] = None) -> AbstractEIEIOMessage:
        data = self.receive(timeout)
        header = _ONE_SHORT.unpack_from(data)[0]
        if header & 0xC000 == 0x4000:
            return read_eieio_command_message(data, 0)
        return read_eieio_data_message(data, 0)

    @overrides(SpallocProxiedConnection.send)
    def send(self, data):
        """
        .. note::
            This class does not allow sending.
        """

    @abstractmethod
    def _get_chip_coords(self, ip_address: str) -> XY:
        """
        Get the coordinates of a chip given its IP address.

        :param str ip_address:
            The IP address of an Ethernet-enabled chip in the job.
        :return: Ethernet-enabled chip coordinates: X, Y
        :rtype: tuple(int, int)
        """
        raise NotImplementedError

    @abstractmethod
    def send_to_chip(
            self, message: bytes, x: int, y: int, port: int = SCP_SCAMP_PORT):
        """
        Send a message on an open socket to a particular board.

        :param bytes message: The message to send.
        :param int x:
            The X coordinate of the Ethernet-enabled chip to send the message
            to.
        :param int y:
            The Y coordinate of the Ethernet-enabled chip to send the message
            to.
        :param int port:
            The UDP port on the Ethernet-enabled chip to send the message to.
            Defaults to the SCP port.
        """
        raise NotImplementedError

    def send_to(self, data: bytes, address: Tuple[str, int]):
        """
        Send a message on an open socket.

        :param bytes message: The message to send.
        :param tuple(str,int) address:
            Where to send it to. Must be the address of an Ethernet-enabled
            chip on a board allocated to the job. Does not mean that SpiNNaker
            is listening on that port (but the SCP port is being listened to if
            the board is booted).
        """
        ip, port = address
        x, y = self._get_chip_coords(ip)
        self.send_to_chip(data, x, y, port)

    @property
    @abstractmethod
    def local_ip_address(self) -> str:  # type: ignore[override]
        """
        The IP address on the server to which the connection is bound.

        :return: The IP address as a dotted string, e.g., 0.0.0.0
        :rtype: str
        """
        raise NotImplementedError

    @property
    @abstractmethod
    def local_port(self) -> int:  # type: ignore[override]
        """
        The port on the server to which the connection is bound.

        :return: The local port number
        :rtype: int
        """
        raise NotImplementedError

    def send_eieio_message_to_core(
            self, eieio_message: AbstractEIEIOMessage, x: int, y: int, p: int,
            ip_address: str):
        """
        Send an EIEIO message (one way) to a given core.

        :param AbstractEIEIOMessage eieio_message:
            The message to send.
        :param int x:
            The X coordinate of the core to send to.
        :param int y:
            The Y coordinate of the core to send to.
        :param int p:
            The ID of the core to send to.
        :param str ip_address:
            The IP address of the Ethernet-enabled chip to route the message
            via.
        """
        sdp_message = SDPMessage(
            SDPHeader(
                flags=SDPFlag.REPLY_NOT_EXPECTED, tag=0,
                destination_port=1, destination_cpu=p,
                destination_chip_x=x, destination_chip_y=y,
                source_port=0, source_cpu=0,
                source_chip_x=0, source_chip_y=0),
            data=eieio_message.bytestring)
        self.send_to(
            _TWO_SKIP + sdp_message.bytestring, (ip_address, SCP_SCAMP_PORT))

    def update_tag(self, x: int, y: int, tag: int, do_receive: bool = True):
        """
        Update the given tag on the given Ethernet-enabled chip to send
        messages to this connection.

        :param int x: The Ethernet-enabled chip's X coordinate
        :param int y: The Ethernet-enabled chip's Y coordinate
        :param int tag: The tag ID to update
        :param bool do_receive: Whether to receive the response or not
        :raises SpinnmanTimeoutException:
            If the message isn't handled within a reasonable timeout.
        :raises SpinnmanUnexpectedResponseCodeException:
            If the message is rejected by SpiNNaker/SCAMP.
        """
        request = IPTagSet(
            x, y, [0, 0, 0, 0], 0, tag, strip=True, use_sender=True)
        request.sdp_header.flags = SDPFlag.REPLY_EXPECTED_NO_P2P
        request.sdp_header.update_for_send(x, y)
        data = _TWO_SKIP + request.bytestring
        for _try in range(_NUM_UPDATE_TAG_TRIES):
            try:
                self.send_to_chip(data, x, y, SCP_SCAMP_PORT)
                if do_receive:
                    response_data = self.receive(_UPDATE_TAG_TIMEOUT)
                    request.get_scp_response().read_bytestring(
                        response_data, len(_TWO_SKIP))
                return
            except SpinnmanTimeoutException as e:
                if _try + 1 == _NUM_UPDATE_TAG_TRIES:
                    raise e

    def update_tag_by_ip(self, ip_address: str, tag: int):
        """
        Update a tag on a board at a given IP address to send messages to this
        connection.

        :param str ip_address: The address of the Ethernet-enabled chip
        :param int tag: The ID of the tag
        """
        x, y = self._get_chip_coords(ip_address)
        self.update_tag(x, y, tag)

1	# Copyright (c) 2022 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	"""	1✔
15	API of the client for the Spalloc web service.
16	"""
17
18	import struct	1✔
19	from typing import Optional, Tuple	1✔
20	from spinn_utilities.abstract_base import AbstractBase, abstractmethod	1✔
21	from spinn_utilities.overrides import overrides	1✔
22	from spinn_utilities.typing.coords import XY	1✔
23	from spinnman.connections.udp_packet_connections import EIEIOConnection	1✔
24	from spinnman.constants import SCP_SCAMP_PORT	1✔
25	from spinnman.exceptions import SpinnmanTimeoutException	1✔
26	from spinnman.messages.eieio import (	1✔
27	AbstractEIEIOMessage,
28	read_eieio_command_message, read_eieio_data_message)
29	from spinnman.messages.sdp import SDPMessage, SDPFlag, SDPHeader	1✔
30	from spinnman.messages.scp.impl import IPTagSet	1✔
31	from .spalloc_proxied_connection import SpallocProxiedConnection	1✔
32	# mypy: disable-error-code=empty-body
33
34	_ONE_SHORT = struct.Struct("<H")	1✔
35	_TWO_SHORTS = struct.Struct("<2H")	1✔
36	_TWO_SKIP: bytes = b'\0\0'	1✔
37	_NUM_UPDATE_TAG_TRIES = 3	1✔
38	_UPDATE_TAG_TIMEOUT = 1.0	1✔
39
40
41	class SpallocEIEIOListener(	1✔
42	EIEIOConnection, SpallocProxiedConnection, metaclass=AbstractBase):
43	"""
44	The socket interface supported by proxied EIEIO listener sockets.
45	This emulates an :py:class:`EIEOConnection` opened with no address
46	specified.
47	"""
48	__slots__ = ()	1✔
49
50	@overrides(EIEIOConnection.receive_eieio_message)	1✔
51	def receive_eieio_message(	1✔
52	self, timeout: Optional[float] = None) -> AbstractEIEIOMessage:
53	data = self.receive(timeout)	×
54	header = _ONE_SHORT.unpack_from(data)[0]	×
55	if header & 0xC000 == 0x4000:	×
56	return read_eieio_command_message(data, 0)	×
57	return read_eieio_data_message(data, 0)	×
58
59	@overrides(SpallocProxiedConnection.send)	1✔
60	def send(self, data):	1✔
61	"""
62	.. note::
63	This class does not allow sending.
64	"""
65
66	@abstractmethod	1✔
67	def _get_chip_coords(self, ip_address: str) -> XY:	1✔
68	"""
69	Get the coordinates of a chip given its IP address.
70
71	:param str ip_address:
72	The IP address of an Ethernet-enabled chip in the job.
73	:return: Ethernet-enabled chip coordinates: X, Y
74	:rtype: tuple(int, int)
75	"""
76	raise NotImplementedError
77
78	@abstractmethod	1✔
79	def send_to_chip(	1✔
80	self, message: bytes, x: int, y: int, port: int = SCP_SCAMP_PORT):
81	"""
82	Send a message on an open socket to a particular board.
83
84	:param bytes message: The message to send.
85	:param int x:
86	The X coordinate of the Ethernet-enabled chip to send the message
87	to.
88	:param int y:
89	The Y coordinate of the Ethernet-enabled chip to send the message
90	to.
91	:param int port:
92	The UDP port on the Ethernet-enabled chip to send the message to.
93	Defaults to the SCP port.
94	"""
95	raise NotImplementedError
96
97	def send_to(self, data: bytes, address: Tuple[str, int]):	1✔
98	"""
99	Send a message on an open socket.
100
101	:param bytes message: The message to send.
102	:param tuple(str,int) address:
103	Where to send it to. Must be the address of an Ethernet-enabled
104	chip on a board allocated to the job. Does not mean that SpiNNaker
105	is listening on that port (but the SCP port is being listened to if
106	the board is booted).
107	"""
108	ip, port = address	×
109	x, y = self._get_chip_coords(ip)	×
110	self.send_to_chip(data, x, y, port)	×
111
112	@property	1✔
113	@abstractmethod	1✔
114	def local_ip_address(self) -> str: # type: ignore[override]	1✔
115	"""
116	The IP address on the server to which the connection is bound.
117
118	:return: The IP address as a dotted string, e.g., 0.0.0.0
119	:rtype: str
120	"""
121	raise NotImplementedError
122
123	@property	1✔
124	@abstractmethod	1✔
125	def local_port(self) -> int: # type: ignore[override]	1✔
126	"""
127	The port on the server to which the connection is bound.
128
129	:return: The local port number
130	:rtype: int
131	"""
132	raise NotImplementedError
133
134	def send_eieio_message_to_core(	1✔
135	self, eieio_message: AbstractEIEIOMessage, x: int, y: int, p: int,
136	ip_address: str):
137	"""
138	Send an EIEIO message (one way) to a given core.
139
140	:param AbstractEIEIOMessage eieio_message:
141	The message to send.
142	:param int x:
143	The X coordinate of the core to send to.
144	:param int y:
145	The Y coordinate of the core to send to.
146	:param int p:
147	The ID of the core to send to.
148	:param str ip_address:
149	The IP address of the Ethernet-enabled chip to route the message
150	via.
151	"""
152	sdp_message = SDPMessage(	×
153	SDPHeader(
154	flags=SDPFlag.REPLY_NOT_EXPECTED, tag=0,
155	destination_port=1, destination_cpu=p,
156	destination_chip_x=x, destination_chip_y=y,
157	source_port=0, source_cpu=0,
158	source_chip_x=0, source_chip_y=0),
159	data=eieio_message.bytestring)
160	self.send_to(	×
161	_TWO_SKIP + sdp_message.bytestring, (ip_address, SCP_SCAMP_PORT))
162
163	def update_tag(self, x: int, y: int, tag: int, do_receive: bool = True):	1✔
164	"""
165	Update the given tag on the given Ethernet-enabled chip to send
166	messages to this connection.
167
168	:param int x: The Ethernet-enabled chip's X coordinate
169	:param int y: The Ethernet-enabled chip's Y coordinate
170	:param int tag: The tag ID to update
171	:param bool do_receive: Whether to receive the response or not
172	:raises SpinnmanTimeoutException:
173	If the message isn't handled within a reasonable timeout.
174	:raises SpinnmanUnexpectedResponseCodeException:
175	If the message is rejected by SpiNNaker/SCAMP.
176	"""
177	request = IPTagSet(	×
178	x, y, [0, 0, 0, 0], 0, tag, strip=True, use_sender=True)
179	request.sdp_header.flags = SDPFlag.REPLY_EXPECTED_NO_P2P	×
180	request.sdp_header.update_for_send(x, y)	×
181	data = _TWO_SKIP + request.bytestring	×
182	for _try in range(_NUM_UPDATE_TAG_TRIES):	×
183	try:	×
184	self.send_to_chip(data, x, y, SCP_SCAMP_PORT)	×
185	if do_receive:	×
186	response_data = self.receive(_UPDATE_TAG_TIMEOUT)	×
187	request.get_scp_response().read_bytestring(	×
188	response_data, len(_TWO_SKIP))
189	return	×
190	except SpinnmanTimeoutException as e:	×
191	if _try + 1 == _NUM_UPDATE_TAG_TRIES:	×
192	raise e	×
193
194	def update_tag_by_ip(self, ip_address: str, tag: int):	1✔
195	"""
196	Update a tag on a board at a given IP address to send messages to this
197	connection.
198
199	:param str ip_address: The address of the Ethernet-enabled chip
200	:param int tag: The ID of the tag
201	"""
202	x, y = self._get_chip_coords(ip_address)	×
203	self.update_tag(x, y, tag)	×

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