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

copy_left 2022-04-26

展开全文

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

使用tkinter布局界面

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

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

界面设计

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

下拉框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()