ARMmbed
/
mbed-os
mirror of https://github.com/ARMmbed/mbed-os.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

pyFlakes issues fixed. Simple refactoring

pull/399/head
Przemek Wirkus 2014-07-09 11:15:29 +01:00
parent 3ce8116fb2
commit 7a5d00bd65
1 changed files with 8 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
workspace_tools/host_tests/udp_link_layer_auto.py
View File

@ -23,10 +23,9 @@ udp_link_layer_auto.py -p COM20 -d E:\ -t 10
import thread import thread
from SocketServer import BaseRequestHandler, UDPServer from SocketServer import BaseRequestHandler, UDPServer
# from socket import socket, AF_INET, SOCK_DGRAM
import socket import socket
import re import re
from host_test import Test, DefaultTest from host_test import DefaultTest
from time import time, sleep from time import time, sleep
from sys import stdout from sys import stdout
@ -41,13 +40,10 @@ dict_udp_sent_datagrams = dict()
class UDPEchoClient_Handler(BaseRequestHandler): class UDPEchoClient_Handler(BaseRequestHandler):
def handle(self): def handle(self):
""" One handle per connection """ """ One handle per connection """
data, socket = self.request _data, _socket = self.request
# Process received datagram # Process received datagram
data_str = repr(data)[1:-1] data_str = repr(_data)[1:-1]
dict_udp_recv_datagrams[data_str] = time() dict_udp_recv_datagrams[data_str] = time()
# print "[UDP_COUNTER] [" + data_str + "]"
# print ".",
def udp_packet_recv(threadName, server_ip, server_port): def udp_packet_recv(threadName, server_ip, server_port):
@ -66,11 +62,8 @@ class UDPEchoServerTest(DefaultTest):
TEST_PACKET_COUNT = 1000 # how many packets should be send TEST_PACKET_COUNT = 1000 # how many packets should be send
TEST_STRESS_FACTOR = 0.001 # stress factor: 10 ms TEST_STRESS_FACTOR = 0.001 # stress factor: 10 ms
pattern_server_ip = "^Server IP Address is (\d+).(\d+).(\d+).(\d+):(\d+)" PATTERN_SERVER_IP = "^Server IP Address is (\d+).(\d+).(\d+).(\d+):(\d+)"
re_detect_server_ip = re.compile(pattern_server_ip) re_detect_server_ip = re.compile(PATTERN_SERVER_IP)
def print_result(self, result):
print "\n{%s}\n{end}" % result
def get_control_data(self, command="stat\n"): def get_control_data(self, command="stat\n"):
BUFFER_SIZE = 256 BUFFER_SIZE = 256
@ -84,7 +77,7 @@ class UDPEchoServerTest(DefaultTest):
def run(self): def run(self):
ip_msg_timeout = self.mbed.options.timeout ip_msg_timeout = self.mbed.options.timeout
serial_ip_msg = "" serial_ip_msg = ""
start_serial_pool = time(); start_serial_pool = time()
while (time() - start_serial_pool) < ip_msg_timeout: while (time() - start_serial_pool) < ip_msg_timeout:
c = self.mbed.serial.read(512) c = self.mbed.serial.read(512)
stdout.write(c) stdout.write(c)