其实网上已经有许多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)
时间： 2024-11-09 07:58:27