c180_testzone/python_test/lds.py

#!/usr/bin/python
#_*_ coding: utf-8 _*_

from Tkinter import *
import sys
import os
import subprocess
import time
import ttk
import tkMessageBox
import threading
import csv
import subprocess
from datetime import datetime

# declaring global variables
script_ver='MXF301-B-C7-v1.1'
log_file='./Logs/temp.txt'
log1_buffer = ""
step_status = ['x', 'ns', 'ns']
result_msg = []

# 'x' = don't care
# 'ns' = not started
# 'p' = passed
# 'f' = failed
# 'r' = running
test_channel = ['1', '6', '11']
device_info={}
params_dict = {}
params_dict['path_loss'] = '25'
params_dict['wave_file'] = './11b_waveforms/wave11rc_1000.mod'
params_dict['per_limit'] = '92'
params_dict['SA_RSSI'] = '-80'
params_dict['BTOn_RSSI']='-30'

counter=0
running=False




###########################################################################
class IO_process:
	
	def __init__(self, log_file_path):
		# log_file_path is the absolute path of the log file
		self.log_file = log_file_path
		self.line_count = 0
		
	def read_file(self):
		try:
			f = open(self.log_file,'r') # open file to read	
			flag = 0
			for i, line in enumerate(f):
				if i == self.line_count: # if reached the correct line
					self.line_count += 1
					flag = 1
					#print self.line_count
					break
			f.close()
			if flag == 1:
				return line.rstrip('\r') # get rid of the newline character
			else:
				return ""
		except:
			return ""
	
	def empty_file(self):
		f = open(self.log_file,'w')
		f.close()
		
	def reset_counter(self):
		self.line_count = 0

###########################################################################
def write_log_to_file(content):
	curr_time=time.strftime("%Y-%m-%d-%H-%M-%S", time.localtime())
	log_dir = "./Logs/"
	log_file = "log_" + curr_time + ".txt"
	log_file_location = log_dir + log_file
	file = open(log_file_location,'a') # open file to append
	file.write(str(content) + '\n')
	file.close()

###########################################################################
def print1(string,newline=0):
	global log1_buffer

	if newline == 0: # if user doesn't want to insert a line feed
		log1_buffer += "\n" + str(string)
	elif newline == 1: # if user wants to insert a line feed
		log1_buffer += "\n" + str(string) + "\n"
		
###########################################################################
def step1_routine():
# bios_version
	global step_status
	global result_msg
	step_num = 1
	step_status[step_num] = 'r' # running
	
	print1("\nstep1----------------------------------------------------")
	print1("Step " + str(step_num) + " Started...")
	time.sleep(1)		
	#a = subprocess.call(['./bios-ver.sh'])
        a = os.popen("ls").read()
        a = a.replace('\n','')
        a = a.replace('\r','')
        a = a.strip()
	if len(a)  != 0:
		print1("Error! No Device found!")
		print1(a)
		step_status[step_num] = 'f' # failed
	else:
		print1("Success! Device found! Rsync completed!")
		print1(a)
		step_status[step_num] = 'p' # passed
                result_msg[step_num] = a
        
def step2_routine():
# bios_version
	global step_status
	global result_msg
	step_num = 2
	step_status[step_num] = 'r' # running
	
	print1("\nstep1----------------------------------------------------")
	print1("Step " + str(step_num) + " Started...")
	time.sleep(1)		
	#a = subprocess.call(['./bios-ver.sh'])
        a = os.popen("ls").read()
        a = a.replace('\n','')
        a = a.replace('\r','')
        a = a.strip()
	if len(a) != 0:
		print1("Error! No Device found!")
		print1(a)
		step_status[step_num] = 'f' # failed
	else:
		print1("Success! Device found! Rsync completed!")
		print1(a)
		step_status[step_num] = 'p' # passed
                result_msg[step_num] = a
#$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$
class GuiThread(object):
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def __init__(self):
		global log_file
                global result_msg
		self.thread_list = {}
		
		self.root = Tk()
		self.root.geometry("1800x1200")
		self.running=False
		self.counter=0
		self.root.protocol("WM_DELETE_WINDOW", self.destroy) # when user closes the window
		self.root.title("WiPER: DFU/SOC Simular Test")
		
		self.note1 = Label(self.root, text="Please press 'Start Test' to begin...",  fg='blue')
		self.button1 = Button(self.root, text="Start Test", command=self.start)  #lambda:
		self.lbl = Label(self.root, text="NOT STARTED", width=15, bg='gray')
		self.lbl2 = Label(self.root, text="Device: ",width=20,bg="#C6EBFF")
		
		
		self.nb = ttk.Notebook(self.root, name='notebook')
		
		self.textframe1 = Frame(self.root)
		self.text1 = Text(self.textframe1)
		self.scrollbar1 = Scrollbar(self.textframe1)
		
		self.paramframe = Frame(self.root)
		
		self.lblframe = Frame(self.root)
		#self.step1_lbl1 = Label(self.lblframe, text="Step1: Check BIOS Version                 ", width=50)
		self.step1_lbl1 = Label(self.lblframe, text="Step1: 检查BIOS版本号                 ", width=50)
		self.step2_lbl1 = Label(self.lblframe, text="Step2: 检查BIOS版本号                 ", width=50)
		self.step1_lbl2 = Label(self.lblframe, text="NOT STARTED", bg='gray', width=50)
		self.step2_lbl2 = Label(self.lblframe, text="NOT STARTED", bg='gray', width=50)

		self.step_lbl_list = []
		self.step_lbl_list.append('x') # don't care
		self.step_lbl_list.append(self.step1_lbl2)
		self.step_lbl_list.append(self.step2_lbl2) 
		
                result_msg.append('P')
                result_msg.append('P')
                result_msg.append('P')
		self.my_io_process = IO_process(log_file)		
		self.readLog1()
		self.set_param_frame()		 

# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def destroy(self):
		if tkMessageBox.askokcancel("Quit?", "Are you sure you want to quit?"):
			self.root.quit()
		

	def counter_label(self):
		def count():
			if self.running:
				global counter

				# To manage the initial delay.
				if self.counter==0: #66600
					display="Starting..."
				else:
					tt=datetime.fromtimestamp(self.counter)
					string=tt.strftime("%H:%M:%S")
					display=string
					#display="kskjskdjd"					
					#self.note1.config(text='RUNNING...'
				#self.note1['text']=display
				#self.lbl.config(text='RUNNING...')
				self.note1.config(text=display)
				#self.note1.config['text']=display
				self.note1.after(1000, count)
				self.counter+=1
			
		count()
		
			
	# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def run(self):
	
		self.note1.pack()
		self.button1.pack(side=TOP, padx = 10, pady = 3)
		self.lbl.pack()
		self.lbl2.pack(fill=BOTH,expand=Y)
		
		self.scrollbar1.config(command=self.text1.yview)
		self.text1.config(yscrollcommand=self.scrollbar1.set)
		self.text1.pack(side=LEFT, fill=Y)
		self.scrollbar1.pack(side=RIGHT, fill=Y)
		
		self.step1_lbl1.grid(row=0,column=0,sticky=W)
		self.step2_lbl1.grid(row=1,column=0,sticky=W)
		
		self.step1_lbl2.grid(row=0,column=1,sticky=W)
		self.step2_lbl2.grid(row=1,column=1,sticky=W)
		
                self.lblframe.pack()
		self.textframe1.pack()
		self.paramframe.pack()
		
		self.nb.add(self.lblframe, text='Status')
		self.nb.add(self.textframe1, text='Log')
		self.nb.add(self.paramframe, text='Parameters')
		self.nb.enable_traversal() # enable keyboard traversal using Ctrl+Tab
		self.nb.pack(fill=BOTH, expand="true", padx=5, pady=1)
		
		self.root.mainloop()
		
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def readLog1(self):
	# This function periodically reads and empties the log buffer
		
		global log1_buffer
		if log1_buffer != "":
			self.text1.insert(END, log1_buffer)
		
		text = self.my_io_process.read_file()
		if text != "":
			self.text1.insert(END, text)

		if log1_buffer != "" or text != "": # if log buffer was populated
			self.text1.yview_pickplace("end") # move the scrollbar to the bottom
			
		log1_buffer = "" # empty the log buffer
		self.root.update() # update the screen
		self.root.after(100, self.readLog1)

## # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def set_param_frame(self):
		global params_dict
		self.L = []
		self.E = []
		
		count = 1
		for key,value in params_dict.iteritems():
			lbl = Label(self.paramframe, text=key)
			enty = Entry(self.paramframe, bd = 4)
			enty.delete(0, END)
			enty.insert(0,value)
			enty.config(state=DISABLED)
			lbl.grid(row=count, sticky=W)
			enty.grid(row=count, column=1)
			self.L.append(lbl)
			self.E.append(enty)
			count = count + 1

## # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def start(self):
	
		global step_status
		#running=True
		#self.counter_label()
		#step_status = ['x', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns', 'ns'] #
		step_status = ['x', 'ns', 'ns'] #
		self.button1.config(state=DISABLED) # disable the start button
		self.text1.delete('1.0', 'end') # delete the log

		self.my_io_process.reset_counter()
		self.thread_list['thread'] = MasterThread(self)
		self.thread_list['thread'].start()
		self.manage()
		self.manage_lblframe()


## # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def manage(self):
	# manage the pop-up dialog boxes
		
		global step_status
		total_no_of_steps = len(step_status) - 1
		flag = 0
		count = 0
		for i in range(1,total_no_of_steps+1):
			if step_status[i] == 'f': # if failed
				#tkMessageBox.showinfo("Message Box", "Error on Step " + str(i) + "! See Log for details.")
				print1("Error on Step " + str(i) + "! See Log for details.")
				flag = 1
				time.sleep(1)
				write_log_to_file(self.text1.get('1.0', 'end'))
				print1("Log written to the external txt file under ./Logs/")
			if step_status[i] == 'p' or step_status[i] == 'f': # if passed
				count = count + 1
				
		if count == total_no_of_steps: # if all the steps were successful
			flag = 1
			tkMessageBox.showinfo("Message Box", "Test succesfully done!") #Please safely remove the device from the chamber.
			print1("Test succesfully done! Please safely remove the device from the chamber.")
			time.sleep(1)
			write_log_to_file(self.text1.get('1.0', 'end'))
			print1("Log written to the external txt file under ./Logs/")
			self.running=False
			self.counter=0
		if flag == 0: # if not(any step failed or all steps passed)
			self.root.after(500, self.manage)
		else: # if any step has failed or if all the steps have passed
			self.button1.config(state=NORMAL) # enable the start button
			self.my_io_process.empty_file()
			return 0

## # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
	def manage_lblframe(self):
	# manage the frame that displays the step-by-step status
	# manage the top-level status Label
	
		global step_status
		global device_info
		global result_msg
		total_no_of_steps = len(step_status) - 1
		try:
			SrNo = device_info['SrNo']
			string = "Device: " + SrNo
			self.lbl2.config(text=string)
		except:
			string = "Device: "
			self.lbl2.config(text=string)
			
		flag = 0
		count1 = 0
		count2 = 0
		count3 = 0
		for i in range (1,total_no_of_steps + 1):
			if step_status[i] == 'r': # if running
				self.step_lbl_list[i]["text"] = "RUNNING..."
				self.step_lbl_list[i]["bg"] = 'yellow'
				self.lbl.config(bg='yellow')
				self.lbl.config(text='RUNNING...')
				self.running=True
				self.counter_label()
			if step_status[i] == 'p': # if passed
				self.step_lbl_list[i]["text"] = result_msg[i]
				self.step_lbl_list[i]["bg"] = 'green'
				count2 = count2 + 1
			if step_status[i] == 'f': # if failed
				self.step_lbl_list[i]["text"] = "ERROR!"
				self.step_lbl_list[i]["bg"] = 'red'
				self.lbl.config(bg='red')
				self.lbl.config(text='ERROR!')
				#flag = 1
				count3 = count3 + 1
			if step_status[i] == 'ns': # if not started
				self.step_lbl_list[i]["text"] = "NOT STARTED"
				self.step_lbl_list[i]["bg"] = 'gray'
				count1 = count1 + 1
				
		if count1 == total_no_of_steps: # if none of the steps have started
			self.lbl.config(bg='gray')
			self.lbl.config(text='NOT STARTED')
			
		if count2 == total_no_of_steps: # if all the steps are done
			self.lbl.config(bg='green')
			self.lbl.config(text='DONE!')
			flag = 1
			
		if flag == 0: # if there is no error and all the steps aren't done yet
			self.root.after(200, self.manage_lblframe)
		else: # if there is an error or if all the steps are done already
			device_info['SrNo'] = ""
			return 0
#$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$
class MasterThread(threading.Thread):
	def __init__(self, guiobj):
		super(MasterThread, self).__init__()
		
	def run(self):
		global step_status
		global script_ver

		print1("FT####################################################",1)
		print1(script_ver)
		step1_routine()
		if step_status[1] == 'p' or step_status[1] == 'f': # if step 1 passed
		    step2_routine()
		print1("\n----------------------------------------------------")
		print1("Test concluded!")
## # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # 		
			
if __name__ == "__main__":
		g = GuiThread()
		g.run()

"""
	print "123"
	print("\n----------------------------------------------------")
	tkMessageBox.showinfo("Message Box", "play video test! 视频测试，播放720p视频.")
	print("\n----------------------------------------------------")
	print("Step " + " Started...")
	print("Step " + " 开始播放720p视频")
	#a = os.popen("sh os-ver.sh 720p").read()
	#subprocess.run('sh os-ver.sh 720p',stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
	#os.system("sudo setenforce 0")
	time.sleep(1)
	os.system("sh os-ver.sh 720p &")
	time.sleep(1)
	#os.system("sudo setenforce 1")
	if not tkMessageBox.askokcancel("PASS?", "视频播放器不要关-确定测试机器播放720p视频正常吗?"):
			print("Error! No Device found!")
			print(a)
			#step_status[step_num] = 'f' # failed	
			#return;	
	os.system("sh os-ver.sh suspend")	
	print "successful"
	
	
	

	os.system("sh os-ver.sh USB211 &")
	time.sleep(2)
	tkMessageBox.showinfo("Message Box", "USB2.0 test! 请插入U盘到usb2.0端口1指定位置.")
	a = os.popen("sh os-ver.sh USB21").read()
	print(a)
	a = a.replace('\n','')
	a = a.replace('\r','')
    #a = os.popen("sh os-ver.sh USB21").read()
    #print(a)
	a = a.replace('\n','')
	a = a.replace('\r','')
	a = a.strip()
	print(a)
	if a == "0":
	    print "wrong"
	else:
	    print "successful"
"""