This link explains that QThread can be used in one of two ways:

1. A QThread instance that runs a default event loop and you can dispatch events to be executed by the thread using signals and a worker class (with a method as a slot).
2. A QThread subclass that reimplements run() and executes custom code but cannot receive events.

Well, I want something that does both. I have a SerialThread that will check a serial port periodically and receive data whenever it is available, but the main thread might also need to send some data and since two serial objects can’t use the same physical serial port, the main thread has to send its data through SerialThread.

I already have a basic system that achieves this dual functionality.

class SerialThread(QThread):
    
    class Worker(QObject):
        def __init__(self, n: int):
            super().__init__()
            self.n = n
        
        @pyqtSlot()
        def work(self):
            print(f"Exec from worker {self.n}...")
            print("Work Thread ID:", format(threading.get_ident(), '04X'))
    
    
    sig_new_data = pyqtSignal(list)
    _sig_work = pyqtSignal()
    
    _worker = Worker(1)
    
    _serial = serial.Serial()
    
    def __init__(self):
        super().__init__()
        self._worker.moveToThread(self)
        self._sig_work.connect(self._worker.work)
    
    def run(self):
        print("Serial Thread ID:", format(threading.get_ident(), '04X'))
        
        self.__open_com_port()
        
        while True:
            self.eventDispatcher().processEvents(QEventLoop.AllEvents)    # executes pending events
            
            if not self._serial.is_open and not self.__open_com_port():
                self.msleep(2000)
                continue
            
            if self.__get_data():
                self.sig_new_data.emit()

    def queue_work(self):    # called from the main thread
        self._sig_work.emit()   # Adds a pending event to the SerialThread event loop.

The problem is, I want to potentially have multiple types of workers that do different things or a Worker class that also takes in a callable. That means I have to create a new Worker instance every time some work is requested and connect _sig_work to it, emit the signal and then disconnect it like so:

def queue_work(thread: SerialThread):
    worker = SerialThread.Worker()
    worker.moveToThread(thread)
    thread._sig_work.connect(worker.work)
    thread._sig_work.emit() 
    thread._sig_work.disconnect(worker.work)

However, this only seems to works when the Worker instance is created inside a class or as a global variable (e.g. it doesn’t work when it’s instantiated in a function or a method). Maybe this has something to do with the garbage collector.

Even if I manage to get that working, this doesn’t seem like a very clean way to do this. What would be the easier/better way to do this or not do this at all (somehow solve the multiple serial object problem)?