I’m writing an app where I expect some routine tasks to be executed in background.
I want to create a class that would run in a QProcess.
Communication of QProcess with other instances would happen through QSignal interfaces.

Scenario:

• Main class starts background process of video_stream and GUI app_window.
• Main class also prints frame_cnt emitted from vide_stream QProcess.
• VideoStream listens to video_stream_request QSignal and prints string on request.
• Main class starts app_window, which has button to emit video_thread_request.
• Both VideoStream and App instantiate CommonSignals with signals used for interprocess communication (created in singleton)

Issue:
VideoStream is emmitting frame_available as expected but does not process incoming video_stream_request signal.
This can be only fixed by commenting out code where thread is being started:

self.moveToThread(self.thread)
self.thread.started.connect(self.process_video)
self.thread.start()

VideoStream process_video runs in QThread (no response to button click event):

VideoStream QThread is commented out:

Most likely way I’m trying to build the app is incorrect.
Would like to ask for any suggestion how to fix it.

Code:
main.py

import sys

from PySide6.QtWidgets import QApplication

from app import App
from common_signals import CommonSignals
from video_stream import VideoStream


class Main:
    def __init__(self) -> None:
        app = QApplication([])
        self.signals = CommonSignals()

        self.signals.frame_available.connect(self.print_frame_cnt)

        video_stream = VideoStream()
        video_stream.start_process()

        self.app_window = App()
        self.app_window.show()

        sys.exit(app.exec())

    def print_frame_cnt(self, cnt):
        print(f"frame_cnt: {cnt}")


if __name__ == "__main__":
    main = Main()

common_signals.py

from PySide6.QtCore import QObject, Signal


def singleton(cls):
    instances = {}

    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance


@singleton
class CommonSignals(QObject):
    video_stream_request = Signal()
    frame_available = Signal(object)

    def __init__(self):
        super().__init__()

video_stream.py

import time
from PySide6.QtCore import QProcess, QObject, Slot, QThread
from common_signals import CommonSignals


class VideoStream(QObject):

    def __init__(self):
        super().__init__()
        self.is_running = False

        self.process = None
        self.signals = CommonSignals()
        self.thread = QThread()
        self.frame_count = 0

        self.signals.video_stream_request.connect(self.receive_request)

    def start_process(self):
        """Starts the video capture in a separate QProcess."""
        self.process = QProcess(self)
        self.is_running = True

        # Move the processing function to the background
        self.moveToThread(self.thread)
        self.thread.started.connect(self.process_video)
        self.thread.start()

    def process_video(self):
        while self.is_running:
            self.frame_count += 1
            self.signals.frame_available.emit(self.frame_count)
            time.sleep(1)
        self.thread.quit()

    @Slot()
    def receive_request(self):
        print(f"Request to VideoStream")

app.py

from PySide6.QtWidgets import QMainWindow, QPushButton, QVBoxLayout, QWidget
from common_signals import CommonSignals


class App(QMainWindow):
    def __init__(self):
        super().__init__()

        self.setGeometry(300, 300, 400, 200)
        central_widget = QWidget(self)
        self.setCentralWidget(central_widget)
        layout = QVBoxLayout()
        central_widget.setLayout(layout)

        self.button = QPushButton("video_thread_request", self)
        layout.addWidget(self.button)

        self.signals = CommonSignals()
        self.button.clicked.connect(self.emit_signal)

    def emit_signal(self):
        """Emit the custom signal when the button is clicked."""
        print("Button clicked, emitting signal.")
        self.signals.video_thread_request.emit()