Правильный способ запуска QProcess в приложении PySide6 ⇐ Python

[Anonymous]

Я пишу приложение, в котором ожидаю, что некоторые рутинные задачи будут выполняться в фоновом режиме.
Я хочу создать класс, который будет выполняться в QProcess.
Происходит взаимодействие QProcess с другими экземплярами. через интерфейсы QSignal.
Сценарий:

[*]Основной класс запускает фоновый процесс video_stream и GUI app_window.
[*]Основной класс также печатает Frame_cnt, отправленный из QProcess vide_stream.
[*]VideoStream слушает QSignal video_stream_request и печатает строку по запросу.
[*] Основной класс запускает app_window, у которого есть кнопка для отправки video_thread_request.
[*]И VideoStream, и App создают экземпляры CommonSignals с сигналами, используемыми для межпроцессного взаимодействия (созданными в синглтоне)

Проблема:
VideoStream отправляет Frame_available, как ожидалось, но не обрабатывает входящий сигнал video_stream_request.
Это можно исправить только закомментировав код, где тема запускается:
self.moveToThread(self.thread)
self.thread.started.connect(self.process_video)
self.thread.start()

VideoStreamprocess_video выполняется в QThread (нет ответа на событие нажатия кнопки):


VideoStream QThread закомментирован:


Скорее всего, я пытаюсь создать приложение неверно.
Хотелось бы узнать, как это исправить.
Код:
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()



Подробнее здесь: https://stackoverflow.com/questions/790 ... yside6-app