-
Notifications
You must be signed in to change notification settings - Fork 5
/
Serial.py
90 lines (67 loc) · 2.66 KB
/
Serial.py
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
from PyQt5.QtSerialPort import QSerialPort, QSerialPortInfo
from PyQt5.QtCore import QIODevice, pyqtSignal, QTextStream
from PyQt5.QtWidgets import QMessageBox
class Serial(QSerialPort):
# Signals
readyRead = pyqtSignal([str], [int])
resourcesUnavailable = pyqtSignal()
line_read = ""
def __init__(self, port_loc="/dev/ttyACM0", baud_rate=9600, parent=None):
super(Serial, self).__init__(parent)
self.parent = parent
self.setBaudRate(baud_rate)
self.setDataBits(self.Data8)
self.setParity(self.NoParity)
self.setStopBits(self.OneStop)
self.setFlowControl(self.NoFlowControl)
self.setPortName(port_loc)
# signal: ready to read, slot: format data
super(Serial, self).readyRead.connect(self.__format_data)
self.error.connect(self.onError)
def setBaudRate_str(self, baud_rate:str):
super(Serial, self).setBaudRate(int(baud_rate))
def __format_data(self):
try:
data = self.readLine().data().decode('ascii')
# sometimes empty of just \n read by `readLine`
if data != "" and data != "\n":
self.readyRead[str].emit(data)
self.readyRead[int].emit(int(data.strip()))
except (ValueError):
pass
def read(self) -> bytes:
while self.waitForReadyRead(0) is False :
pass
byte = super(Serial, self).read(1)
return byte
def write(self, data:bytes):
super(Serial, self).write(data)
self.flush()
def write_line_read(self, data:str):
if self.isOpen() is True:
for character in data:
self.write(character.encode("ascii"))
@staticmethod
def get_available_ports_systemLocations_and_manufacturers():
available_ports = QSerialPortInfo.availablePorts()
ports_info = []
for port in available_ports:
ports_info.append( (port.systemLocation(), port.manufacturer()) )
return ports_info
def onError(self, error):
if error == QSerialPort.ResourceError:
self.resourcesUnavailable.emit()
elif error == QSerialPort.PermissionError:
QMessageBox.warning(self.parent, \
"Warning", "Can not open %s\nIt may be already opened" % self.portName())
def on_new_portName(self, port):
if self.isOpen() is True:
self.close()
self.setPortName(port)
if self.open() is False:
print("Can't open port %s" % self.portName())
def open(self):
if super(Serial, self).open(QIODevice.ReadWrite) is True:
return True
else:
return False