mbed-os/workspace_tools/host_tests/udp_link_layer_auto.py

"""
mbed SDK
Copyright (c) 2011-2013 ARM Limited

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

    http://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.
"""

"""
How to use:
make.py -m LPC1768 -t ARM -d E:\ -n NET_14
udp_link_layer_auto.py -p COM20 -d E:\ -t 10
"""

import re
import uuid
import socket
import thread
from sys import stdout
from time import time, sleep
from host_test import DefaultTest
from SocketServer import BaseRequestHandler, UDPServer


# Received datagrams (with time)
dict_udp_recv_datagrams = dict()

# Sent datagrams (with time)
dict_udp_sent_datagrams = dict()


class UDPEchoClient_Handler(BaseRequestHandler):
    def handle(self):
        """ One handle per connection
        """
        _data, _socket = self.request
        # Process received datagram
        data_str = repr(_data)[1:-1]
        dict_udp_recv_datagrams[data_str] = time()


def udp_packet_recv(threadName, server_ip, server_port):
    """ This function will receive packet stream from mbed device
    """
    server = UDPServer((server_ip, server_port), UDPEchoClient_Handler)
    print "[UDP_COUNTER] Listening for connections... %s:%d"% (server_ip, server_port)
    server.serve_forever()


class UDPEchoServerTest(DefaultTest):
    ECHO_SERVER_ADDRESS = ""    # UDP IP of datagram bursts
    ECHO_PORT = 0               # UDP port for datagram bursts
    CONTROL_PORT = 23           # TCP port used to get stats from mbed device, e.g. counters
    s = None                    # Socket

    TEST_PACKET_COUNT = 1000    # how many packets should be send
    TEST_STRESS_FACTOR = 0.001  # stress factor: 10 ms
    PACKET_SATURATION_RATIO = 29.9 # Acceptable packet transmission in %

    PATTERN_SERVER_IP = "Server IP Address is (\d+).(\d+).(\d+).(\d+):(\d+)"
    re_detect_server_ip = re.compile(PATTERN_SERVER_IP)

    def get_control_data(self, command="stat\n"):
        BUFFER_SIZE = 256
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.connect((self.ECHO_SERVER_ADDRESS, self.CONTROL_PORT))
        except Exception, e:
            data = None
        s.send(command)
        data = s.recv(BUFFER_SIZE)
        s.close()
        return data

    def run(self):
        serial_ip_msg = self.mbed.serial_readline()
        if serial_ip_msg is None:
            self.print_result("ioerr_serial")
            return
        stdout.write(serial_ip_msg)
        stdout.flush()
        # Searching for IP address and port prompted by server
        m = self.re_detect_server_ip.search(serial_ip_msg)
        if m and len(m.groups()):
            self.ECHO_SERVER_ADDRESS = ".".join(m.groups()[:4])
            self.ECHO_PORT = int(m.groups()[4]) # must be integer for socket.connect method
            print "HOST: UDP Server found at: " + self.ECHO_SERVER_ADDRESS + ":" + str(self.ECHO_PORT)
            stdout.flush()

        # Open client socket to burst datagrams to UDP server in mbed
        try:
            self.s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        except Exception, e:
            self.s = None
            print "HOST: Error: %s" % e
            self.print_result('error')
            return

        # UDP replied receiver works in background to get echoed datagrams
        SERVER_IP = str(socket.gethostbyname(socket.getfqdn()))
        SERVER_PORT = self.ECHO_PORT + 1
        thread.start_new_thread(udp_packet_recv, ("Thread-udp-recv", SERVER_IP, SERVER_PORT))
        sleep(0.5)

        # Burst part
        for no in range(self.TEST_PACKET_COUNT):
            TEST_STRING = str(uuid.uuid4())
            payload = str(no) + "__" + TEST_STRING
            self.s.sendto(payload, (self.ECHO_SERVER_ADDRESS, self.ECHO_PORT))
            dict_udp_sent_datagrams[payload] = time()
            sleep(self.TEST_STRESS_FACTOR)

        if self.s is not None:
            self.s.close()

        # Wait 5 seconds for packets to come
        result = True
        print
        print "HOST: Test Summary:"
        for d in range(5):
            sleep(1.0)
            summary_datagram_success = (float(len(dict_udp_recv_datagrams)) / float(self.TEST_PACKET_COUNT)) * 100.0
            # print dict_udp_recv_datagrams
            print "HOST: Datagrams received after +%d sec: %.3f%% (%d / %d), stress=%.3f ms" % (d, summary_datagram_success, len(dict_udp_recv_datagrams), self.TEST_PACKET_COUNT, self.TEST_STRESS_FACTOR)
            result = result and (summary_datagram_success >= self.PACKET_SATURATION_RATIO)
            stdout.flush()

        # Getting control data from test
        print
        print "HOST: Mbed Summary:"
        mbed_stats = self.get_control_data()
        print mbed_stats
        print
        stdout.flush()

        if result:
            self.print_result('success')
        else:
            self.print_result('failure')

        # Receiving serial data from mbed
        print
        print "HOST: Remaining mbed serial port data:"
        try:
            while True:
                c = self.mbed.serial_read(512)
                if c is None:
                    self.print_result("ioerr_serial")
                    break
                stdout.write(c)
                stdout.flush()
        except KeyboardInterrupt, _:
            print "\n[CTRL+c] exit"

if __name__ == '__main__':
    UDPEchoServerTest().run()