UI interface:
Implementation Code
# -*- coding: utf-8 -*-
# Form implementation generated from reading ui file ''
#
# Created by: PyQt5 UI code generator 5.15.2
#
# WARNING: Any manual changes made to this file will be lost when pyuic5 is
# run again. Do not edit this file unless you know what you are doing.
import sys
from PyQt5 import *
from import *
from PyQt5 import QtWidgets
from PyQt5 import QtGui
from PyQt5 import QtCore
import logging
import os
import json
import configparser
import psutil
#pip install psutil
#import serial
import subprocess
import io
#import glob
#import re# regular expressions
#pip install pyserial
class AnalysisThread():
test_result_signal = (tuple) # New signal to pass a tuple with device paths and test results
finished_signal = () # Add a new signal to indicate thread completion
def __init__(self,devicepath,type,parent=None):
super(AnalysisThread, self).__init__(parent)
= devicepath
= type
self.is_running = True
def run(self):
result = False
index=0
if ().rnetTestCount>0:
index=().rnetTestCount-1
result = ().GetTestPortInfoAndTest()
print('resutl:',result)
if result:
print("Before removal:", ().rnetPortName)
().()
print("After removal:", ().rnetPortName)
device_info = []
device_info.append()# Connection status
device_info.append(().rnConnetcion[index])# Connection status
device_info.append(().rnetRate[index])# rate
device_info.append(().rnetMacAddress[index])#MacAddress
().TableAddItem(device_info,True)# Add PASS data
# of ports
().lbl_GetPortNum.setText(str(().rnetTestCount))# Set the number of test ports
().lbl_GetPortNum.setStyleSheet("background-color: lightgreen;") # Post port background color set to light green
# Connection status
().lbl_GetConntion.setText(().rnConnetcion[index])#Set the connection status
().lbl_GetPortNum.setStyleSheet("background-color: lightgreen;") # Post port background color set to light green
# rate
().lbl_GetRate.setText(().rnetRate[index]+'M')
().lbl_GetRate.setStyleSheet("background-color: lightgreen;") # Post port background color set to light green
else:
if ().rnetFailInfo:
device_info = []
device_info.append(().rnetFailInfo[0])
device_info.append(().rnetFailInfo[1])
device_info.append(().rnetFailInfo[2])
device_info.append(().rnetFailInfo[3])
().TableAddItem(device_info, False) # Add PASS data
self.test_result_signal.emit((device_info,False))
# of ports
().lbl_GetPortNum.setText(str(().rnetTestCount+1)) # Set the number of test ports
().lbl_GetPortNum.setStyleSheet("background-color: red;") # Post port background color set to light green
# Connection Status
().lbl_GetConntion.setText(().rnetFailInfo[1]) # Set connection status
().lbl_GetConntion.setStyleSheet("background-color: red;") # Post port background color set to light green
# Rate
().lbl_GetRate.setText(().rnetFailInfo[2] + 'M')
().lbl_GetRate.setStyleSheet("background-color: red;") # Post port background color set to light green
().lbl_Result.setStyleSheet("background-color: black; color: red;")
().lbl_Result.setText("Test FAIL!!!")
self.finished_signal.emit() # Signal when the thread is done
()
def stop(self):
self.is_running = False # Or use safer stopping logic
class Ui_Form(QWidget):
updateTimer=pyqtSignal(bool)#TimeThreadStarter
def __init__(self):
super().__init__()
self.finished_threads = 0 # of threads used to track completion
self.analysis_threads = [] # Store all instances of AnalysisThread
=''#Error messages
= () # Create objects
= '' # Project name
= [] # Test parameter information
= '' # Test standard parameters
# Parameter information
=1# of NIC test ports
=[]# NIC rate
# NIC information read
=0# of NIC test ports
=[]# Get all network interface names
=[]# NIC rate
=[]#NIC MACAddress
=[]# Connection status
=[]# Fail messages read
# Generate log messages
= ('my_logger') # Step 1 Create the logger
() # Step 2 Specify the log level
self.file_handler = ('./log/') # Step 3 Create the file processor
= ('%(asctime)s - %(levelname)s - %(message)s') # Step 4 Create the formatter
self.file_handler.setFormatter() # Step 4 Add the formatter to the processor
(self.file_handler) # Step 5 Add the processor to the logger
# Read the configuration
('./Conf/', encoding='utf-8') # Read the configuration file and create it if it doesn't exist
# Read the test configuration
= ('TestItemNameArrays', 'NetOpening') # Project name
= int(('TestItemWinFrmSleepExit', 'NetOpening')) # Project Test Fail delayed exit
= ('./Conf/') # Read test parameter information
if ()==True:# Get standard parameters
testArgs=[]# Test parameters
testArgs=('|')
testCount=testArgs[0]# of test ports
netRate=testArgs[1]# Test rate
=int(testCount[('=')+1:])
=netRate[('=')+1:].split(',')#Intercept rate information
()# Initialization
self.lbl_Result.setStyleSheet("background-color: black; color: red;")
=self.get_network_interfaces()
for rn in :
if 'lo' in rn:
(rn)
break
if len()!=:# Get test interface information
self.lbl_GetPortNum.setText(str(len()))
self.lbl_GetPortNum.setStyleSheet("color: red;")
self.lbl_Result.setStyleSheet("background-color: black; color: red;")
(f"Detecting a mismatch between the number of test network adapters and parameters Err:{}", False)
=f"Detecting a mismatch between the number of test network adapters and parameters Err:{}";
# Create a timer to check if the video has finished playing
= QTimer(self)
(1000) # Checked every second
(self.check_Test)
()
# Connect signal to slot
()
# Get all network interface names
def get_network_interfaces(self):
return list(psutil.net_if_addrs().keys())
def get_net_interfaces(self):
net_if_addrs = psutil.net_if_addrs()
return net_if_addrs
# Get port information and test interface_name: Adaptor Configurator Name
def GetTestPortInfoAndTest(self,interface_name):
try:
()# Clear the fail message
testdata=self.get_interface_details(interface_name)
self.label_2.setText(f'be facing (us){interface_name} port testing..')
if is None:
self.lbl_Result.setStyleSheet("background-color: black; color: lightyellow;")
self.lbl_Result.setText("Test in progress...")
else:
self.lbl_Result.setStyleSheet("background-color: black; color: red;")
self.lbl_Result.setText()
if testdata and testdata['speed'] and testdata['operstate'] and testdata['mac_address']:
if int([])==int(testdata['speed']):
(testdata['mac_address'])
(testdata['speed'])
(testdata['operstate'])
+=1
return True
else:
(interface_name)
(testdata['operstate'])
(testdata['speed'])
(testdata['mac_address'])
return False
else:
return False
except Exception as e:
(f"GetTestPortInfoAndTest Err:{e}", False)
return False
def read_sys_file(self,path):
try:
with open(path, 'r') as file:
return ().strip()
except IOError:
return None
def get_interface_details(self,interface):
speed = self.read_sys_file(f'/sys/class/net/{interface}/speed')
operstate = self.read_sys_file(f'/sys/class/net/{interface}/operstate')
mac_address = self.read_sys_file(f'/sys/class/net/{interface}/address')
return {
'speed': speed,
'operstate': operstate,
'mac_address': mac_address
}
# Define triggers
def handleTimer(self, start):
if start:
()
else:
()
# Start multiple threads
def start_analysis_threads(self):
# Create multiple threads
self.finished_threads = 0 # Reset counters
self.analysis_threads = [] # Reset the thread list
print("Remaining devices to test:", ) # Print a list of remaining devices
# = []
for portname in :
thread=AnalysisThread(portname,'lan',self)
thread.test_result_signal.connect(self.handle_test_result) # Connection signals
thread.finished_signal.connect(self.thread_finished)
()
#(thread)
self.analysis_threads.append(thread)
# Determine if all threads have completed execution
def thread_finished(self):
self.finished_threads += 1
if self.finished_threads == len(self.analysis_threads):
(True) # Restart the timer when all threads are complete
#
def handle_test_result(self, result):
devicepath, isPass = result
if isPass:
# If the test passes, remove the device from the to-be-tested list
if devicepath in :
(devicepath)
(devicepath, isPass)
# Verify that the test is complete
def check_Test(self):
if ==:
macinfo=''#mac information
rate=''# rate
for mac in :
macinfo+=mac+','
macinfo=macinfo[:-1] # Remove the last character
for rte in :
rate+=rte+','
rate=rate[:-1]# Remove the last character
(,f'PortNum={}|MAC={macinfo}|Rate={rate}','PASS')
self.lbl_Result.setStyleSheet("background-color: black; color: green;")
self.lbl_Result.setText("Test PASS!!!")
(0)
elif self.finished_threads < len(self.analysis_threads):
# Wait for all threads to finish
pass
else:
(False) # Restart the timer
self.start_analysis_threads()
def setupUi(self):
("Form")
(1006, 800)
icon = ()
(("IMAGE/"), , )
(icon)
= (self)
("gridLayout")
= ()
("verticalLayout")
= ()
("horizontalLayout")
self.horizontalLayout_2 = ()
self.horizontalLayout_2.setObjectName("horizontalLayout_2")
self.verticalLayout_2 = ()
self.verticalLayout_2.setObjectName("verticalLayout_2")
self.horizontalLayout_6 = ()
self.horizontalLayout_6.setObjectName("horizontalLayout_6")
self.lbl_Logo = (self)
self.lbl_Logo.setText("")
self.lbl_Logo.setPixmap(("IMAGE/"))
self.lbl_Logo.setAlignment()
self.lbl_Logo.setObjectName("lbl_Logo")
self.horizontalLayout_6.addWidget(self.lbl_Logo)
self.label_3 = (self)
font = ()
(12)
self.label_3.setFont(font)
self.label_3.setStyleSheet("color: rgb(255, 255, 255);\n"
"background-color: rgb(170, 170, 127);")
self.label_3.setAlignment()
self.label_3.setObjectName("label_3")
self.horizontalLayout_6.addWidget(self.label_3)
self.verticalLayout_2.addLayout(self.horizontalLayout_6)
self.label_2 = (self)
font = ()
(14)
self.label_2.setFont(font)
self.label_2.setStyleSheet("background-color: rgb(85, 255, 127);")
self.label_2.setAlignment()
self.label_2.setObjectName("label_2")
self.verticalLayout_2.addWidget(self.label_2)
self.horizontalLayout_2.addLayout(self.verticalLayout_2)
(self.horizontalLayout_2)
self.verticalLayout_3 = ()
self.verticalLayout_3.setObjectName("verticalLayout_3")
self.horizontalLayout_3 = ()
self.horizontalLayout_3.setObjectName("horizontalLayout_3")
self.label_7 = (self)
font = ()
(12)
self.label_7.setFont(font)
self.label_7.setStyleSheet("background-color: rgb(170, 170, 127);\n"
"color: rgb(255, 255, 255);")
self.label_7.setAlignment()
self.label_7.setObjectName("label_7")
self.horizontalLayout_3.addWidget(self.label_7)
self.label_8 = (self)
font = ()
(12)
self.label_8.setFont(font)
self.label_8.setStyleSheet("background-color: rgb(85, 170, 127);\n"
"color: rgb(255, 255, 255);")
self.label_8.setAlignment()
self.label_8.setObjectName("label_8")
self.horizontalLayout_3.addWidget(self.label_8)
self.label_6 = (self)
font = ()
(12)
self.label_6.setFont(font)
self.label_6.setStyleSheet("background-color: rgb(170, 255, 127);\n"
"color: rgb(255, 255, 255);")
self.label_6.setAlignment()
self.label_6.setObjectName("label_6")
self.horizontalLayout_3.addWidget(self.label_6)
self.label_5 = (self)
font = ()
(12)
self.label_5.setFont(font)
self.label_5.setStyleSheet("background-color: rgb(255, 170, 0);\n"
"color: rgb(255, 255, 255);")
self.label_5.setAlignment()
self.label_5.setObjectName("label_5")
self.horizontalLayout_3.addWidget(self.label_5)
self.verticalLayout_3.addLayout(self.horizontalLayout_3)
self.horizontalLayout_4 = ()
self.horizontalLayout_4.setObjectName("horizontalLayout_4")
self.label_10 = (self)
font = ()
(12)
self.label_10.setFont(font)
self.label_10.setAlignment()
self.label_10.setObjectName("label_10")
self.horizontalLayout_4.addWidget(self.label_10)
self.lbl_SetPortNum = (self)
font = ()
(12)
self.lbl_SetPortNum.setFont(font)
self.lbl_SetPortNum.setAlignment()
self.lbl_SetPortNum.setObjectName("lbl_SetPortNum")
self.horizontalLayout_4.addWidget(self.lbl_SetPortNum)
self.lbl_SetConntion = (self)
font = ()
(12)
self.lbl_SetConntion.setFont(font)
self.lbl_SetConntion.setAlignment()
self.lbl_SetConntion.setObjectName("lbl_SetConntion")
self.horizontalLayout_4.addWidget(self.lbl_SetConntion)
self.lbl_SetRate = (self)
font = ()
(12)
self.lbl_SetRate.setFont(font)
self.lbl_SetRate.setAlignment()
self.lbl_SetRate.setObjectName("lbl_SetRate")
self.horizontalLayout_4.addWidget(self.lbl_SetRate)
self.verticalLayout_3.addLayout(self.horizontalLayout_4)
self.horizontalLayout_5 = ()
self.horizontalLayout_5.setObjectName("horizontalLayout_5")
self.label_12 = (self)
font = ()
(12)
self.label_12.setFont(font)
self.label_12.setAlignment()
self.label_12.setObjectName("label_12")
self.horizontalLayout_5.addWidget(self.label_12)
self.lbl_GetPortNum = (self)
font = ()
(12)
self.lbl_GetPortNum.setFont(font)
self.lbl_GetPortNum.setAlignment()
self.lbl_GetPortNum.setObjectName("lbl_GetPortNum")
self.horizontalLayout_5.addWidget(self.lbl_GetPortNum)
self.lbl_GetConntion = (self)
font = ()
(12)
self.lbl_GetConntion.setFont(font)
self.lbl_GetConntion.setAlignment()
self.lbl_GetConntion.setObjectName("lbl_GetConntion")
self.horizontalLayout_5.addWidget(self.lbl_GetConntion)
self.lbl_GetRate = (self)
font = ()
(12)
self.lbl_GetRate.setFont(font)
self.lbl_GetRate.setAlignment()
self.lbl_GetRate.setObjectName("lbl_GetRate")
self.horizontalLayout_5.addWidget(self.lbl_GetRate)
self.verticalLayout_3.addLayout(self.horizontalLayout_5)
(self.verticalLayout_3)
()
= (self)
font = ()
(12)
(font)
("tableWidget")
(6)
(0) # Initialize table rows to 0
()
item = ()
()
font = ()
(12)
(font)
((170, 170, 127))
(0, item)
item = ()
()
font = ()
(12)
(font)
((0, 170, 127))
(1, item)
item = ()
()
font = ()
(12)
(font)
((170, 255, 127))
(2, item)
item = ()
()
font = ()
(12)
(font)
((255, 170, 127))
(3, item)
item = ()
()
font = ()
(12)
(font)
((85, 255, 127))
(4, item)
item = ()
()
font = ()
(12)
(font)
((85, 255, 0))
(5, item)
()
self.lbl_Result = (self)
font = ()
(18)
self.lbl_Result.setFont(font)
self.lbl_Result.setStyleSheet("background-color: rgb(0, 0, 0);")
self.lbl_Result.setAlignment()
self.lbl_Result.setText("Test in progress...")
self.lbl_Result.setObjectName("lbl_Result")
(self.lbl_Result)
(, 0, 0, 1, 1)
self.lbl_Result.setStyleSheet("background-color: black; color: red;")
()
(self)
( | ) # Show only the minimize button and close button
def retranslateUi(self):
_translate =
(_translate("Form", "LAN Test"))
self.label_3.setText(_translate("Form", "Control process"))
self.label_2.setText(_translate("Form", "Current Realtek PCIe GbE Family Controller test..."))
self.label_7.setText(_translate("Form", "Test parameter name"))
self.label_8.setText(_translate("Form", "Number of ports"))
self.label_6.setText(_translate("Form", "Connection Status"))
self.label_5.setText(_translate("Form", "Test rate"))
self.label_10.setText(_translate("Form", "Setting parameters"))
self.lbl_SetPortNum.setText(_translate("Form", str()))
self.lbl_SetConntion.setText(_translate("Form", "True"))
self.lbl_SetRate.setText(_translate("Form", f'{str([0])}M'))
self.label_12.setText(_translate("Form", "Current test status"))
self.lbl_GetPortNum.setText(_translate("Form", "N/A"))
self.lbl_GetConntion.setText(_translate("Form", "N/A"))
self.lbl_GetRate.setText(_translate("Form", "N/A"))
item = (0)
(_translate("Form", "NO"))
item = (1)
(_translate("Form", "Project name"))
item = (2)
(_translate("Form", "Connection Status"))
item = (3)
(_translate("Form", "Rate."))
item = (4)
(_translate("Form", "MacAddress"))
item = (5)
(_translate("Form", "Test results"))
self.lbl_Result.setText(_translate("COMLPTTest", "Test in progress..."))
(0,60)
(1,140)
(2,140)
(3,140)
(4,280)
(5,200)
() # Form status read-only
() # form, click event
# Read json information
def ReadJsonInfo(self,fileName):
try:
if (fileName):
f=open(fileName,'r',encoding='utf-8')
return (())
except Exception as e:
("Read "+fileName+" Err:" + str(e),False)
(1)
# Read the item parameter information, itemName: item name.
def ReadJsonTestArgs(self, itemName):
try:
= ('./Conf/')
for js in :
if itemName in js['ItemName']:
= js['Standard']
return True
('Read ItemName:' + itemName + ' Info Is Empty!!', False)
(1)
except Exception as e:
("Read ItemName:" + itemName + " Info Err:" + str(e), False)
(1)
# Manually close the window
def closeEvent(self, event):
# Create a message box with two buttons: "Yes" and "No".
#reply = (self, 'quit', "Are you sure you want to quit?" , | , )
# If the user clicks "Yes", execute () to close the window.
#if reply == :
# ()
# (1)
#else:
# If the user clicks "No", the event is ignored and the window is not closed.
# ()
(1)
# Define a function that centers the function window.
def Center(self):
# Get screen size
screen_geometry = ().availableGeometry()
# Calculate window centering
x = (screen_geometry.width() - ()) // 2
y = (screen_geometry.height() - ()) // 2
# Set the window position
(x, y)
# Information printed
def ShowLog(self, log, isPass):
try:
if isPass == True:
self.lbl_Result.setStyleSheet("color:green;")
(str(log))
self.lbl_Result.setText("TEST PASS")
else:
self.lbl_Result.setStyleSheet("color:red;")
(str(log))
self.lbl_Result.setText(log)
except Exception as e:
print("\033[1;31m" + str(e) + " \033[0m")
(1)
# update test parameters json, itemName: item name, readValue: read value, testResult: test result
def UpdateJsonTestArgs(self, itemName, readValue, testResult):
try:
updateTestArgs = [] # Updated test parameters
= ('./Conf/')
for js in :
if itemName in js['ItemName']:
js['Read'] = readValue # The value read
js['TestResult'] = testResult # Test results
(js)
else:
(js)
with open("./Conf/", "w") as write_file:
(updateTestArgs, write_file)
return True
except Exception as e:
("Read ItemName:" + itemName + " Info Err:" + str(e), False)
(1)
# def TableAddItem(self, port, isPass):
# # Check if the device is already in the table
# for rowIndex in range(()):
# print('port',port)
# if (rowIndex, 1) and (rowIndex, 2):
# if (rowIndex, 1).text() == port[0] and (rowIndex, 2).text() == port[1] and (rowIndex,3).text()==port[2]:
# # Update test results for existing lines
# testResult = 'PASS' if isPass else 'FAIL'
# (rowIndex, 4).setText(testResult)
# if testResult == 'PASS':
# (rowIndex, 4).setForeground((51, 204, 51))
# else:
# (rowIndex, 4).setForeground((255, 51, 51))
# return
def TableAddItem(self, port, isPass):
# Check if the device is already in the table
rowToUpdate = None
print('()',())
for rowIndex in range(()):
if (rowIndex, 1) and (rowIndex, 2):
if (rowIndex, 1).text() == port[0] and (rowIndex, 2).text() == \
port[1]:
rowToUpdate = rowIndex
break
if rowToUpdate is not None:
# Update existing rows
testResult = 'PASS' if isPass else 'FAIL'
(rowToUpdate, 5).setText(testResult)
if testResult == 'PASS':
(rowToUpdate, 5).setForeground((51, 204, 51))
else:
(rowToUpdate, 5).setForeground((255, 51, 51))
else:
# Add new line
newRow = ()
(newRow)
(newRow + 1) # Increase the number of table rows
# ... Adding data to a new row ...
# Add data to new row
(newRow, 0, QTableWidgetItem(str(newRow + 1)))
(newRow, 1, QTableWidgetItem(port[0]))
(newRow, 2, QTableWidgetItem(port[1]))
(newRow, 3, QTableWidgetItem(port[2]))
(newRow, 4, QTableWidgetItem(port[3]))
testResult = 'PASS' if isPass else 'FAIL'
resultItem = QTableWidgetItem(testResult)
()
(newRow, 5, resultItem)
if testResult == 'PASS':
((51, 204, 51))
else:
((255, 51, 51))
#Start the line re-scripting
def TestThread(self):
pass
#self.t_autoplay=Thread(target=)
#self.t_autoplay.start()
if __name__=='__main__':
app=QApplication()
win=Ui_Form()
() # Centered
()
(app.exec_())To this point this article on Python + PyQt5 to achieve network port function test details of the article is introduced to this, more related to Python network port function test content, please search for my previous articles or continue to browse the following related articles I hope that you will support me in the future more!