python使用Serial模塊讀取串口數據

前言

　　pyserial 模塊封裝了對串口的訪問，兼容各種平台。

 

使用

模塊安裝

pip install pyserial

 

初始化

import serial
ser = serial.Serial('com1', 9600, timeout=1)

ser = serial.Serial("/dev/ttyUSB0", 9600, timeout=0.5) # 使用USB連接串行口
ser = serial.Serial("/dev/ttyAMA0", 9600, timeout=0.5) # 使用樹莓派的GPIO口連接串行口
ser = serial.Serial(1, 9600, timeout=0.5)# winsows系統使用com1口連接串行口
ser = serial.Serial("com1", 9600, timeout=0.5)# winsows系統使用com1口連接串行口
ser = serial.Serial("/dev/ttyS1", 9600, timeout=0.5)# Linux系統使用com1口連接串行口

 

對象屬性

• name——設備名字
• port——讀或者寫端口
• baudrate——波特率
• bytesize——字節大小
• parity——校驗位
• stopbits——停止位
• timeout——讀超時設置
• writeTimeout——寫超時
• xonxoff——軟件流控
• rtscts——硬件流控
• dsrdtr——硬件流控
• interCharTimeout——字符間隔超時

 

對象常用方法

• ser.isOpen()——查看端口是否被打開
• ser.open() ——打開端口
• ser.close()——關閉端口
• ser.read()——從端口讀字節數據。默認1個字節
• ser.read_all()——從端口接收全部數據
• ser.write("hello")——向端口寫數據
• ser.readline()——讀一行數據
• ser.readlines()——讀多行數據
• in_waiting()——返回接收緩存中的字節數
• flush()——等待所有數據寫出
• flushInput()——丟棄接收緩存中的所有數據
• flushOutput()——終止當前寫操作，並丟棄發送緩存中的數據。

 

實例

#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)