46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
"""
|
|
Serial port reader — runs in a QThread, emits each NMEA sentence.
|
|
"""
|
|
import serial
|
|
import serial.tools.list_ports
|
|
from PyQt5.QtCore import QThread, pyqtSignal
|
|
|
|
|
|
class SerialReader(QThread):
|
|
sentence = pyqtSignal(str)
|
|
connected = pyqtSignal(bool, str)
|
|
error = pyqtSignal(str)
|
|
|
|
def __init__(self, port: str, baud: int = 4800, parent=None):
|
|
super().__init__(parent)
|
|
self.port = port
|
|
self.baud = baud
|
|
self._running = False
|
|
|
|
def run(self):
|
|
self._running = True
|
|
try:
|
|
ser = serial.Serial(self.port, self.baud, timeout=1.0)
|
|
self.connected.emit(True, self.port)
|
|
buf = ''
|
|
while self._running:
|
|
raw = ser.read(ser.in_waiting or 1)
|
|
buf += raw.decode('ascii', errors='ignore')
|
|
while '\n' in buf:
|
|
line, buf = buf.split('\n', 1)
|
|
line = line.strip()
|
|
if line.startswith('$') or line.startswith('!'):
|
|
self.sentence.emit(line)
|
|
ser.close()
|
|
except serial.SerialException as e:
|
|
self.connected.emit(False, self.port)
|
|
self.error.emit(str(e))
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
self.wait(2000)
|
|
|
|
@staticmethod
|
|
def available_ports():
|
|
return [p.device for p in serial.tools.list_ports.comports()]
|