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

23.08

/spinnman/utilities/utility_functions.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 socket
from spinnman.model import BMPConnectionData
from spinnman.messages.scp.impl import IPTagSet
from spinnman.messages.sdp import SDPMessage, SDPHeader, SDPFlag
from spinnman.constants import SCP_SCAMP_PORT, CPU_INFO_BYTES, CPU_INFO_OFFSET
from spinnman.connections.udp_packet_connections import (
    SCAMPConnection, UDPConnection)
from spinnman.exceptions import SpinnmanTimeoutException


def work_out_bmp_from_machine_details(
        hostname: str) -> BMPConnectionData:
    """
    Work out the BMP connection IP address given the machine details.
    This is assumed to be the IP address of the machine, with 1 subtracted
    from the final part e.g. if the machine IP address is 192.168.0.5, the
    BMP IP address is assumed to be 192.168.0.4

    :param str hostname: the SpiNNaker machine main hostname or IP address
    :return: The BMP connection data
    :rtype: BMPConnectionData
    """
    # take the IP address, split by dots, and subtract 1 off last bit
    ip_bits = socket.gethostbyname(hostname).split(".")
    ip_bits[-1] = str(int(ip_bits[-1]) - 1)
    bmp_ip_address = ".".join(ip_bits)

    # add board scope for each split
    # if None, the end user didn't enter anything, so assume one board
    # starting at position 0
    board_range = [0]

    # Assume a single board with no cabinet or frame specified
    return BMPConnectionData(ip_address=bmp_ip_address,
                             boards=board_range, port_num=SCP_SCAMP_PORT)


def get_vcpu_address(p: int) -> int:
    """
    Get the address of the vcpu_t structure for the given core.

    :param int p: The core
    :rtype: int
    """
    return CPU_INFO_OFFSET + (CPU_INFO_BYTES * p)


def send_port_trigger_message(
        connection: UDPConnection, board_address: str):
    """
    Sends a port trigger message using a connection to (hopefully) open a
    port in a NAT and/or firewall to allow incoming packets to be received.

    :param UDPConnection connection:
        The UDP connection down which the trigger message should be sent
    :param str board_address:
        The IP address of the SpiNNaker board to which the message should be
        sent
    """

    # Set up the message so that no reply is expected and it is sent to an
    # invalid port for SCAMP.  The current version of SCAMP will reject
    # this message, but then fail to send a response since the
    # REPLY_NOT_EXPECTED flag is set (see scamp-3.c line 728 and 625-644)
    trigger_message = SDPMessage(SDPHeader(
        flags=SDPFlag.REPLY_NOT_EXPECTED, tag=0, destination_port=3,
        destination_cpu=0, destination_chip_x=0, destination_chip_y=0))
    trigger_message.sdp_header.update_for_send(0, 0)
    connection.send_to(
        trigger_message.bytestring, (board_address, SCP_SCAMP_PORT))


def reprogram_tag(connection: SCAMPConnection, tag: int, strip: bool = True):
    """
    Reprogram an IP Tag to send responses to a given SCAMPConnection.

    :param SCAMPConnection connection: The connection to target the tag at
    :param int tag: The id of the tag to set
    :param bool strip:
        True if the tag should strip SDP headers from outgoing messages
    :raises SpinnmanTimeoutException:
        If things time out several times
    """
    request = IPTagSet(
        connection.chip_x, connection.chip_y, [0, 0, 0, 0], 0, tag,
        strip=strip, use_sender=True)
    data = connection.get_scp_data(request)
    exn = None
    for _ in range(3):
        try:
            connection.send(data)
            _, _, response, offset = connection.receive_scp_response()
            request.get_scp_response().read_bytestring(response, offset)
            return
        except SpinnmanTimeoutException as e:
            exn = e
    # Should be impossible to get here with exn=None
    raise exn or Exception


def reprogram_tag_to_listener(
        connection: UDPConnection, x: int, y: int, ip_address: str, tag: int,
        strip: bool = True, read_response: bool = True):
    """
    Reprogram an IP Tag to send responses to a given connection that is
    not connected to a specific board. Such connections are normally
    receive-only connections.

    :param UDPConnection connection: The connection to target the tag at
    :param int x:
        The X coordinate of the Ethernet-enabled chip that should send to the
        connection
    :param int y:
        The Y coordinate of the Ethernet-enabled chip that should send to the
        connection
    :param str ip_address:
        The IP address of the Ethernet-enabled chip that should be given the
        message
    :param int tag: The id of the tag to set
    :param bool strip:
        True if the tag should strip SDP headers from outgoing messages
    :param bool read_response:
        True if the response to the reprogramming should be read
    :raises SpinnmanTimeoutException:
        If things time out several times
    """
    request = IPTagSet(
        x, y, [0, 0, 0, 0], 0, tag,
        strip=strip, use_sender=True)
    request.sdp_header.update_for_send(x, y)
    send_data = b'\0\0' + request.bytestring
    exn = None
    for _ in range(3):
        try:
            connection.send_to(send_data, (ip_address, SCP_SCAMP_PORT))
            if read_response:
                request.get_scp_response().read_bytestring(
                    connection.receive(), 2)
            return
        except SpinnmanTimeoutException as e:
            exn = e
    # Should be impossible to get here with exn=None
    raise exn or Exception

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 socket	1✔
16	from spinnman.model import BMPConnectionData	1✔
17	from spinnman.messages.scp.impl import IPTagSet	1✔
18	from spinnman.messages.sdp import SDPMessage, SDPHeader, SDPFlag	1✔
19	from spinnman.constants import SCP_SCAMP_PORT, CPU_INFO_BYTES, CPU_INFO_OFFSET	1✔
20	from spinnman.connections.udp_packet_connections import (	1✔
21	SCAMPConnection, UDPConnection)
22	from spinnman.exceptions import SpinnmanTimeoutException	1✔
23
24
25	def work_out_bmp_from_machine_details(	1✔
26	hostname: str) -> BMPConnectionData:
27	"""
28	Work out the BMP connection IP address given the machine details.
29	This is assumed to be the IP address of the machine, with 1 subtracted
30	from the final part e.g. if the machine IP address is 192.168.0.5, the
31	BMP IP address is assumed to be 192.168.0.4
32
33	:param str hostname: the SpiNNaker machine main hostname or IP address
34	:return: The BMP connection data
35	:rtype: BMPConnectionData
36	"""
37	# take the IP address, split by dots, and subtract 1 off last bit
38	ip_bits = socket.gethostbyname(hostname).split(".")	×
39	ip_bits[-1] = str(int(ip_bits[-1]) - 1)	×
40	bmp_ip_address = ".".join(ip_bits)	×
41
42	# add board scope for each split
43	# if None, the end user didn't enter anything, so assume one board
44	# starting at position 0
45	board_range = [0]	×
46
47	# Assume a single board with no cabinet or frame specified
48	return BMPConnectionData(ip_address=bmp_ip_address,	×
49	boards=board_range, port_num=SCP_SCAMP_PORT)
50
51
52	def get_vcpu_address(p: int) -> int:	1✔
53	"""
54	Get the address of the vcpu_t structure for the given core.
55
56	:param int p: The core
57	:rtype: int
58	"""
59	return CPU_INFO_OFFSET + (CPU_INFO_BYTES * p)	×
60
61
62	def send_port_trigger_message(	1✔
63	connection: UDPConnection, board_address: str):
64	"""
65	Sends a port trigger message using a connection to (hopefully) open a
66	port in a NAT and/or firewall to allow incoming packets to be received.
67
68	:param UDPConnection connection:
69	The UDP connection down which the trigger message should be sent
70	:param str board_address:
71	The IP address of the SpiNNaker board to which the message should be
72	sent
73	"""
74
75	# Set up the message so that no reply is expected and it is sent to an
76	# invalid port for SCAMP. The current version of SCAMP will reject
77	# this message, but then fail to send a response since the
78	# REPLY_NOT_EXPECTED flag is set (see scamp-3.c line 728 and 625-644)
79	trigger_message = SDPMessage(SDPHeader(	×
80	flags=SDPFlag.REPLY_NOT_EXPECTED, tag=0, destination_port=3,
81	destination_cpu=0, destination_chip_x=0, destination_chip_y=0))
82	trigger_message.sdp_header.update_for_send(0, 0)	×
83	connection.send_to(	×
84	trigger_message.bytestring, (board_address, SCP_SCAMP_PORT))
85
86
87	def reprogram_tag(connection: SCAMPConnection, tag: int, strip: bool = True):	1✔
88	"""
89	Reprogram an IP Tag to send responses to a given SCAMPConnection.
90
91	:param SCAMPConnection connection: The connection to target the tag at
92	:param int tag: The id of the tag to set
93	:param bool strip:
94	True if the tag should strip SDP headers from outgoing messages
95	:raises SpinnmanTimeoutException:
96	If things time out several times
97	"""
98	request = IPTagSet(	×
99	connection.chip_x, connection.chip_y, [0, 0, 0, 0], 0, tag,
100	strip=strip, use_sender=True)
101	data = connection.get_scp_data(request)	×
102	exn = None	×
103	for _ in range(3):	×
104	try:	×
105	connection.send(data)	×
106	_, _, response, offset = connection.receive_scp_response()	×
107	request.get_scp_response().read_bytestring(response, offset)	×
108	return	×
109	except SpinnmanTimeoutException as e:	×
110	exn = e	×
111	# Should be impossible to get here with exn=None
112	raise exn or Exception	×
113
114
115	def reprogram_tag_to_listener(	1✔
116	connection: UDPConnection, x: int, y: int, ip_address: str, tag: int,
117	strip: bool = True, read_response: bool = True):
118	"""
119	Reprogram an IP Tag to send responses to a given connection that is
120	not connected to a specific board. Such connections are normally
121	receive-only connections.
122
123	:param UDPConnection connection: The connection to target the tag at
124	:param int x:
125	The X coordinate of the Ethernet-enabled chip that should send to the
126	connection
127	:param int y:
128	The Y coordinate of the Ethernet-enabled chip that should send to the
129	connection
130	:param str ip_address:
131	The IP address of the Ethernet-enabled chip that should be given the
132	message
133	:param int tag: The id of the tag to set
134	:param bool strip:
135	True if the tag should strip SDP headers from outgoing messages
136	:param bool read_response:
137	True if the response to the reprogramming should be read
138	:raises SpinnmanTimeoutException:
139	If things time out several times
140	"""
141	request = IPTagSet(	×
142	x, y, [0, 0, 0, 0], 0, tag,
143	strip=strip, use_sender=True)
144	request.sdp_header.update_for_send(x, y)	×
145	send_data = b'\0\0' + request.bytestring	×
146	exn = None	×
147	for _ in range(3):	×
148	try:	×
149	connection.send_to(send_data, (ip_address, SCP_SCAMP_PORT))	×
150	if read_response:	×
151	request.get_scp_response().read_bytestring(	×
152	connection.receive(), 2)
153	return	×
154	except SpinnmanTimeoutException as e:	×
155	exn = e	×
156	# Should be impossible to get here with exn=None
157	raise exn or Exception	×

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