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

[Anonymous]

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

• Main класс запускает фоновый процесс video_stream и GUI app_window.
• Основной класс также печатает Frame_cnt, созданный из vide_stream QProcess.
  VideoStream слушает video_stream_request QSignal и печатает строку по запросу.
• Основной класс запускает app_window, в котором есть кнопка для отправки video_thread_request< /code>.
• И VideoStream, и приложение создают экземпляры 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 закомментирован:


Скорее всего, я пытаюсь создать приложение неверно.
Хотел бы узнать, как это исправить.
Код:
import sys
import time

from PySide6.QtCore import QProcess, QObject, Slot, QThread
from PySide6.QtCore import Signal
from PySide6.QtWidgets import QApplication
from PySide6.QtWidgets import QMainWindow, QPushButton, QVBoxLayout, QWidget

def singleton(cls):
"""Singleton decorator for use on CommonSignals class."""
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):
"""QSignals class for communicating between instances."""
video_stream_request = Signal()
frame_available = Signal(object)

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

class VideoStream(QObject):

def __init__(self):
"""Class with routine task in background that expects to receive outside requests time by time."""
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")

class App(QMainWindow):
"""GUI window that throws event to VideoStream via click of the button."""
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_stream_request.emit()

class Main:
"""Main class."""
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()



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