pyqt_signal

serial_thread.py

from PyQt5.QtCore import QThread, pyqtSignal
from struct import unpack
from serial import Serial

class SerialWorker(QThread):
    # update_data = pyqtSignal(list)
    raw_data = pyqtSignal(bytes)
    
    def __init__(self):
        super(SerialWorker, self).__init__()
        self.is_treaming = False

    def run(self):
        # print('serial read begin')
        while True:
            if self.is_treaming:
                try:
                    dataRaw = self.serial.read(7)
                    if (len(dataRaw) > 1):
                        print(dataRaw)
                        data = unpack('BBBBBBB', dataRaw) # return 
                        print(data)
                    
                    if int(len(dataRaw)) > 0:
                        self.raw_data.emit(dataRaw)
                        data = unpack('BBBBBBB', dataRaw) # return a tuple
                        # self.update_data.emit(list(data)) # emit signal
                except:
                    pass
    
    def connect(self, port, baud):
        #connect serial thread into device via serial com
        self.port = port
        self.baud = baud
        self.serial = Serial(self.port, self.baud, timeout=0.1) # init serial
        self.is_treaming = True
        # print('connected')

    def write(self, data):
        if self.serial == None: return
        if self.serial.is_open:
            self.serial.write(data)
            print(f'serial sent: {data}')

    def disconnect(self):       
        self.serial.close()
        self.is_treaming = False
        # print('disconnected')
    
    def quit(self) -> None:
        self.quit()