您好,登录后才能下订单哦!
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
这篇文章将为大家详细讲解有关python3中Serial串口助手如何接收读取数据,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以有所收获。
创建串口助手首先需要创建一个类,重构类的实现过程如下:
#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串口助手如何接收读取数据”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,使各位可以学到更多知识,如果觉得文章不错,请把它分享出去让更多的人看到。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。