Асинхронная связь между двумя потоками QThreads ⇐ Python

[Anonymous]

Я работаю над оптимизацией приложения PyQt (для управления машиной для испытаний на растяжение), которое иногда работает медленно. Приложение использует два QThreads, оба работают в цикле с time.sleep внутри.

• Один QThread (AnalogReaderThread) непрерывно считывает аналоговые значения, выполняет вычисления и управляет двигателями. .
• Другой QThread (VideoThread) захватывает видеокадры и отслеживает маркеры. Координаты этих маркеров необходимо отправить в первый поток.

Типичное время цикла составляет около 0,1 секунды. Эти два QThreads работают асинхронно на разных частотах, и я думаю, как лучше организовать связь между ними.
У меня есть два потенциальных решения:
< ol>
[*]VideoThread отправляет сигнал (через механизм сигнала/слота) каждый цикл с координатами маркера, которые хранятся во временных переменных в первом потоке и доступны при необходимости.
[*]VideoThread имеет метод получения, который возвращает последние известные координаты маркера.

Я привел два минимальных примера, описывающих мою реализацию. Пожалуйста, дайте мне знать ваши идеи:
Сигналы/слоты, используемые для асинхронной связи:
import sys
import time
import random
import numpy as np
from PyQt5.QtCore import QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QLabel, QMainWindow, QVBoxLayout, QWidget

class AnalogReaderThread(QThread):
update_analog_data = pyqtSignal(float, float)

def __init__(self, thread):
super().__init__()
self._thread = thread
self._thread.update_image_data.connect(self.update_image_data)
self._mean_brightness = 0
def run(self):
while True:
analog_data = self.read_analog_data()
self.update_analog_data.emit(analog_data, self._mean_brightness)
time.sleep(1)

def read_analog_data(self):
return random.uniform(0, 100)

def update_image_data(self, mean_brightness):
self._mean_brightness = mean_brightness

class VideoThread(QThread):
update_image_data = pyqtSignal(float)

def run(self):
while True:
image_data = self.read_image_data()
mean_brightness = self.process_image(image_data)
self.update_image_data.emit(mean_brightness)
time.sleep(0.5)

def read_image_data(self):
# Generate a random image (100x100 pixels, 3 color channels)
image = np.random.randint(0, 256, (100, 100, 3), dtype=np.uint8)
return image

def process_image(self, image_data):
# Convert to grayscale
grayscale = np.mean(image_data, axis=2)
# Calculate mean brightness
mean_brightness = np.mean(grayscale)
return mean_brightness

class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.initUI()

def initUI(self):
self.setWindowTitle("Multi-QThread Example")
self.setGeometry(100, 100, 400, 300)

self.analog_label = QLabel("Analog data: --", self)
self.image_label = QLabel("Mean brightness: --", self)

layout = QVBoxLayout()
layout.addWidget(self.analog_label)
layout.addWidget(self.image_label)

container = QWidget()
container.setLayout(layout)
self.setCentralWidget(container)

self.video_thread = VideoThread()
self.video_thread.start()

self.analog_thread = AnalogReaderThread(self.video_thread)
self.analog_thread.update_analog_data.connect(self.update_analog_label)
self.analog_thread.start()

@pyqtSlot(float, float)
def update_analog_label(self, data, mean_brightness):
self.analog_label.setText(f"Analog data: {data:.2f}")
self.image_label.setText(f"Mean brightness: {mean_brightness:.2f}")

if __name__ == "__main__":
app = QApplication(sys.argv)
mainWin = MainWindow()
mainWin.show()
sys.exit(app.exec_())

И геттеры:
import sys
import time
import random
import numpy as np
from PyQt5.QtCore import QThread, pyqtSignal, pyqtSlot, QMutex
from PyQt5.QtWidgets import QApplication, QMainWindow, QLabel, QVBoxLayout, QWidget

class AnalogReaderThread(QThread):
update_analog_data = pyqtSignal(float, float)

def __init__(self, video_thread):
super().__init__()
self._video_thread = video_thread

def run(self):
while True:
analog_data = self.read_analog_data()
mean_brightness = self._video_thread.get_mean_brightness()
self.update_analog_data.emit(analog_data, mean_brightness)
time.sleep(1)

def read_analog_data(self):
return random.uniform(0, 100)

class VideoThread(QThread):
update_image_data = pyqtSignal(float)

def __init__(self):
super().__init__()
self._mean_brightness = 0
self._mutex = QMutex()

def run(self):
while True:
image_data = self.read_image_data()
mean_brightness = self.process_image(image_data)
self._mutex.lock()
self._mean_brightness = mean_brightness
self._mutex.unlock()
self.update_image_data.emit(self._mean_brightness)
time.sleep(0.5)

def read_image_data(self):
# Generate a random image (100x100 pixels, 3 color channels)
image = np.random.randint(0, 256, (100, 100, 3), dtype=np.uint8)
return image

def process_image(self, image_data):
# Convert to grayscale
grayscale = np.mean(image_data, axis=2)
# Calculate mean brightness
mean_brightness = np.mean(grayscale)
return mean_brightness

def get_mean_brightness(self):
self._mutex.lock()
value = self._mean_brightness
self._mutex.unlock()
return value

class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.initUI()

def initUI(self):
self.setWindowTitle("Multi-QThread Example")
self.setGeometry(100, 100, 400, 300)

self.analog_label = QLabel("Analog data: --", self)
self.image_label = QLabel("Mean brightness: --", self)

layout = QVBoxLayout()
layout.addWidget(self.analog_label)
layout.addWidget(self.image_label)

container = QWidget()
container.setLayout(layout)
self.setCentralWidget(container)

self.video_thread = VideoThread()
self.video_thread.start()

self.analog_thread = AnalogReaderThread(self.video_thread)
self.analog_thread.update_analog_data.connect(self.update_analog_label)
self.analog_thread.start()

@pyqtSlot(float, float)
def update_analog_label(self, data, mean_brightness):
self.analog_label.setText(f"Analog data: {data:.2f}")
self.image_label.setText(f"Mean brightness: {mean_brightness:.2f}")

if __name__ == "__main__":
app = QApplication(sys.argv)
mainWin = MainWindow()
mainWin.show()
sys.exit(app.exec_())


Подробнее здесь: https://stackoverflow.com/questions/787 ... 2-qthreads