通过两个进程分别读写串口，并把发送与接收到的内容记录在blog中，收到q时程序结束并退出
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
import threading,time
import serial
import string
class serthread:
def __init__(self, port=0):
#初始化串口、blog文件名称
self.my_serial = serial.serial()
self.my_serial.port=port
self.my_serial.baudrate = 9600
self.my_serial.timeout = 1 self.alive = false
self.waitend = none
fname=time.strftime("%y%m%d")#blog名称为当前时间
self.rfname='r'+fname #接收blog名称
self.sfname='s'+fname #发送blog名称
self.thread_read= none
self.thread_send=none
def waiting(self):
# 等待event停止标志
if not self.waitend is none:
self.waitend.wait()
def start(self):
#开串口以及blog文件
self.rfile=open(self.rfname,'w')
self.sfile=open(self.sfname,'w')
self.my_serial.open()
if self.my_serial.isopen():
self.waitend = threading.event()
self.alive = true
self.thread_read = threading.thread(target=self.reader)
self.thread_read.setdaemon(true)
self.thread_send=threading.thread(target=self.sender)
self.thread_send.setdaemon(true)
self.thread_read.start()
self.thread_send.start()
return true
else:
return false
def reader(self):
while self.alive:
try:
n=self.my_serial.inwaiting()
data=''
if n:
data= self.my_serial.read(n).decode('utf-8')
print ('recv'+' '+time.strftime("%y-%m-%d %x")+' '+data.strip())
print (time.strftime("%y-%m-%d %x:")+data.strip(),file=self.rfile)
if len(data)==1 and ord(data[len(data)-1])==113: #收到字母q，程序退出
break
except exception as ex:
print (ex)
self.waitend.set()
self.alive = false
def sender(self):
while self.alive:
try:
snddata=input("input data:\n")
self.my_serial.write(snddata.encode('utf-8'))
print ('sent'+' '+ time.strftime("%y-%m-%d %x"))
print (snddata,file=self.sfile)
except exception as ex:
print (ex)
self.waitend.set()
self.alive = false
def stop(self):
self.alive = false
#self.thread_read.join()
#self.thread_send.join()
if self.my_serial.isopen():
self.my_serial.close()
self.rfile.close()
self.sfile.close()
if __name__ == '__main__':
ser = serthread('com4')
try:
if ser.start():
ser.waiting()
ser.stop()
else:
pass;
except exception as ex:
print (ex)
if ser.alive:
ser.stop()
print ('end ok .');
del ser;
|
以上这篇python3 实现串口两进程同时读写就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持服务器之家。
原文链接：https://blog.csdn.net/colcloud/article/details/42454839