1. 程式人生 > 程式設計 >Python實現串列埠通訊(pyserial)過程解析

Python實現串列埠通訊(pyserial)過程解析

pyserial模組封裝了對串列埠的訪問,相容各種平臺。

安裝

pip insatll pyserial

初始化

簡單初始化示例

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

所有引數

ser = serial.Serial(
port=None,# number of device,numbering starts at
# zero. if everything fails,the user
# can specify a device string,note
# that this isn't portable anymore
# if no port is specified an unconfigured
# an closed serial port object is created
baudrate=9600,# baud rate
bytesize=EIGHTBITS,# number of databits
parity=PARITY_NONE,# enable parity checking
stopbits=STOPBITS_ONE,# number of stopbits
timeout=None,# set a timeout value,None for waiting forever
xonxoff=0,# enable software flow control
rtscts=0,# enable RTS/CTS flow control
interCharTimeout=None  # Inter-character timeout,None to disable
)

不同平臺下初始化

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

serial.Serial類(另外初始化的方法)

class serial.Serial()
{
  def __init__(port=None,baudrate=9600,bytesize=EIGHTBITS,parity=PARITY_NONE,stopbits=STOPBITS_ONE,timeout=None,xonxoff=False,rtscts=False,writeTimeout=None,dsrdtr=False,interCharTimeout=None)
}

ser物件屬性

name:裝置名字
port:讀或者寫埠
baudrate:波特率
bytesize:位元組大小
parity:校驗位
stopbits:停止位
timeout:讀超時設定
writeTimeout:寫超時
xonxoff:軟體流控
rtscts:硬體流控
dsrdtr:硬體流控
interCharTimeout:字元間隔超時

ser物件常用方法

ser.isOpen():檢視埠是否被開啟。
ser.open() :開啟埠‘。
ser.close():關閉埠。
ser.read():從埠讀位元組資料。預設1個位元組。
ser.read_all():從埠接收全部資料。
ser.write("hello"):向埠寫資料。
ser.readline():讀一行資料。
ser.readlines():讀多行資料。
in_waiting():返回接收快取中的位元組數。
flush():等待所有資料寫出。
flushInput():丟棄接收快取中的所有資料。
flushOutput():終止當前寫操作,並丟棄傳送快取中的資料。

封裝參考

import serial
import serial.tools.list_ports

class Communication():

  #初始化
  def __init__(self,com,bps,timeout):
    self.port = com
    self.bps = bps
    self.timeout =timeout
    global Ret
    try:
      # 開啟串列埠,並得到串列埠物件
       self.main_engine= serial.Serial(self.port,self.bps,timeout=self.timeout)
      # 判斷是否開啟成功
       if (self.main_engine.is_open):
        Ret = True
    except Exception as e:
      print("---異常---:",e)

  # 列印裝置基本資訊
  def Print_Name(self):
    print(self.main_engine.name) #裝置名字
    print(self.main_engine.port)#讀或者寫埠
    print(self.main_engine.baudrate)#波特率
    print(self.main_engine.bytesize)#位元組大小
    print(self.main_engine.parity)#校驗位
    print(self.main_engine.stopbits)#停止位
    print(self.main_engine.timeout)#讀超時設定
    print(self.main_engine.writeTimeout)#寫超時
    print(self.main_engine.xonxoff)#軟體流控
    print(self.main_engine.rtscts)#軟體流控
    print(self.main_engine.dsrdtr)#硬體流控
    print(self.main_engine.interCharTimeout)#字元間隔超時

  #開啟串列埠
  def Open_Engine(self):
    self.main_engine.open()

  #關閉串列埠
  def Close_Engine(self):
    self.main_engine.close()
    print(self.main_engine.is_open) # 檢驗串列埠是否開啟

  # 列印可用串列埠列表
  @staticmethod
  def Print_Used_Com():
    port_list = list(serial.tools.list_ports.comports())
    print(port_list)





  #接收指定大小的資料
  #從串列埠讀size個位元組。如果指定超時,則可能在超時後返回較少的位元組;如果沒有指定超時,則會一直等到收完指定的位元組數。
  def Read_Size(self,size):
    return self.main_engine.read(size=size)

  #接收一行資料
  # 使用readline()時應該注意:開啟串列埠時應該指定超時,否則如果串列埠沒有收到新行,則會一直等待。
  # 如果沒有超時,readline會報異常。
  def Read_Line(self):
    return self.main_engine.readline()

  #發資料
  def Send_data(self,data):
    self.main_engine.write(data)

  #更多示例
  # self.main_engine.write(chr(0x06).encode("utf-8")) # 十六制傳送一個數據
  # print(self.main_engine.read().hex()) # # 十六進位制的讀取讀一個位元組
  # print(self.main_engine.read())#讀一個位元組
  # print(self.main_engine.read(10).decode("gbk"))#讀十個位元組
  # print(self.main_engine.readline().decode("gbk"))#讀一行
  # print(self.main_engine.readlines())#讀取多行,返回列表,必須匹配超時(timeout)使用
  # print(self.main_engine.in_waiting)#獲取輸入緩衝區的剩餘位元組數
  # print(self.main_engine.out_waiting)#獲取輸出緩衝區的位元組數
  # print(self.main_engine.readall())#讀取全部字元。

  #接收資料
  #一個整型資料佔兩個位元組
  #一個字元佔一個位元組

  def Recive_data(self,way):
    # 迴圈接收資料,此為死迴圈,可用執行緒實現
    print("開始接收資料:")
    while True:
      try:
        # 一個位元組一個位元組的接收
        if self.main_engine.in_waiting:
          if(way == 0):
            for i in range(self.main_engine.in_waiting):
              print("接收ascii資料:"+str(self.Read_Size(1)))
              data1 = self.Read_Size(1).hex()#轉為十六進位制
              data2 = int(data1,16)#轉為十進位制
              if (data2 == "exit"): # 退出標誌
                break
              else:
                 print("收到資料十六進位制:"+data1+" 收到資料十進位制:"+str(data2))
          if(way == 1):
            #整體接收
            # data = self.main_engine.read(self.main_engine.in_waiting).decode("utf-8")#方式一
            data = self.main_engine.read_all()#方式二
            if (data == "exit"): # 退出標誌
              break
            else:
               print("接收ascii資料:",data)
      except Exception as e:
        print("異常報錯:",e)


Communication.Print_Used_Com()
Ret =False #是否建立成功標誌

Engine1 = Communication("com12",115200,0.5)
if (Ret):
  Engine1.Recive_data(0)
while(1)
  {
   //傳送測試
   uint8_t a = 61;
   delayms(300);
   printf("%c",a);
}
開始接收資料:
接收ascii資料:b'='
收到資料十六進位制:3d 收到資料十進位制:61

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支援我們。