Morpheus / mbed-tools

Important changes to repositories hosted on mbed.com

Mbed hosted mercurial repositories are deprecated and are due to be permanently deleted in July 2026.

To keep a copy of this software download the repository Zip archive or clone locally using Mercurial.

It is also possible to export all your personal repositories from the account settings page.

Fork of mbed-tools by -deleted-

host_tests/udp_link_layer_auto.py@0:4a2e5f0422d6, 2016-03-30 (annotated)

Committer:
screamer
Date:
Wed Mar 30 16:23:44 2016 +0000
Revision:
0:4a2e5f0422d6
Initial revision

Who changed what in which revision?

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