serial_communication.py
6.1 KB
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
'''
@ author: jie
@ tools: pycharm
@ content: 串口通讯实现类
@ date: 2021.5.10
'''
import imp
import serial
import serial.tools.list_ports
import logging
from config import Config
serl = None
# from serial_config import ReadConfig
# from app.induction_config import read_com
class SerialCommunication():
def __init__(self):
global serl
# serial_cf = ReadConfig()
# self.port = serial_cf.get_serial("DEFAULT_COM")
# self.bps = int(serial_cf.get_serial("DEFAULT_BAUDRATE"))
# self.timeout = float(serial_cf.get_serial("DEFAULT_TIMEOUT"))
# self.port = read_com()
# self.uploads_path = Config.UPLOAD_FOLDER
with open(Config.IP_PATH + "/serialcom.txt","r") as f:
serialcom=f.read()
self.port = serialcom.strip("\n")
self.bps = 115200
self.timeout = 1
if serl is None:
try:
# 打开串口，并得到串口对象
serl = serial.Serial(port=self.port,baudrate=self.bps,timeout=self.timeout)
# 判断是否打开成功
# if (self.main_engine.is_open):
# print ("串口已打开")
except Exception as e:
print("---异常---：", e)
self.main_engine=serl
# 打印设备基本信息
def print_name(self):
print(self.main_engine.name) #设备名字
print(self.main_engine.port)#读或者写端口
print(self.main_engine.baudrate)#波特率
print(self.main_engine.bytesize)#字节大小
print(self.main_engine.parity)#校验位
print(self.main_engine.stopbits)#停止位
print(self.main_engine.timeout)#读超时设置
print(self.main_engine.writeTimeout)#写超时
print(self.main_engine.xonxoff)#软件流控
print(self.main_engine.rtscts)#软件流控
print(self.main_engine.dsrdtr)#硬件流控
print(self.main_engine.interCharTimeout)#字符间隔超时
#打开串口
def Open_Engine(self):
self.main_engine.open()
#关闭串口
def Close_Engine(self):
self.main_engine.close()
print(self.main_engine.is_open) # 检验串口是否打开
# 打印可用串口列表
def get_used_port(self):
port_list = list(serial.tools.list_ports.comports())
# print(port_list)
return port_list
# 检查串口连接状态
def check_serial(self):
try:
if (self.main_engine.is_open):
return '串口连接正常'
else:
return '串口连接失败'
except Exception as e:
error_msg = '发生错误,{}'.format(e)
return error_msg
#接收指定大小的数据
#从串口读size个字节。如果指定超时，则可能在超时后返回较少的字节；如果没有指定超时，则会一直等到收完指定的字节数。
def read_Size(self,size):
return self.main_engine.read(size=size)
#接收一行数据
# 使用readline()时应该注意：打开串口时应该指定超时，否则如果串口没有收到新行，则会一直等待。
# 如果没有超时，readline会报异常。
def read_Line(self):
line = self.main_engine.readline().decode('utf-8').rstrip()
return line
def read_alldata(self):
if self.main_engine.in_waiting:
return self.main_engine.read_all()
def read_hex_data(self):
if self.main_engine.in_waiting:
data = self.main_engine.read_all()
# data= str(binascii.b2a_hex(t.read(num)))[2:-1] #十六进制显示方法2
#发数据
def send_data(self,data):
# text = '{"ADDR":"99","colorset:[[1,255,0,0],[2,0,255,255],[5,132,142,110]]}\n{"ADDR":"1","color":[[1,10,20,30],[2,31,37,49],[5,1,3,5,7,9,11,14,18]]}\n{"ADDR":"3","color":[[1,10,20,30],[2,31,37,49],[5,1,3,5,7,9,11,14,18]]}\n'
data = bytes(data.encode('utf-8'))
self.main_engine.write(data)
# 十六进制数据发送
def send_hex_data(self,data):
data = bytes(data.encode('utf-8')) # 先将输入的字符串转化成字节码
hexstring = data.hex()
self.main_engine.write(hexstring)
#更多示例
# self.main_engine.write(chr(0x06).encode("utf-8")) # 十六制发送一个数据
# print(self.main_engine.read().hex()) # # 十六进制的读取读一个字节
# print(self.main_engine.read())#读一个字节
# print(self.main_engine.read(10).decode("gbk"))#读十个字节
# print(self.main_engine.readline().decode("gbk"))#读一行
# print(self.main_engine.readlines())#读取多行，返回列表，必须匹配超时（timeout)使用
# print(self.main_engine.in_waiting)#获取输入缓冲区的剩余字节数
# print(self.main_engine.out_waiting)#获取输出缓冲区的字节数
# print(self.main_engine.readall())#读取全部字符。
#
def Recive_data(self,way):
# 循环接收数据，此为死循环，可用线程实现
print("开始接收数据：")
while True:
try:
# 一个字节一个字节的接收
if self.main_engine.in_waiting:
if(way == 0):
for i in range(self.main_engine.in_waiting):
print("接收ascii数据："+str(self.Read_Size(1)))
data1 = self.Read_Size(1).hex()#转为十六进制
data2 = int(data1,16)#转为十进制print("收到数据十六进制："+data1+" 收到数据十进制："+str(data2))
if(way == 1):
#整体接收
# data = self.main_engine.read(self.main_engine.in_waiting).decode("utf-8")#方式一
data = self.main_engine.read_all()#方式二print("接收ascii数据：", data)
except Exception as e:
print("异常报错：",e)
# if __name__ == '__main__':
# myser = Communication("/dev/ttyUSB0",115200,1)
# myser.print_name()
# myser.get_used_port()
# myser.delete_port()
# myser.show_port()