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

53.73

/spinnman/connections/udp_packet_connections/udp_connection.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.

import logging
import socket
import select
from contextlib import suppress
from typing import Callable, Optional, Tuple
from spinn_utilities.log import FormatAdapter
from spinn_utilities.overrides import overrides
from spinn_utilities.ping import Ping
from spinnman.exceptions import (SpinnmanIOException, SpinnmanEOFException)
from spinnman.connections.abstract_classes import Connection
from spinnman.utilities.socket_utils import (
    bind_socket, connect_socket, get_udp_socket, get_socket_address,
    resolve_host, set_receive_buffer_size, receive_message,
    receive_message_and_address, send_message, send_message_to_address)
from spinnman.connections.abstract_classes import Listenable

logger = FormatAdapter(logging.getLogger(__name__))
_RECEIVE_BUFFER_SIZE = 1048576
_PING_COUNT = 5
_REPR_TEMPLATE = "UDPConnection(local={}:{}, remote={}:{})"
_MSG_MAX = 300


class UDPConnection(Connection, Listenable[bytes]):
    """
    A connection that routes messages via UDP to some remote host.

    Subclasses of this should be aware that UDP messages may be dropped,
    reordered, or duplicated, and that there's no way to tell whether the
    other end of the connection truly exists except by listening for
    occasional messages from them. There is also an upper size limit on
    messages, formally of 64kB, but usually of about 1500 bytes (the typical
    maximum size of an Ethernet packet);
    SDP messages have lower maximum lengths.
    """

    __slots__ = (
        "_can_send",
        "_local_ip_address",
        "_local_port",
        "_remote_ip_address",
        "_remote_port",
        "_socket")

    def __init__(
            self, local_host: Optional[str] = None,
            local_port: Optional[int] = None,
            remote_host: Optional[str] = None,
            remote_port: Optional[int] = None):
        """
        :param str local_host: The local host name or IP address to bind to.
            If not specified defaults to bind to all interfaces, unless
            remote_host is specified, in which case binding is done to the
            IP address that will be used to send packets
        :param int local_port: The local port to bind to, between 1025 and
            65535. If not specified, defaults to a random unused local port
        :param str remote_host: The remote host name or IP address to send
            packets to. If not specified, the socket will be available for
            listening only, and will throw and exception if used for sending
        :param int remote_port: The remote port to send packets to.  If
            remote_host is None, this is ignored.  If remote_host is specified
            specified, this must also be specified for the connection to allow
            sending
        :raise SpinnmanIOException:
            If there is an error setting up the communication channel
        """
        self._socket = get_udp_socket()
        set_receive_buffer_size(self._socket, _RECEIVE_BUFFER_SIZE)

        # Get the host and port to bind to locally
        local_bind_host = "" if local_host is None else local_host
        local_bind_port = 0 if local_port is None else local_port
        bind_socket(self._socket, local_bind_host, local_bind_port)

        # Mark the socket as non-sending, unless the remote host is
        # specified - send requests will then cause an exception
        self._can_send = False
        self._remote_ip_address: Optional[str] = None
        self._remote_port: Optional[int] = None

        # Get the host to connect to remotely
        if remote_host is not None and remote_port is not None:
            self._remote_port = remote_port
            self._remote_ip_address = resolve_host(remote_host)
            connect_socket(self._socket, self._remote_ip_address, remote_port)
            self._can_send = True

        # If things are closed here, it's a catastrophic problem
        if self.__is_closed:
            raise SpinnmanEOFException()

        # Get the details of where the socket is connected
        self._local_ip_address, self._local_port = \
            get_socket_address(self._socket)

        # Set a general timeout on the socket
        self._socket.settimeout(1.0)

    @property
    def __is_closed(self) -> bool:
        """
        Is the socket closed?

        .. note::
            Just because a socket is not closed doesn't mean that you're going
            to be able to successfully write to it or read from it; some
            failures are only detected on use. But closed sockets definitely
            behave in certain ways!

        :rtype: bool
        """
        # Reach into Python'#s guts
        # pylint: disable=protected-access
        return self._socket._closed  # type: ignore[attr-defined]

    @overrides(Connection.is_connected)
    def is_connected(self) -> bool:
        # Closed sockets are never connected!
        if self.__is_closed:
            return False

        # If this is not a sending socket, it is not connected
        if not self._can_send:
            return False

        # check if machine is active and on the network
        for _ in range(_PING_COUNT):
            # Assume connected if ping works
            if Ping.ping(self._remote_ip_address) == 0:
                return True

        # If the ping fails this number of times, the host cannot be contacted
        return False

    @property
    def local_ip_address(self) -> str:
        """
        The local IP address to which the connection is bound,
        as a dotted string, e.g., `0.0.0.0`.

        :rtype: str
        """
        return self._local_ip_address

    @property
    def local_port(self) -> int:
        """
        The number of the local port to which the connection is bound.

        :rtype: int
        """
        return self._local_port

    @property
    def remote_ip_address(self) -> Optional[str]:
        """
        The remote IP address to which the connection is connected,
        or `None` if not connected remotely.

        :rtype: str
        """
        return self._remote_ip_address

    @property
    def remote_port(self) -> Optional[int]:
        """
        The remote port number to which the connection is connected,
        or `None` if not connected remotely.

        :rtype: int
        """
        return self._remote_port

    def receive(self, timeout: Optional[float] = None) -> bytes:
        """
        Receive data from the connection.

        :param float timeout: The timeout in seconds, or `None` to wait forever
        :return: The data received as a byte-string
        :rtype: bytes
        :raise SpinnmanTimeoutException:
            If a timeout occurs before any data is received
        :raise SpinnmanIOException: If an error occurs receiving the data
        """
        if self.__is_closed:
            raise SpinnmanEOFException()
        return receive_message(self._socket, timeout, _MSG_MAX)

    def receive_with_address(self, timeout=None):
        """
        Receive data from the connection along with the address where the
        data was received from.

        :param float timeout: The timeout, or `None` to wait forever
        :return: A tuple of the data received and a tuple of the
            (address, port) received from
        :rtype: tuple(bytes, tuple(str, int))
        :raise SpinnmanTimeoutException:
            If a timeout occurs before any data is received
        :raise SpinnmanIOException: If an error occurs receiving the data
        """
        if self.__is_closed:
            raise SpinnmanEOFException()
        return receive_message_and_address(self._socket, timeout, _MSG_MAX)

    def send(self, data: bytes):
        """
        Send data down this connection.

        :param data: The data to be sent
        :type data: bytes or bytearray
        :raise SpinnmanIOException: If there is an error sending the data
        """
        if self.__is_closed:
            raise SpinnmanEOFException()
        if not self._can_send:
            raise SpinnmanIOException(
                "Remote host and/or port not set - data cannot be sent with"
                " this connection")
        while not send_message(self._socket, data):
            if self.__is_closed:
                raise SpinnmanEOFException()

    def send_to(self, data: bytes, address: Tuple[str, int]):
        """
        Send data down this connection.

        :param data: The data to be sent as a byte-string
        :type data: bytes or bytearray
        :param tuple(str,int) address:
            A tuple of (address, port) to send the data to
        :raise SpinnmanIOException: If there is an error sending the data
        """
        if self.__is_closed:
            raise SpinnmanEOFException()
        while not send_message_to_address(self._socket, data, address):
            if self.__is_closed:
                raise SpinnmanEOFException()

    @overrides(Connection.close)
    def close(self) -> None:
        if self.__is_closed:
            return
        with suppress(Exception):
            self._socket.shutdown(socket.SHUT_WR)
        self._socket.close()

    def is_ready_to_receive(self, timeout: float = 0):
        if self.__is_closed:
            return True
        return len(select.select([self._socket], [], [], timeout)[0]) == 1

    def __repr__(self) -> str:
        return _REPR_TEMPLATE.format(
            self.local_ip_address, self.local_port,
            self.remote_ip_address, self.remote_port)

    @overrides(Listenable.get_receive_method)
    def get_receive_method(self) -> Callable[[], bytes]:
        return self.receive

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	import logging	1✔
16	import socket	1✔
17	import select	1✔
18	from contextlib import suppress	1✔
19	from typing import Callable, Optional, Tuple	1✔
20	from spinn_utilities.log import FormatAdapter	1✔
21	from spinn_utilities.overrides import overrides	1✔
22	from spinn_utilities.ping import Ping	1✔
23	from spinnman.exceptions import (SpinnmanIOException, SpinnmanEOFException)	1✔
24	from spinnman.connections.abstract_classes import Connection	1✔
25	from spinnman.utilities.socket_utils import (	1✔
26	bind_socket, connect_socket, get_udp_socket, get_socket_address,
27	resolve_host, set_receive_buffer_size, receive_message,
28	receive_message_and_address, send_message, send_message_to_address)
29	from spinnman.connections.abstract_classes import Listenable	1✔
30
31	logger = FormatAdapter(logging.getLogger(__name__))	1✔
32	_RECEIVE_BUFFER_SIZE = 1048576	1✔
33	_PING_COUNT = 5	1✔
34	_REPR_TEMPLATE = "UDPConnection(local={}:{}, remote={}:{})"	1✔
35	_MSG_MAX = 300	1✔
36
37
38	class UDPConnection(Connection, Listenable[bytes]):	1✔
39	"""
40	A connection that routes messages via UDP to some remote host.
41
42	Subclasses of this should be aware that UDP messages may be dropped,
43	reordered, or duplicated, and that there's no way to tell whether the
44	other end of the connection truly exists except by listening for
45	occasional messages from them. There is also an upper size limit on
46	messages, formally of 64kB, but usually of about 1500 bytes (the typical
47	maximum size of an Ethernet packet);
48	SDP messages have lower maximum lengths.
49	"""
50
51	__slots__ = (	1✔
52	"_can_send",
53	"_local_ip_address",
54	"_local_port",
55	"_remote_ip_address",
56	"_remote_port",
57	"_socket")
58
59	def __init__(	1✔
60	self, local_host: Optional[str] = None,
61	local_port: Optional[int] = None,
62	remote_host: Optional[str] = None,
63	remote_port: Optional[int] = None):
64	"""
65	:param str local_host: The local host name or IP address to bind to.
66	If not specified defaults to bind to all interfaces, unless
67	remote_host is specified, in which case binding is done to the
68	IP address that will be used to send packets
69	:param int local_port: The local port to bind to, between 1025 and
70	65535. If not specified, defaults to a random unused local port
71	:param str remote_host: The remote host name or IP address to send
72	packets to. If not specified, the socket will be available for
73	listening only, and will throw and exception if used for sending
74	:param int remote_port: The remote port to send packets to. If
75	remote_host is None, this is ignored. If remote_host is specified
76	specified, this must also be specified for the connection to allow
77	sending
78	:raise SpinnmanIOException:
79	If there is an error setting up the communication channel
80	"""
81	self._socket = get_udp_socket()	1✔
82	set_receive_buffer_size(self._socket, _RECEIVE_BUFFER_SIZE)	1✔
83
84	# Get the host and port to bind to locally
85	local_bind_host = "" if local_host is None else local_host	1✔
86	local_bind_port = 0 if local_port is None else local_port	1✔
87	bind_socket(self._socket, local_bind_host, local_bind_port)	1✔
88
89	# Mark the socket as non-sending, unless the remote host is
90	# specified - send requests will then cause an exception
91	self._can_send = False	1✔
92	self._remote_ip_address: Optional[str] = None	1✔
93	self._remote_port: Optional[int] = None	1✔
94
95	# Get the host to connect to remotely
96	if remote_host is not None and remote_port is not None:	1✔
97	self._remote_port = remote_port	1✔
98	self._remote_ip_address = resolve_host(remote_host)	1✔
99	connect_socket(self._socket, self._remote_ip_address, remote_port)	1✔
100	self._can_send = True	1✔
101
102	# If things are closed here, it's a catastrophic problem
103	if self.__is_closed:	1!
104	raise SpinnmanEOFException()	×
105
106	# Get the details of where the socket is connected
107	self._local_ip_address, self._local_port = \	1✔
108	get_socket_address(self._socket)
109
110	# Set a general timeout on the socket
111	self._socket.settimeout(1.0)	1✔
112
113	@property	1✔
114	def __is_closed(self) -> bool:	1✔
115	"""
116	Is the socket closed?
117
118	.. note::
119	Just because a socket is not closed doesn't mean that you're going
120	to be able to successfully write to it or read from it; some
121	failures are only detected on use. But closed sockets definitely
122	behave in certain ways!
123
124	:rtype: bool
125	"""
126	# Reach into Python'#s guts
127	# pylint: disable=protected-access
128	return self._socket._closed # type: ignore[attr-defined]	1✔
129
130	@overrides(Connection.is_connected)	1✔
131	def is_connected(self) -> bool:	1✔
132	# Closed sockets are never connected!
133	if self.__is_closed:	×
134	return False	×
135
136	# If this is not a sending socket, it is not connected
137	if not self._can_send:	×
138	return False	×
139
140	# check if machine is active and on the network
141	for _ in range(_PING_COUNT):	×
142	# Assume connected if ping works
143	if Ping.ping(self._remote_ip_address) == 0:	×
144	return True	×
145
146	# If the ping fails this number of times, the host cannot be contacted
147	return False	×
148
149	@property	1✔
150	def local_ip_address(self) -> str:	1✔
151	"""
152	The local IP address to which the connection is bound,
153	as a dotted string, e.g., `0.0.0.0`.
154
155	:rtype: str
156	"""
157	return self._local_ip_address	×
158
159	@property	1✔
160	def local_port(self) -> int:	1✔
161	"""
162	The number of the local port to which the connection is bound.
163
164	:rtype: int
165	"""
166	return self._local_port	×
167
168	@property	1✔
169	def remote_ip_address(self) -> Optional[str]:	1✔
170	"""
171	The remote IP address to which the connection is connected,
172	or `None` if not connected remotely.
173
174	:rtype: str
175	"""
176	return self._remote_ip_address	1✔
177
178	@property	1✔
179	def remote_port(self) -> Optional[int]:	1✔
180	"""
181	The remote port number to which the connection is connected,
182	or `None` if not connected remotely.
183
184	:rtype: int
185	"""
186	return self._remote_port	×
187
188	def receive(self, timeout: Optional[float] = None) -> bytes:	1✔
189	"""
190	Receive data from the connection.
191
192	:param float timeout: The timeout in seconds, or `None` to wait forever
193	:return: The data received as a byte-string
194	:rtype: bytes
195	:raise SpinnmanTimeoutException:
196	If a timeout occurs before any data is received
197	:raise SpinnmanIOException: If an error occurs receiving the data
198	"""
199	if self.__is_closed:	1!
200	raise SpinnmanEOFException()	×
201	return receive_message(self._socket, timeout, _MSG_MAX)	1✔
202
203	def receive_with_address(self, timeout=None):	1✔
204	"""
205	Receive data from the connection along with the address where the
206	data was received from.
207
208	:param float timeout: The timeout, or `None` to wait forever
209	:return: A tuple of the data received and a tuple of the
210	(address, port) received from
211	:rtype: tuple(bytes, tuple(str, int))
212	:raise SpinnmanTimeoutException:
213	If a timeout occurs before any data is received
214	:raise SpinnmanIOException: If an error occurs receiving the data
215	"""
216	if self.__is_closed:	×
217	raise SpinnmanEOFException()	×
218	return receive_message_and_address(self._socket, timeout, _MSG_MAX)	×
219
220	def send(self, data: bytes):	1✔
221	"""
222	Send data down this connection.
223
224	:param data: The data to be sent
225	:type data: bytes or bytearray
226	:raise SpinnmanIOException: If there is an error sending the data
227	"""
228	if self.__is_closed:	1!
229	raise SpinnmanEOFException()	×
230	if not self._can_send:	1!
231	raise SpinnmanIOException(	×
232	"Remote host and/or port not set - data cannot be sent with"
233	" this connection")
234	while not send_message(self._socket, data):	1!
235	if self.__is_closed:	×
236	raise SpinnmanEOFException()	×
237
238	def send_to(self, data: bytes, address: Tuple[str, int]):	1✔
239	"""
240	Send data down this connection.
241
242	:param data: The data to be sent as a byte-string
243	:type data: bytes or bytearray
244	:param tuple(str,int) address:
245	A tuple of (address, port) to send the data to
246	:raise SpinnmanIOException: If there is an error sending the data
247	"""
248	if self.__is_closed:	×
249	raise SpinnmanEOFException()	×
250	while not send_message_to_address(self._socket, data, address):	×
251	if self.__is_closed:	×
252	raise SpinnmanEOFException()	×
253
254	@overrides(Connection.close)	1✔
255	def close(self) -> None:	1✔
256	if self.__is_closed:	×
257	return	×
258	with suppress(Exception):	×
259	self._socket.shutdown(socket.SHUT_WR)	×
260	self._socket.close()	×
261
262	def is_ready_to_receive(self, timeout: float = 0):	1✔
263	if self.__is_closed:	×
264	return True	×
265	return len(select.select([self._socket], [], [], timeout)[0]) == 1	×
266
267	def __repr__(self) -> str:	1✔
268	return _REPR_TEMPLATE.format(	×
269	self.local_ip_address, self.local_port,
270	self.remote_ip_address, self.remote_port)
271
272	@overrides(Listenable.get_receive_method)	1✔
273	def get_receive_method(self) -> Callable[[], bytes]:	1✔
274	return self.receive	×

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