python3 Serial 串口助手的接收讀取數據


  其實網上已經有許多python語言書寫的串口,但大部分都是python2寫的,沒有找到一個合適的python編寫的串口助手,只能自己來寫一個串口助手,由於我只需要串口能夠接收讀取數據就可以了,故而這個串口助手只實現了數據的接收讀取。

  創建串口助手首先需要創建一個類,重構類的實現過程如下:

 1 #coding=gb18030
 2 
 3 import threading
 4 import time
 5 import serial
 6 
 7 class ComThread:
 8     def __init__(self, Port='COM3'):
 9     #構造串口的屬性
10         self.l_serial = None
11         self.alive = False
12         self.waitEnd = None
13         self.port = Port
14         self.ID = None
15         self.data = None
16    #定義串口等待的函數
17     def waiting(self):
18         if not self.waitEnd is None:
19             self.waitEnd.wait()
20 
21     def SetStopEvent(self):
22         if not self.waitEnd is None:
23             self.waitEnd.set()
24         self.alive = False
25         self.stop()
26     #啟動串口的函數
27     def start(self):
28         self.l_serial = serial.Serial()
29         self.l_serial.port = self.port
30         self.l_serial.baudrate = 115200
31         #設置等待時間,若超出這停止等待
32         self.l_serial.timeout = 2
33         self.l_serial.open()
34         #判斷串口是否已經打開
35         if self.l_serial.isOpen():
36             self.waitEnd = threading.Event()
37             self.alive = True
38             self.thread_read = None
39             self.thread_read = threading.Thread(target=self.FirstReader)
40             self.thread_read.setDaemon(1)
41             self.thread_read.start()
42             return True
43         else:
44             return False

  創建好類后,就要實現串口讀取的功能,其讀取數據的函數如下:

  首先要創建一個字符串來存放接收到的數據:

            data = ''
            data = data.encode('utf-8')#由於串口使用的是字節,故而要進行轉碼,否則串口會不識別

 

   在創建好變量來接收數據后就要開始接收數據:

1             n = self.l_serial.inWaiting()#獲取接收到的數據長度
2             if n: 
3                  #讀取數據並將數據存入data
4                  data = data + self.l_serial.read(n)
5                  #輸出接收到的數據
6                  print('get data from serial port:', data)
7                  #顯示data的類型,便於如果出錯時檢查錯誤
8                  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*

1
80*
<class 'str'> <class 'str'>

 

 

串口助手完整代碼如下:

  1 #coding=gb18030
  2 
  3 import threading
  4 import time
  5 import serial
  6 
  7 class ComThread:
  8     def __init__(self, Port='COM3'):
  9         self.l_serial = None
 10         self.alive = False
 11         self.waitEnd = None
 12         self.port = Port
 13         self.ID = None
 14         self.data = None
 15 
 16     def waiting(self):
 17         if not self.waitEnd is None:
 18             self.waitEnd.wait()
 19 
 20     def SetStopEvent(self):
 21         if not self.waitEnd is None:
 22             self.waitEnd.set()
 23         self.alive = False
 24         self.stop()
 25 
 26     def start(self):
 27         self.l_serial = serial.Serial()
 28         self.l_serial.port = self.port
 29         self.l_serial.baudrate = 115200
 30         self.l_serial.timeout = 2
 31         self.l_serial.open()
 32         if self.l_serial.isOpen():
 33             self.waitEnd = threading.Event()
 34             self.alive = True
 35             self.thread_read = None
 36             self.thread_read = threading.Thread(target=self.FirstReader)
 37             self.thread_read.setDaemon(1)
 38             self.thread_read.start()
 39             return True
 40         else:
 41             return False
 42 
 43     def SendDate(self,i_msg,send):
 44         lmsg = ''
 45         isOK = False
 46         if isinstance(i_msg):
 47             lmsg = i_msg.encode('gb18030')
 48         else:
 49             lmsg = i_msg
 50         try:
 51             # 發送數據到相應的處理組件
 52             self.l_serial.write(send)
 53         except Exception as ex:
 54             pass;
 55         return isOK
 56 
 57     def FirstReader(self):
 58         while self.alive:
 59             time.sleep(0.1)
 60 
 61             data = ''
 62             data = data.encode('utf-8')
 63 
 64             n = self.l_serial.inWaiting()
 65             if n:
 66                  data = data + self.l_serial.read(n)
 67                  print('get data from serial port:', data)
 68                  print(type(data))
 69 
 70             n = self.l_serial.inWaiting()
 71             if len(data)>0 and n==0:
 72                 try:
 73                     temp = data.decode('gb18030')
 74                     print(type(temp))
 75                     print(temp)
 76                     car,temp = str(temp).split("\n",1)
 77                     print(car,temp)
 78 
 79                     string = str(temp).strip().split(":")[1]
 80                     str_ID,str_data = str(string).split("*",1)
 81 
 82                     print(str_ID)
 83                     print(str_data)
 84                     print(type(str_ID),type(str_data))
 85 
 86                     if str_data[-1]== '*':
 87                         break
 88                     else:
 89                         print(str_data[-1])
 90                         print('str_data[-1]!=*')
 91                 except:
 92                     print("讀卡錯誤,請重試!\n")
 93 
 94         self.ID = str_ID
 95         self.data = str_data[0:-1]
 96         self.waitEnd.set()
 97         self.alive = False
 98 
 99     def stop(self):
100         self.alive = False
101         self.thread_read.join()
102         if self.l_serial.isOpen():
103             self.l_serial.close()
104 #調用串口,測試串口
105 def main():
106     rt = ComThread()
107     rt.sendport = '**1*80*'
108     try:
109         if  rt.start():
110             print(rt.l_serial.name)
111             rt.waiting()
112             print("The data is:%s,The Id is:%s"%(rt.data,rt.ID))
113             rt.stop()
114         else:
115             pass
116     except Exception as se:
117         print(str(se))
118 
119     if rt.alive:
120         rt.stop()
121 
122     print('')
123     print ('End OK .')
124     temp_ID=rt.ID
125     temp_data=rt.data
126     del rt
127     return temp_ID,temp_data
128 
129 
130 if __name__ == '__main__':
131 
132     #設置一個主函數,用來運行窗口,便於若其他地方下需要調用串口是可以直接調用main函數
133     ID,data = main()
134 
135     print("******")
136     print(ID,data)

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM