這個(gè)屬于我項(xiàng)目中一個(gè)函數(shù)，跟大家分享一下我的思路及最終實(shí)現(xiàn)

在編寫(xiě)串口通信工具中，需要實(shí)現(xiàn)一個(gè)函數(shù)，自動(dòng)找到對(duì)應(yīng)com 口，并且連接該com口，保證后續(xù)通信正常作為初始化過(guò)程的一部分。

思路

在win 設(shè)備管理器中，經(jīng)常會(huì)出現(xiàn)多個(gè)com 口，但并不是每個(gè)com 口都是目標(biāo)設(shè)備所鏈接的。嘗試打開(kāi)每個(gè)com 口，輸入enter 按鍵， 正確的com 口，會(huì)有ack log 返回，表明通信 正常

否則，沒(méi)有任何log 返回，則判斷為非目標(biāo)設(shè)備所連接的com 口。

實(shí)現(xiàn)

嘗試去打開(kāi)所有com 口，然后發(fā)送enter， 如果在一段時(shí)間內(nèi)有返回值，檢查com 口收到的字節(jié)數(shù)，如果非零，則表明找到了對(duì)應(yīng)的com 口。

完整測(cè)試代碼如下：

import serialimport serial.tools.list_portsimport threadingimport binasciiimport timefrom datetime import datetime# default valuebaunRate = 115200is_read = Falseis_write = Falsewrite_buff = []sys_buff = []mSerial = Nonecallback = Noneis_opened = 0is_registed = 0class SerialPort: def __init__(self,port,buand): self.port = serial.Serial(port,buand) self.port.close() if not self.port.isOpen(): self.port.open() #the index of data_bytes for read operation,私有屬性 #only used in read lines self.__read_ptr = 0 self.__read_head = 0 #store all read bytes # used in read date， read lines self.__data_bytes = bytearray() def port_open(self): if not self.port.isOpen(): self.port.open() def port_close(self): self.port.close() def send(self): global is_write global write_buff while is_write: if len(write_buff):msg = write_buff.pop(0)msg = msg+'n'cmd = msg.encode()try: self.port.write(cmd)except: write_buff.clear() is_write = False write_buff.clear() def read_data(self): global is_read global is_opened byte_cnt = 0 while is_read: try:count = self.port.inWaiting()if count > 0: rec_str = self.port.read(count) self.__data_bytes = self.__data_bytes+rec_str#print('receive:',rec_str.decode()) #print(rec_str) byte_cnt += count if not is_opened: is_opened = 1 #print('累計(jì)收到：',byte_cnt) #time.sleep(0.5)self.read_lines() except:deinit() #將當(dāng)前所有的數(shù)據(jù)都讀出，讀取位置不變，每次讀取指針依次移動(dòng)，不漏數(shù)據(jù)， 讀取行為一直在進(jìn)行 def read_lines(self): #reset line_cnt = 0 data_len = len(self.__data_bytes) #print ('') #print ('begin: prt=:', self.__read_ptr, ' head =', self.__read_head,'current len =',data_len) if self.__read_ptr >=data_len: return #get all lines in current data_bytes while(self.__read_ptr < data_len-1): if(self.__data_bytes[self.__read_ptr+1] == 0x0a and self.__data_bytes[self.__read_ptr] == 0x0d):tmp = bytearray()tmp = self.__data_bytes[self.__read_head:self.__read_ptr]try: line = tmp.decode()except: self.__read_head = self.__read_ptr + 2 self.__read_ptr = self.__read_head continueiprint(line)line_cnt += 1self.__read_head = self.__read_ptr + 2self.__read_ptr = self.__read_head else:self.__read_ptr = self.__read_ptr + 1def auto_open_serial(): global baunRate global mSerial global callback global is_registed global is_opened #reset deinit() # 列出所有當(dāng)前的com口 port_list = list(serial.tools.list_ports.comports()) port_list_name = [] #get all com if len(port_list) <= 0: iprint('the serial port can’t find!') return False else: for itms in port_list: port_list_name.append(itms.device) #try open #print(port_list_name) for i in port_list_name: try: mSerial = SerialPort(i,baunRate) iprint('try open %s'%i) start_task() send('') #return True time.sleep(1) if is_opened:iprint('connect %s successfully'%i)return True else:deinit()if i == port_list_name[len(port_list_name)-1]: iprint('uart don’t open') breakcontinue except: iprint(' uart don’t open') deinit() return Falsedef deinit(): global mSerial global is_write global is_read global write_buff global is_opened if mSerial: mSerial.port_close() is_opened = 0 is_read = False is_write = False write_buff = [] mSerial = None time.sleep(1)def init(): global mSerial global callback global is_registed global is_opened global is_read global is_write #retry retry_time = 0 while not auto_open_serial(): if not is_opened: iprint('wait for uart connect, retry %s'%str(retry_time)) else: return True retry_time += 1 time.sleep(2) if retry_time == 10: iprint(' open uart fail') return Falsedef send(msg): global mSerial global is_write global write_buff if is_write: write_buff.append(msg)def start_task(): global mSerial global is_write global is_read if mSerial: is_write = True t1 = threading.Thread (target=mSerial.send) t1.setDaemon (False) t1.start () is_read = True t2 = threading.Thread (target=mSerial.read_data) t2.setDaemon (False) t2.start ()def iprint(msg): global callback global is_registed msg = '[Uart] '+str(msg) if is_registed: callback.append(msg) else: print(msg)def start_sys_cmd(): global is_registed if is_registed: t3 = threading.Thread (target=process_receive_sys_cmd) t3.setDaemon (False) t3.start()def process_receive_sys_cmd(): global sys_buff global is_registed global callback #print('process_receive_sys_cmd') while is_registed: #print ('wait,process_receive_sys_cmd') if len(sys_buff): #print ('receive,process_receive_sys_cmd') line = sys_buff.pop(0) if 'init' in line:if is_opened and is_read and is_write: iprint('already open uart') breakiprint('start init')init() if is_opened: break iprint('Eixt uart sys thread')def register_cback(list): global callback global is_registed callback = list is_registed = 1def unregister_cback(): global callback callback.clear()if __name__ == ’__main__’: receive = [] register_cback(receive) sys_buff.append('init') start_sys_cmd() def process_receive_msg(): global receive while True: #print('wait') if len(receive):#print('receive')print(receive.pop(0)) t = threading.Thread(target=process_receive_msg) t.setDaemon(False) t.start()

到此這篇關(guān)于python 自動(dòng)識(shí)別并連接串口的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)python 自動(dòng)識(shí)別并連接串口內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！