Anders Blomdell / mbed-sdk-tools

Clone of official tools

host_tests/udp_link_layer_auto.py@47:21ae3e5a7128, 2021-02-04 (annotated)

Committer:
Anders Blomdell
Date:
Thu Feb 04 17:17:13 2021 +0100
Revision:
47:21ae3e5a7128
Parent:
0:66f3b5499f7f 
Add a few normpath calls

Who changed what in which revision?

UserRevisionLine numberNew contents of line
screamer 0:66f3b5499f7f 1 """
screamer 0:66f3b5499f7f 2 mbed SDK
screamer 0:66f3b5499f7f 3 Copyright (c) 2011-2013 ARM Limited
screamer 0:66f3b5499f7f 4
screamer 0:66f3b5499f7f 5 Licensed under the Apache License, Version 2.0 (the "License");
screamer 0:66f3b5499f7f 6 you may not use this file except in compliance with the License.
screamer 0:66f3b5499f7f 7 You may obtain a copy of the License at
screamer 0:66f3b5499f7f 8
screamer 0:66f3b5499f7f 9 http://www.apache.org/licenses/LICENSE-2.0
screamer 0:66f3b5499f7f 10
screamer 0:66f3b5499f7f 11 Unless required by applicable law or agreed to in writing, software
screamer 0:66f3b5499f7f 12 distributed under the License is distributed on an "AS IS" BASIS,
screamer 0:66f3b5499f7f 13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
screamer 0:66f3b5499f7f 14 See the License for the specific language governing permissions and
screamer 0:66f3b5499f7f 15 limitations under the License.
screamer 0:66f3b5499f7f 16 """
screamer 0:66f3b5499f7f 17
screamer 0:66f3b5499f7f 18 """
screamer 0:66f3b5499f7f 19 How to use:
screamer 0:66f3b5499f7f 20 make.py -m LPC1768 -t ARM -d E:\ -n NET_14
screamer 0:66f3b5499f7f 21 udp_link_layer_auto.py -p COM20 -d E:\ -t 10
screamer 0:66f3b5499f7f 22 """
screamer 0:66f3b5499f7f 23
screamer 0:66f3b5499f7f 24 import re
screamer 0:66f3b5499f7f 25 import uuid
screamer 0:66f3b5499f7f 26 import socket
screamer 0:66f3b5499f7f 27 import thread
screamer 0:66f3b5499f7f 28 from sys import stdout
screamer 0:66f3b5499f7f 29 from time import time, sleep
screamer 0:66f3b5499f7f 30 from host_test import DefaultTest
screamer 0:66f3b5499f7f 31 from SocketServer import BaseRequestHandler, UDPServer
screamer 0:66f3b5499f7f 32
screamer 0:66f3b5499f7f 33
screamer 0:66f3b5499f7f 34 # Received datagrams (with time)
screamer 0:66f3b5499f7f 35 dict_udp_recv_datagrams = dict()
screamer 0:66f3b5499f7f 36
screamer 0:66f3b5499f7f 37 # Sent datagrams (with time)
screamer 0:66f3b5499f7f 38 dict_udp_sent_datagrams = dict()
screamer 0:66f3b5499f7f 39
screamer 0:66f3b5499f7f 40
screamer 0:66f3b5499f7f 41 class UDPEchoClient_Handler(BaseRequestHandler):
screamer 0:66f3b5499f7f 42 def handle(self):
screamer 0:66f3b5499f7f 43 """ One handle per connection
screamer 0:66f3b5499f7f 44 """
screamer 0:66f3b5499f7f 45 _data, _socket = self.request
screamer 0:66f3b5499f7f 46 # Process received datagram
screamer 0:66f3b5499f7f 47 data_str = repr(_data)[1:-1]
screamer 0:66f3b5499f7f 48 dict_udp_recv_datagrams[data_str] = time()
screamer 0:66f3b5499f7f 49
screamer 0:66f3b5499f7f 50
screamer 0:66f3b5499f7f 51 def udp_packet_recv(threadName, server_ip, server_port):
screamer 0:66f3b5499f7f 52 """ This function will receive packet stream from mbed device
screamer 0:66f3b5499f7f 53 """
screamer 0:66f3b5499f7f 54 server = UDPServer((server_ip, server_port), UDPEchoClient_Handler)
screamer 0:66f3b5499f7f 55 print "[UDP_COUNTER] Listening for connections... %s:%d"% (server_ip, server_port)
screamer 0:66f3b5499f7f 56 server.serve_forever()
screamer 0:66f3b5499f7f 57
screamer 0:66f3b5499f7f 58
screamer 0:66f3b5499f7f 59 class UDPEchoServerTest(DefaultTest):
screamer 0:66f3b5499f7f 60 ECHO_SERVER_ADDRESS = "" # UDP IP of datagram bursts
screamer 0:66f3b5499f7f 61 ECHO_PORT = 0 # UDP port for datagram bursts
screamer 0:66f3b5499f7f 62 CONTROL_PORT = 23 # TCP port used to get stats from mbed device, e.g. counters
screamer 0:66f3b5499f7f 63 s = None # Socket
screamer 0:66f3b5499f7f 64
screamer 0:66f3b5499f7f 65 TEST_PACKET_COUNT = 1000 # how many packets should be send
screamer 0:66f3b5499f7f 66 TEST_STRESS_FACTOR = 0.001 # stress factor: 10 ms
screamer 0:66f3b5499f7f 67 PACKET_SATURATION_RATIO = 29.9 # Acceptable packet transmission in %
screamer 0:66f3b5499f7f 68
screamer 0:66f3b5499f7f 69 PATTERN_SERVER_IP = "Server IP Address is (\d+).(\d+).(\d+).(\d+):(\d+)"
screamer 0:66f3b5499f7f 70 re_detect_server_ip = re.compile(PATTERN_SERVER_IP)
screamer 0:66f3b5499f7f 71
screamer 0:66f3b5499f7f 72 def get_control_data(self, command="stat\n"):
screamer 0:66f3b5499f7f 73 BUFFER_SIZE = 256
screamer 0:66f3b5499f7f 74 try:
screamer 0:66f3b5499f7f 75 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
screamer 0:66f3b5499f7f 76 s.connect((self.ECHO_SERVER_ADDRESS, self.CONTROL_PORT))
screamer 0:66f3b5499f7f 77 except Exception, e:
screamer 0:66f3b5499f7f 78 data = None
screamer 0:66f3b5499f7f 79 s.send(command)
screamer 0:66f3b5499f7f 80 data = s.recv(BUFFER_SIZE)
screamer 0:66f3b5499f7f 81 s.close()
screamer 0:66f3b5499f7f 82 return data
screamer 0:66f3b5499f7f 83
screamer 0:66f3b5499f7f 84 def test(self):
screamer 0:66f3b5499f7f 85 serial_ip_msg = self.mbed.serial_readline()
screamer 0:66f3b5499f7f 86 if serial_ip_msg is None:
screamer 0:66f3b5499f7f 87 return self.RESULT_IO_SERIAL
screamer 0:66f3b5499f7f 88 stdout.write(serial_ip_msg)
screamer 0:66f3b5499f7f 89 stdout.flush()
screamer 0:66f3b5499f7f 90 # Searching for IP address and port prompted by server
screamer 0:66f3b5499f7f 91 m = self.re_detect_server_ip.search(serial_ip_msg)
screamer 0:66f3b5499f7f 92 if m and len(m.groups()):
screamer 0:66f3b5499f7f 93 self.ECHO_SERVER_ADDRESS = ".".join(m.groups()[:4])
screamer 0:66f3b5499f7f 94 self.ECHO_PORT = int(m.groups()[4]) # must be integer for socket.connect method
screamer 0:66f3b5499f7f 95 self.notify("HOST: UDP Server found at: " + self.ECHO_SERVER_ADDRESS + ":" + str(self.ECHO_PORT))
screamer 0:66f3b5499f7f 96
screamer 0:66f3b5499f7f 97 # Open client socket to burst datagrams to UDP server in mbed
screamer 0:66f3b5499f7f 98 try:
screamer 0:66f3b5499f7f 99 self.s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
screamer 0:66f3b5499f7f 100 except Exception, e:
screamer 0:66f3b5499f7f 101 self.s = None
screamer 0:66f3b5499f7f 102 self.notify("HOST: Error: %s"% e)
screamer 0:66f3b5499f7f 103 return self.RESULT_ERROR
screamer 0:66f3b5499f7f 104
screamer 0:66f3b5499f7f 105 # UDP replied receiver works in background to get echoed datagrams
screamer 0:66f3b5499f7f 106 SERVER_IP = str(socket.gethostbyname(socket.getfqdn()))
screamer 0:66f3b5499f7f 107 SERVER_PORT = self.ECHO_PORT + 1
screamer 0:66f3b5499f7f 108 thread.start_new_thread(udp_packet_recv, ("Thread-udp-recv", SERVER_IP, SERVER_PORT))
screamer 0:66f3b5499f7f 109 sleep(0.5)
screamer 0:66f3b5499f7f 110
screamer 0:66f3b5499f7f 111 # Burst part
screamer 0:66f3b5499f7f 112 for no in range(self.TEST_PACKET_COUNT):
screamer 0:66f3b5499f7f 113 TEST_STRING = str(uuid.uuid4())
screamer 0:66f3b5499f7f 114 payload = str(no) + "__" + TEST_STRING
screamer 0:66f3b5499f7f 115 self.s.sendto(payload, (self.ECHO_SERVER_ADDRESS, self.ECHO_PORT))
screamer 0:66f3b5499f7f 116 dict_udp_sent_datagrams[payload] = time()
screamer 0:66f3b5499f7f 117 sleep(self.TEST_STRESS_FACTOR)
screamer 0:66f3b5499f7f 118
screamer 0:66f3b5499f7f 119 if self.s is not None:
screamer 0:66f3b5499f7f 120 self.s.close()
screamer 0:66f3b5499f7f 121
screamer 0:66f3b5499f7f 122 # Wait 5 seconds for packets to come
screamer 0:66f3b5499f7f 123 result = True
screamer 0:66f3b5499f7f 124 self.notify("HOST: Test Summary:")
screamer 0:66f3b5499f7f 125 for d in range(5):
screamer 0:66f3b5499f7f 126 sleep(1.0)
screamer 0:66f3b5499f7f 127 summary_datagram_success = (float(len(dict_udp_recv_datagrams)) / float(self.TEST_PACKET_COUNT)) * 100.0
screamer 0:66f3b5499f7f 128 self.notify("HOST: Datagrams received after +%d sec: %.3f%% (%d / %d), stress=%.3f ms"% (d,
screamer 0:66f3b5499f7f 129 summary_datagram_success,
screamer 0:66f3b5499f7f 130 len(dict_udp_recv_datagrams),
screamer 0:66f3b5499f7f 131 self.TEST_PACKET_COUNT,
screamer 0:66f3b5499f7f 132 self.TEST_STRESS_FACTOR))
screamer 0:66f3b5499f7f 133 result = result and (summary_datagram_success >= self.PACKET_SATURATION_RATIO)
screamer 0:66f3b5499f7f 134 stdout.flush()
screamer 0:66f3b5499f7f 135
screamer 0:66f3b5499f7f 136 # Getting control data from test
screamer 0:66f3b5499f7f 137 self.notify("...")
screamer 0:66f3b5499f7f 138 self.notify("HOST: Mbed Summary:")
screamer 0:66f3b5499f7f 139 mbed_stats = self.get_control_data()
screamer 0:66f3b5499f7f 140 self.notify(mbed_stats)
screamer 0:66f3b5499f7f 141 return self.RESULT_SUCCESS if result else self.RESULT_FAILURE
screamer 0:66f3b5499f7f 142
screamer 0:66f3b5499f7f 143
screamer 0:66f3b5499f7f 144 if __name__ == '__main__':
screamer 0:66f3b5499f7f 145 UDPEchoServerTest().run()