Python进阶：一个串口通信工具（tkinter pyserial openpyxl）

纸上得来终觉浅，唯有实践出真知。这篇文章我们用Python来写一个串口通信的模拟器，使用到的技术包括tkinter、pySerial、openpyxl和python多线程。

使用tkinter布局界面

tkinter是python自带的GUI工具包。开发界面虽然有点丑，但是不复杂的界面用起来还是比较简单方便的。

下面介绍一下界面布局及界面组件的功能。

界面设计

1. 串口设置区域，用下拉框展示系统串口号、波特率、数据位、停止位、校验位供用户选择。
2. 打开串口和关闭串口按钮。
3. 信息展示区域，展示串口打开关闭、异常信息和发送及接收到的串口数据。
4. 选择串口指令文件按钮，弹出选择文件对话框，选择指令模板excel文件。
5. 展示当前选择的指令文件路径。
6. 指令展示区。分指令描述和指令内容两列展示指令，单击指令描述直接发送选择的指令，双击指令内容则在下方指令详情区域展示选择的指令，可编辑选择的指令。
7. 指令描述输入框。
8. 指令内容对话框。
9. 添加指令按钮。点击后将指令添加到上方指令展示区，此时并不保存到文件。
10. 修改当前选择的指令。
11. 将指令展示区的指令保存到指令模板文件。
12. 发送指令到当前打开的串口。
13. 清空按钮。清空信息展示区内容。

下拉框Combobox

以串口号下拉框为例，介绍一下tk中Combobox的使用方法。

#串口号下拉框#保存选中的串口号self.com_choose=StringVar()#下拉框self.com_choose_combo=ttk.Combobox(self.com_choose_frame,width=30,textvariable=self.com_choose)#设置为只读，不允许修改self.com_choose_combo['state']='readonly'self.com_choose_combo.grid(row=0,column=1,columnspan=2,sticky=W+E,padx=2)#下拉框的值通过com_name_get()方法获取self.com_choose_combo['values']=self.com_name_get()

使用pyserial获取系统串口：

def com_name_get(self):        self.port_list=list(serial.tools.list_ports.comports())        self.com_port_names=[]        if len(self.port_list)>0:            for i in range(len(self.port_list)):                self.com_name=str(self.port_list[i])                self.com_port_names.append(self.com_name)        return self.com_port_names

选择文件

按钮绑定点击事件，command=self.thread_file()。在thread_file函数中我们新建一个线程来打开选择文件窗口，这样主界面线程不会卡住。

self.file_choose_button=Button(self.init_window_name,text='选择文件',bg='lightblue',width=10,command=self.thread_file)

#新建选择文件线程    def thread_file(self):        thisthread=threading.Thread(target=self.file_choose)        thisthread.start()    #选择文件打开，并在界面中显示    def file_choose(self):        self.root=Tk()        self.root.withdraw()        file_path=filedialog.askopenfilename()        if file_path:            self.open_file(file_path)

展示选择的文件路径，Entry组件文本处理方法如下：

self.file_path_text.delete(0,END)self.file_path_text.insert(END,file_path)

excel文件解析

使用openpyxl处理excel文件。

wb=openpyxl.load_workbook(file_path)sheet=wb[wb.sheetnames[0]]#然后从sheet中循环获取数据就可以了

指令展示区点击事件

指令展示使用Treeview组件，分别针对不同的列绑定不同的鼠标事件。

#鼠标左键双击self.code_tree.bind('<Double-1>', self.set_cell_value)#鼠标左键单击self.code_tree.bind('<1>', self.cell_operate)

处理左键单击事件，需要获取当前点击的行和列，如果是第一列column=='#1'，则发送对应的指令到串口。

# 发送按钮执行def cell_operate(self,event):    # 列    column= self.code_tree.identify_column(event.x)     # 行    row = self.code_tree.identify_row(event.y)     # 点击指令描述列发送指令    if column=='#1' and self.code_tree.get_children():        data = self.code_tree.item(row, 'values')[1]        self.writeSerial(data)

双击指令内容，则将当前指令信息填充到下方指令编辑区。

# 双击进入编辑状态 def set_cell_value(self,event): # 列 column= self.code_tree.identify_column(event.x) # 行 row = self.code_tree.identify_row(event.y) self.selected_command_row=row if (column=='#2') and self.code_tree.get_children(): self.command_name.delete(0,END) self.command_name.insert(0,self.code_tree.item(row, 'values')[0]) self.command.delete(0,END) self.command.insert(0,self.code_tree.item(row, 'values')[1])

串口操作

使用pyserial操作串口。

打开串口代码如下，根据用户选择的串口号以及设置信息打开串口。

self.ser=serial.Serial(port=self.ser_name, baudrate=self.ser_baudrate, bytesize=self.ser_bytesize, parity=self.ser_parity, stopbits=self.ser_stopbits)self.ser.timeout=0.01

打开串口成功后，需要启动一个新的线程来监听串口通信，循环获取串口收到的数据。

# 开启接收串口数据线程self.ReadUARTThread = threading.Thread(target=self.ReadUART, daemon=True)self.ReadUARTThread.start()

def ReadUART(self):        try:            while self.connected:                newline=self.ser.readline()#字节类型                if newline:                    # print(newline)                    self.result_text.insert(END,f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}←{newline.decode('gbk')}\n')                    self.result_text.see(tkinter.END)                    self.result_text.update()        except:            # print(e)            newline=f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}:串口{self.ser_name}已关闭\n'            self.result_text.insert(END,newline)            self.result_text.see(tkinter.END)            self.result_text.update()

串口发送数据很简单，直接使用Serial的write函数就可以了。

self.ser.write(data.encode('gbk'))

总结

代码可以使用工具打包成exe后使用更方便。这个工具很简单，但涉及到的知识点也不少，初学Python的朋友可以看看。

运行效果

附所有代码

干货分享，代码全部奉上。

requirements.txt，python版本为3.8

pyserial~=3.5openpyxl~=3.0.7pyinstaller~=4.2

simulator.py

import tkinterfrom tkinter import ttk,filedialog,scrolledtextfrom tkinter import *import openpyxlimport threadingimport timefrom datetime import datetimeimport serialimport serial.tools.list_portsimport reimport osclass MY_GUI(): #构造函数 def __init__(self,name): self.init_window_name=name self.connected=False #窗口控件设置初始化 def set_init_window(self): self.init_window_name.title('指令模拟器(串口)') self.init_window_name.geometry('1168x620+20+10') self.init_window_name.resizable(False, False) self.init_window_name.attributes('-alpha',1) #串口选择框架 self.com_choose_frame=Frame(self.init_window_name,width=10,height=100) self.com_choose_frame.place(x=20,y=12) #串口号标签 self.com_label=Label(self.com_choose_frame,text='串口号: ') self.com_label.grid(row=0,column=0,sticky=W) #串口号下拉框 self.com_choose=StringVar() self.com_choose_combo=ttk.Combobox(self.com_choose_frame,width=30,textvariable=self.com_choose) self.com_choose_combo['state']='readonly' self.com_choose_combo.grid(row=0,column=1,columnspan=2,sticky=W+E,padx=2) self.com_choose_combo['values']=self.com_name_get() # 波特率标签 self.baudrate_label=Label(self.com_choose_frame,text='波特率: ') self.baudrate_label.grid(row=0,column=3,sticky=W,padx=6) #波特率选项框 self.baudrate_value=StringVar(value='115200') self.baudrate_choose_combo=ttk.Combobox(self.com_choose_frame,width=6,textvariable=self.baudrate_value) self.baudrate_choose_combo['values']=('115200','9600') self.baudrate_choose_combo['state']='readonly' self.baudrate_choose_combo.grid(row=0,column=4,sticky=W,padx=2) # 数据位标签 self.bytesize=Label(self.com_choose_frame,text='数据位: ') self.bytesize.grid(row=0,column=5,sticky=W,padx=6) #数据位选项框 self.bytesize_value=StringVar(value='8') self.bytesize_choose_combo=ttk.Combobox(self.com_choose_frame,width=6,textvariable=self.bytesize_value) self.bytesize_choose_combo['values']=('5','6','7','8') self.bytesize_choose_combo['state']='readonly' self.bytesize_choose_combo.grid(row=0,column=6,padx=2) # 停止位标签 self.stopbits=Label(self.com_choose_frame,text='停止位: ') self.stopbits.grid(row=1,column=3,sticky=W,padx=6) # 停止位选项框 self.stopbits_value=StringVar(value='1') self.stopbits_choose_combo=ttk.Combobox(self.com_choose_frame,width=6,textvariable=self.stopbits_value) self.stopbits_choose_combo['values']=('1','1.5','2') self.stopbits_choose_combo['state']='readonly' self.stopbits_choose_combo.grid(row=1,column=4,padx=2) # 校验位标签 self.parity=Label(self.com_choose_frame,text='校验位: ') self.parity.grid(row=1,column=5,sticky=W,padx=6) # 校验位选项框 self.parity_value=StringVar(value='None') self.parity_choose_combo=ttk.Combobox(self.com_choose_frame,width=6,textvariable=self.parity_value) self.parity_choose_combo['values']=('None','Odd','Even','Mark','Space') self.parity_choose_combo['state']='readonly' self.parity_choose_combo.grid(row=1,column=6,padx=2,pady=2) #串口框架内部按钮 self.connect_button=Button(self.com_choose_frame,text='打开串口',bg='lightblue',width=30,command=self.com_connect) self.connect_button.grid(row=1,column=1,columnspan=2,padx=1,pady=5) #文件路径文本框 self.file_path_text=Entry(self.init_window_name,width=57) self.file_path_text.place(x=20,y=95) #选择文件按钮 self.file_choose_button=Button(self.init_window_name,text='选择文件',bg='lightblue',width=10,command=self.thread_file) self.file_choose_button.place(x=450,y=90) #代码解析后进行显示 self.code_frame=Frame(self.init_window_name,width=78,height=29,bg='white') self.code_frame.place(x=20,y=130) #解析后的代码放在表格内显示 self.code_tree=ttk.Treeview(self.code_frame,show='headings',height=16,columns=('0','1')) self.code_bar=ttk.Scrollbar(self.code_frame,orient=VERTICAL,command=self.code_tree.yview) self.code_tree.configure(yscrollcommand=self.code_bar.set) self.code_tree.grid(row=0,column=0,sticky=NSEW) self.code_bar.grid(row=0,column=1,sticky=NS) self.code_tree.column('0',width=100,anchor='center') self.code_tree.column('1',width=450) self.code_tree.heading('0',text='指令描述') self.code_tree.heading('1',text='指令') # 双击左键进入编辑 self.code_tree.bind('<Double-1>', self.set_cell_value) self.code_tree.bind('<1>', self.cell_operate) # 默认打开command.xlsx self.codeline_counter=0 if os.path.exists('command.xlsx'): self.open_file(os.path.abspath('command.xlsx')) elif os.path.exists('command.xls'): self.open_file(os.path.abspath('command.xls')) #文件路径文本框 self.command_name_label=Label(self.init_window_name,text='指令描述： ') self.command_name_label.place(x=20,y=500) self.command_name=Entry(self.init_window_name,width=68) self.command_name.place(x=90,y=500) self.command=Entry(self.init_window_name,width=78) self.command.place(x=20,y=535) self.command_add_button=Button(self.init_window_name,text='添加',bg='lightblue',width=10,command=self.command_add) self.command_add_button.place(x=40,y=565) self.command_update_button=Button(self.init_window_name,text='修改',bg='lightblue',width=10,command=self.command_update) self.command_update_button.place(x=140,y=565) self.command_save_button=Button(self.init_window_name,text='保存到文件',bg='lightblue',width=10,command=self.command_save) self.command_save_button.place(x=240,y=565) self.command_send=Button(self.init_window_name,text='发送',bg='lightblue',width=10,command=self.com_send) self.command_send.place(x=480,y=565) #处理结果显示滚动文本框 self.result_data_label=Label(self.init_window_name,text='输出结果') self.result_data_label.place(x=600,y=15) self.clear_button=Button(self.init_window_name,text='清空',bg='lightblue',width=10,command=self.clear_result_text) self.clear_button.place(x=680,y=10) self.result_text=scrolledtext.ScrolledText(self.init_window_name,width=77,height=42) self.result_text.place(x=600,y=50) #自动获取当前连接的串口名 def com_name_get(self): self.port_list=list(serial.tools.list_ports.comports()) self.com_port_names=[] if len(self.port_list)>0: for i in range(len(self.port_list)): self.com_name=str(self.port_list[i]) self.com_port_names.append(self.com_name) return self.com_port_names #打开串口按键的执行内容 def com_connect(self): if self.connected: self.com_cancel() return self.ser_name=str(self.com_choose.get()) for i in range(len(self.port_list)): if self.ser_name == str(self.port_list[i]): self.ser_name, desc, hwid = self.port_list[i] self.result_text.insert(END,f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}:准备连接串口{self.ser_name}\n') self.ser_baudrate=int(self.baudrate_value.get()) self.ser_bytesize=int(self.bytesize_value.get()) self.ser_stopbits=float(self.stopbits_value.get()) self.ser_parity=str(self.parity_value.get())[0:1] try: self.ser=serial.Serial(port=self.ser_name, baudrate=self.ser_baudrate, bytesize=self.ser_bytesize, parity=self.ser_parity, stopbits=self.ser_stopbits) self.ser.timeout=0.01 self.result_text.insert(END,f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}:串口{self.ser_name}打开成功\n') self.result_text.see(tkinter.END) self.result_text.update() # 按钮变成“关闭串口” self.connected=True self.connect_button['text']='关闭串口' # 开启接收串口数据线程 self.ReadUARTThread = threading.Thread(target=self.ReadUART, daemon=True) self.ReadUARTThread.start() except: newline=f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}:串口{self.ser_name}打开失败,串口不存在或被占用\n' self.result_text.insert(END,newline) self.result_text.see(tkinter.END) self.result_text.update() #关闭串口按键的执行内容 def com_cancel(self): try: self.ser.close() # 按钮变成“打开串口” self.connected=False self.connect_button['text']='打开串口' except: newline=time.ctime(time.time())+':'+'串口未打开'+'\n' self.result_text.insert(END,newline) self.result_text.see(tkinter.END) self.result_text.update() def clear_result_text(self): self.result_text.delete('1.0',END) def ReadUART(self): try: while self.connected: newline=self.ser.readline()#字节类型 if newline: # print(newline) self.result_text.insert(END,f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}←{newline.decode('gbk')}\n') self.result_text.see(tkinter.END) self.result_text.update() except: # print(e) newline=f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}:串口{self.ser_name}已关闭\n' self.result_text.insert(END,newline) self.result_text.see(tkinter.END) self.result_text.update() def com_send(self): data = self.command.get() if data: self.writeSerial(data) def writeSerial(self, data): try: if self.connected and self.ser: # print(data) self.ser.write(data.encode('gbk')) newline=f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}→{data}\n' self.result_text.insert(END,newline) self.result_text.see(tkinter.END) self.result_text.update() except: newline=f'{datetime.now().strftime('%H:%M:%S.%f')[:-3]}:串口{self.ser_name}发送数据失败\n' self.result_text.insert(END,newline) self.result_text.see(tkinter.END) self.result_text.update() #新建选择文件线程 def thread_file(self): thisthread=threading.Thread(target=self.file_choose) thisthread.start() #选择文件打开，并在界面中显示 def file_choose(self): self.root=Tk() self.root.withdraw() file_path=filedialog.askopenfilename() if file_path: self.open_file(file_path) def open_file(self, file_path): self.file_path_text.delete(0,END) self.file_path_text.insert(END,file_path) wb=openpyxl.load_workbook(file_path) sheet=wb[wb.sheetnames[0]] # 只支持最大200行 self.code_sheet=[[0 for i in range(3)]for j in range(200)] if self.code_tree.get_children(): for item in self.code_tree.get_children(): self.code_tree.delete(item) pattern=re.compile(r'_x00(.*?)_',re.S) for i in range(200): if sheet.cell(row=i+2,column=1).value: self.codeline_counter +=1 self.code_context=[] # 指令描述 self.code_sheet[i][0]=sheet.cell(row=i+2,column=1).value self.code_context.append(self.code_sheet[i][0]) # 指令 self.code_sheet[i][1] = sheet.cell(row=i+2,column=2).value special_char=re.findall(pattern,self.code_sheet[i][1]) for c in special_char: self.code_sheet[i][1]=self.code_sheet[i][1].replace('_x00'+c+'_', bytes.fromhex(c).decode('utf-8')) self.code_context.append(self.code_sheet[i][1]) # 双击进入编辑状态 def set_cell_value(self,event): # 列 column= self.code_tree.identify_column(event.x) # 行 row = self.code_tree.identify_row(event.y) self.selected_command_row=row if (column=='#2') and self.code_tree.get_children(): self.command_name.delete(0,END) self.command_name.insert(0,self.code_tree.item(row, 'values')[0]) self.command.delete(0,END) self.command.insert(0,self.code_tree.item(row, 'values')[1]) # 发送按钮执行 def cell_operate(self,event): column= self.code_tree.identify_column(event.x)# 列 row = self.code_tree.identify_row(event.y) # 行 # 点击指令描述列发送指令 if column=='#1' and self.code_tree.get_children(): data = self.code_tree.item(row, 'values')[1] self.writeSerial(data) # 修改指令执行 def command_update(self): if self.selected_command_row and self.command_name.get() and self.command.get(): self.code_tree.set(self.selected_command_row, column='#1', value=self.command_name.get()) self.code_tree.set(self.selected_command_row, column='#2', value=self.command.get()) # 添加指令执行 def command_add(self): row = len(self.code_tree.get_children()) if self.command_name.get() and self.command.get(): self.code_tree.insert('', row, values=[self.command_name.get(), self.command.get()]) print(self.command_name.get()) # 保存到文件指令执行 def command_save(self): row = len(self.code_tree.get_children()) file_path = self.file_path_text.get() if row>0 and file_path: try: wb=openpyxl.load_workbook(file_path) sheet=wb[wb.sheetnames[0]] i=2 pattern=re.compile(r'([\x00-\x20])',re.S) for item in self.code_tree.get_children(): command_val=self.code_tree.item(item, 'values')[1] special_char=re.findall(pattern,command_val) for c in special_char: hexstr = hex(ord(c)) if len(hexstr) == 3: command_val = command_val.replace(c,'_x00'+hexstr.replace('x','')+'_') elif len(hexstr) == 4: command_val = command_val.replace(c,'_x00'+hexstr.replace('0x','')+'_') # print(command_val) sheet.cell(row=i,column=1).value = self.code_tree.item(item, 'values')[0] sheet.cell(row=i,column=2).value = command_val i+=1 wb.save(file_path) tkinter.messagebox.showinfo('保存成功',f'保存文件成功{file_path}') except PermissionError as e: tkinter.messagebox.showinfo('保存失败',e) except: tkinter.messagebox.showinfo('保存失败','描述或指令含有特殊字符？') #主线程def start(): init_window=Tk() my_window=MY_GUI(init_window) my_window.set_init_window() init_window.mainloop()start()

指令模板excel格式如下：

描述	指令
参数设置	EP#BSTMODE=T
设备状态	SQ
结束复位	EP#RESET