Python3 Serial串口助手数据接收与解析实现详解

青禾

发布时间：2026-10-13| 4人浏览过

原创

本文详细介绍如何使用Python3的Serial库编写串口助手，重点解决数据接收与解析问题。通过创建ComThread类并采用多线程机制，实现串口数据的实时读取。针对特定下位机协议（+ID+*Data+*格式），演示了从字节流到字符型的转换、数据分割及ID与数据提取的具体代码实现，为嵌入式开发中的数据交互提供参考。

Python3串口助手开发背景与目标

尽管网络上存在许多用Python编写的串口工具，但多数基于Python2版本，难以直接满足当前需求。因此，有必要自行开发一个轻量级的串口助手。鉴于核心需求仅涉及数据的接收与读取，本方案专注于实现这一功能，省略了发送等其他复杂模块。

串口通信类初始化与线程管理

构建串口助手的第一步是定义一个处理类，该类的初始化及基础结构配置如下所示：


#coding=gb18030

import threading
import time
import serial

class ComThread:
 def __init__(self, Port='COM3'):
 #构造串口的属性
  self.l_serial = None
  self.alive = False
  self.waitEnd = None
  self.port = Port
  self.ID = None
  self.data = None
 #定义串口等待的函数
 def waiting(self):
  if not self.waitEnd is None:
   self.waitEnd.wait()

 def SetStopEvent(self):
  if not self.waitEnd is None:
   self.waitEnd.set()
  self.alive = False
  self.stop()
 #启动串口的函数
 def start(self):
  self.l_serial = serial.Serial()
  self.l_serial.port = self.port
  self.l_serial.baudrate = 115200
  #设置等待时间，若超出这停止等待
  self.l_serial.timeout = 2
  self.l_serial.open()
  #判断串口是否已经打开
  if self.l_serial.isOpen():
   self.waitEnd = threading.Event()
   self.alive = True
   self.thread_read = None
   self.thread_read = threading.Thread(target=self.FirstReader)
   self.thread_read.setDaemon(1)
   self.thread_read.start()
   return True
  else:
   return False

类定义完成后，需实现核心的串口读取逻辑。该功能通过特定的函数调用完成，具体代码实现如下：

串口数据接收逻辑实现

在开始读取之前，需初始化一个字符串变量用于暂存接收到的原始数据：


   data = ''
   data = data.encode('utf-8')#由于串口使用的是字节，故而要进行转码，否则串口会不识别

变量初始化后，程序进入数据接收循环。具体操作包括获取缓冲区数据长度并读取内容：


n = self.l_serial.inWaiting()#获取接收到的数据长度
   if n: 
     #读取数据并将数据存入data
     data = data + self.l_serial.read(n)
     #输出接收到的数据
     print('get data from serial port:', data)
     #显示data的类型，便于如果出错时检查错误
     print(type(data))

特定协议数据解析与提取

数据接收结束后，需对原始数据进行清洗和解析以提取有效信息。由于不同下位机采用的通信协议各异，解析方法也需相应调整。本文示例中使用的协议格式为“+ID+*Data+*”，具体的提取逻辑代码如下：


#获取还没接收到的数据长度
n = self.l_serial.inWaiting()
#判断是否已经将下位机传输过来的数据全部提取完毕，防止之前没有获取全部数据
if len(data)>0 and n==0:
 try:
  #将数据转换为字符型
  temp = data.decode('gb18030')
  #输出temp类型，看转码是否成功
  print(type(temp))
  #输出接收到的数据
  print(temp)
  将数据按换行分割并输出
  car,temp = str(temp).split("\n",1)
  print(car,temp)
 
  #将temp按':'分割，并获取第二个数据
  string = str(temp).strip().split(":")[1]
  #由于前面分割后的数据类型是列表，因此需要转换成字符串，而后按照'*'分割，得到的也就是我们需要的Id和data
  str_ID,str_data = str(string).split("*",1)
 
  print(str_ID)
  print(str_data)
  print(type(str_ID),type(str_data))
  #判断data最后一位是否是'*'，若是则退出，若不是则输出异常
  if str_data[-1]== '*':
   break
  else:
   print(str_data[-1])
   print('str_data[-1]!=*')
 except:
  print("读卡错误，请重试！\n")

执行上述代码后的典型输出结果如下所示：


get data from serial port:b'ID:4A622E29\n\xbf\xa8\xd6\xd0\xbf\xe924\xca\xfd\xbe\xdd\xce\xaa:1*80*\r\n'
<class 'bytes'>
<class 'str'>
ID:4A622E29
卡中块24数据为:1*80*

ID:4A622E29 卡中块24数据为:1*80*
80*
<class 'str'> <class 'str'>

完整代码示例与运行结果

以下是整合了串口初始化、多线程读取及数据解析功能的完整串口助手代码：


#coding=gb18030

import threading
import time
import serial

class ComThread:
 def __init__(self, Port='COM3'):
  self.l_serial = None
  self.alive = False
  self.waitEnd = None
  self.port = Port
  self.ID = None
  self.data = None

 def waiting(self):
  if not self.waitEnd is None:
   self.waitEnd.wait()

 def SetStopEvent(self):
  if not self.waitEnd is None:
   self.waitEnd.set()
  self.alive = False
  self.stop()

 def start(self):
  self.l_serial = serial.Serial()
  self.l_serial.port = self.port
  self.l_serial.baudrate = 115200
  self.l_serial.timeout = 2
  self.l_serial.open()
  if self.l_serial.isOpen():
   self.waitEnd = threading.Event()
   self.alive = True
   self.thread_read = None
   self.thread_read = threading.Thread(target=self.FirstReader)
   self.thread_read.setDaemon(1)
   self.thread_read.start()
   return True
  else:
   return False

 def SendDate(self,i_msg,send):
  lmsg = ''
  isOK = False
  if isinstance(i_msg):
   lmsg = i_msg.encode('gb18030')
  else:
   lmsg = i_msg
  try:
   # 发送数据到相应的处理组件
   self.l_serial.write(send)
  except Exception as ex:
   pass;
  return isOK

 def FirstReader(self):
  while self.alive:
   time.sleep(0.1)

   data = ''
   data = data.encode('utf-8')

   n = self.l_serial.inWaiting()
   if n:
     data = data + self.l_serial.read(n)
     print('get data from serial port:', data)
     print(type(data))

   n = self.l_serial.inWaiting()
   if len(data)>0 and n==0:
    try:
     temp = data.decode('gb18030')
     print(type(temp))
     print(temp)
     car,temp = str(temp).split("\n",1)
     print(car,temp)

     string = str(temp).strip().split(":")[1]
     str_ID,str_data = str(string).split("*",1)

     print(str_ID)
     print(str_data)
     print(type(str_ID),type(str_data))

     if str_data[-1]== '*':
      break
     else:
      print(str_data[-1])
      print('str_data[-1]!=*')
    except:
     print("读卡错误，请重试！\n")

  self.ID = str_ID
  self.data = str_data[0:-1]
  self.waitEnd.set()
  self.alive = False

 def stop(self):
  self.alive = False
  self.thread_read.join()
  if self.l_serial.isOpen():
   self.l_serial.close()
#调用串口，测试串口
def main():
 rt = ComThread()
 rt.sendport = '**1*80*'
 try:
  if rt.start():
   print(rt.l_serial.name)
   rt.waiting()
   print("The data is:%s,The Id is:%s"%(rt.data,rt.ID))
   rt.stop()
  else:
   pass
 except Exception as se:
  print(str(se))

 if rt.alive:
  rt.stop()

 print('')
 print ('End OK .')
 temp_ID=rt.ID
 temp_data=rt.data
 del rt
 return temp_ID,temp_data


if __name__ == '__main__':

 #设置一个主函数，用来运行窗口，便于若其他地方下需要调用串口是可以直接调用main函数
 ID,data = main()

 print("******")
 print(ID,data)

以上即为基于Python3 Serial库实现串口数据接收与解析的完整方法。希望该示例能为需要进行串口通信开发的读者提供有价值的参考，欢迎进一步交流探讨。