Мигающие индикаторы PyQt5 Python

Рейтинг: 3Ответов: 1Опубликовано: 11.02.2023

У меня есть проблема с передачей данных между потоками. Есть небольшое приложение, идея которого в следующем:
есть кнопка, таблица и два индикатора - rx и rx. По нажатию на кнопку таблица должна начать заполняться данными, а индикаторы мигать.

У меня это реализовано в трех потоках.
Основной - само главное окно.
Дополнительный поток table_thread - обрабатывает данные и передает их в главный поток для заполнения таблицы. Так же этот поток сообщает третьему потоку rx_tx_blink_thread, что произошла запись и индикаторы должны мигнуть.

Все это реализовано с помощью воркеров fill_in_table_widget и rx_tx_blink_widget, с использованием метода moveToThread.

Основная проблема состоит в том, что нужно из воркера fill_in_table_widget нужно передать информацию о том, что нужно мигнуть в воркер rx_tx_blink_widget. А этот воркер должен передать информацию об изменении цвета индикатора в основной поток, затем подождать 50 мс и потом снова передать информацию об изменении цвета на изначальный.

У меня эта цепочка реализована так:

       self.table_thread = QtCore.QThread()
       self.fill_in_table_widget = FillTableWorker()
       self.fill_in_table_widget.moveToThread(self.table_thread)
       self.fill_in_table_widget.new_table_row.connect(self.add_row_in_table)
       self.table_thread.started.connect(self.fill_in_table_widget.run)

       self.rx_tx_blink_thread = QtCore.QThread()
       self.rx_tx_blink_widget = RxTxWorker()
       self.rx_tx_blink_widget.moveToThread(self.rx_tx_blink_thread)
       self.rx_tx_blink_widget.rx_tx_blink_signal.connect(self.rx_tx_blink)
       self.rx_tx_blink_thread.started.connect(self.rx_tx_blink_widget.run)
       # вот тут я соединяю сигнал  tx_rx воркера fill_in_table_widget  
       # с методом blink воркера rx_tx_blink_widget. А из этого метода я передаю сигнал в 
       # метод главного окна **rx_tx_blink**, но до него сигнал судя по всему не доходит и 
       #индикаторы не меняют цвет.
       self.fill_in_table_widget.tx_rx.connect(self.rx_tx_blink_widget.blink)

Помогите, пожалуйста, понять в чем проблема и как реализовать такую связь, чтобы первый дополнительный поток передавал сигнал второму дополнительному потоку, а тот уже передал его в главный поток, реализовывая задержку 50 мс.

Ниже прикрепляю весь код, он должен запускаться:

import sys
from PyQt5 import QtCore
from PyQt5.QtWidgets import *
from PyQt5.QtGui import *
from time import sleep


class MainWindow(QMainWindow):
   def __init__(self):
       super().__init__()
       # конфигурация главного окна и центрального виджета
       self.central_widget = QWidget(self)
       self.setGeometry(50, 50, 1500, 800)
       self.setWindowTitle('Чтение БКУ')
       self.setCentralWidget(self.central_widget)
       self.layout_main_window = QVBoxLayout()
       self.central_widget.setLayout(self.layout_main_window)
       # кнопка считать
       self.btn_start_reading = QPushButton("Считать")
       self.btn_start_reading.clicked.connect(self.start_read)
       self.layout_main_window.addWidget(self.btn_start_reading)
       # таблица
       self.table = QTableWidget()
       self.table.setColumnCount(9)
       self.table.setRowCount(13500)
       self.table.setHorizontalHeaderLabels(['№', 'Дата и время', 'БКУ', 'КЛ', 'АУ', 'Канал', 'Код события', 'Доп. параметр', 'Описание'])
       self.table.horizontalHeader().setStretchLastSection(True)
       self.table.horizontalHeader().setVisible(True)
       self.layout_main_window.addWidget(self.table)
       # индикаторы
       self.tx = QLabel()
       self.tx.setStyleSheet(rx_tx_yellow_stylesheet)
       self.rx = QLabel()
       self.rx.setStyleSheet(rx_tx_yellow_stylesheet)
       self.layout_main_window.addWidget(self.rx)
       self.layout_main_window.addWidget(self.tx)


   def start_read(self):
       # конфигурация доп потоков
       self.table_thread = QtCore.QThread()
       self.fill_in_table_widget = FillTableWorker()
       self.fill_in_table_widget.moveToThread(self.table_thread)
       self.fill_in_table_widget.new_table_row.connect(self.add_row_in_table)
       self.table_thread.started.connect(self.fill_in_table_widget.run)

       self.rx_tx_blink_thread = QtCore.QThread()
       self.rx_tx_blink_widget = RxTxWorker()
       self.rx_tx_blink_widget.moveToThread(self.rx_tx_blink_thread)
       self.rx_tx_blink_widget.rx_tx_blink_signal.connect(self.rx_tx_blink)
       self.rx_tx_blink_thread.started.connect(self.rx_tx_blink_widget.run)
       # соединяем два потока между собой
       self.fill_in_table_widget.tx_rx.connect(self.rx_tx_blink_widget.blink)

       self.table.clearContents() # очищаем таблицу от старых данных
       self.table_thread.start() # запускаем поток записи данных в таблицу

       # если раскомментирвоать поток снизу, программа начинает виснусть.
       # self.rx_tx_blink_thread.start() # запускаем поток мигания индикаторов

   
   @QtCore.pyqtSlot(int, object)   
   def add_row_in_table(self, i, list):
       self.table.setItem(i, 0, list[0])
       self.table.setItem(i, 1, list[1])
       self.table.setItem(i, 2, list[2])
       self.table.setItem(i, 3, list[3])
       self.table.setItem(i, 4, list[4])
       self.table.setItem(i, 5, list[5])
       self.table.setItem(i, 6, list[6])
       self.table.setItem(i, 7, list[7])
       self.table.setItem(i, 8, list[8])

   
   @QtCore.pyqtSlot(str, str)
   def rx_tx_blink(self, rx, tx):
       if rx == 'blink':
           self.rx.setStyleSheet(rx_stylesheet)
       elif rx == 'white':
           self.rx.setStyleSheet(rx_tx_yellow_stylesheet)
       if tx == 'blink':
           self.tx.setStyleSheet(tx_stylesheet)
       elif tx == 'white':
           self.tx.setStyleSheet(rx_tx_yellow_stylesheet)


rx_tx_yellow_stylesheet = """
       border-radius: 6px;
       min-height: 12px;
       max-height: 12px;
       min-width: 12px;
       max-width: 12px;
       background-color: yellow;
   """

rx_stylesheet = """
       border-radius: 6px;
       min-height: 12px;
       max-height: 12px;
       min-width: 12px;
       max-width: 12px;
       background-color: green;
   """
tx_stylesheet = """
       border-radius: 6px;
       min-height: 12px;
       max-height: 12px;
       min-width: 12px;
       max-width: 12px;
       background-color: red;
   """


class FillTableWorker(QtCore.QObject):
   new_table_row = QtCore.pyqtSignal(int, object)
   tx_rx = QtCore.pyqtSignal(str)

   def __init__(self):
       super().__init__()
       self.i = 0

   def run(self):
       i = self.i
       for i in range(12000):
           event = ['12:10:2022_11:37:56', 1, 1, 1, 1, 138, 1, 1]
           self.tx_rx.emit('Tx')
           list = [
               QTableWidgetItem(str(i+1)),
               QTableWidgetItem(str(event[0])),
               QTableWidgetItem(str(event[1])),
               QTableWidgetItem(str(event[2])),
               QTableWidgetItem(str(event[3])),
               QTableWidgetItem(str(event[4])),
               QTableWidgetItem(str(event[5])),
               QTableWidgetItem(str(event[6])),
               QTableWidgetItem(str(event[7])),
           ]
           self.new_table_row.emit(i, list)
           self.tx_rx.emit('Rx')
           i += 1
           sleep(0.05)


class RxTxWorker(QtCore.QObject):
   rx_tx_blink_signal = QtCore.pyqtSignal(str, str)

   def __init__(self):
       super().__init__()
       self.rx_color = 'white'
       self.tx_color = 'white'

   def run(self):
       while True:
           self.rx_tx_blink_signal.emit(self.rx_color, self.tx_color)


   @QtCore.pyqtSlot(str)
   def blink(self, text):
       if text == 'Rx':
           self.rx_color = 'blink'
           sleep(0.05)
           self.rx_color = 'white'
       elif text == 'Tx':
           self.tx_color = 'blink'
           sleep(0.05)
           self.tx_color = 'white'


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

Ответы

▲ 2Принят

Я не понял вашу идею с третьим потоком rx_tx_blink_thread и попробовал реализовать вашу идею используя:

void QTimer::singleShot(int msec, const QObject *receiver, const char *member)

Эта статическая функция вызывает слот через заданный интервал времени.
Использовать эту функцию очень удобно, потому что вам не нужно заморачиваться с timerEvent или создавать локальный объект QTimer.

Скажите, получилось у меня или что-то пошло не так.

import sys
# from time import sleep              # заменил на QtCore.QThread.msleep(50)

from PyQt5 import QtCore
from PyQt5.QtWidgets import *
from PyQt5.QtGui import *


rx_tx_yellow_stylesheet = """
       border-radius: 6px;
       min-height: 12px;
       max-height: 12px;
       min-width: 12px;
       max-width: 12px;
       background-color: yellow;
   """

rx_stylesheet = """
       border-radius: 6px;
       min-height: 12px;
       max-height: 12px;
       min-width: 12px;
       max-width: 12px;
       background-color: green;
   """
tx_stylesheet = """
       border-radius: 6px;
       min-height: 12px;
       max-height: 12px;
       min-width: 12px;
       max-width: 12px;
       background-color: red;
   """


class FillTableWorker(QtCore.QObject):
   new_table_row = QtCore.pyqtSignal(int, object)
#   tx_rx = QtCore.pyqtSignal(str)

   def __init__(self):
       super().__init__()
       self.i = 0

   def run(self):
       i = self.i
       for i in range(12000):
           event = ['12:10:2022_11:37:56', 1, 1, 1, 1, 138, 1, 1]
#           self.tx_rx.emit('Tx')
           list = [
               QTableWidgetItem(str(i+1)),
               QTableWidgetItem(str(event[0])),
               QTableWidgetItem(str(event[1])),
               QTableWidgetItem(str(event[2])),
               QTableWidgetItem(str(event[3])),
               QTableWidgetItem(str(event[4])),
               QTableWidgetItem(str(event[5])),
               QTableWidgetItem(str(event[6])),
               QTableWidgetItem(str(event[7])),
           ]
           self.new_table_row.emit(i, list)
#           self.tx_rx.emit('Rx')
           i += 1
#           sleep(0.05)
           QtCore.QThread.msleep(50)


class MainWindow(QMainWindow):
   def __init__(self):
       super().__init__()
       self.central_widget = QWidget(self)
       self.setGeometry(50, 50, 1300, 600)
       self.setWindowTitle('Чтение БКУ')
       self.setCentralWidget(self.central_widget)
       self.layout_main_window = QVBoxLayout()
       self.central_widget.setLayout(self.layout_main_window)

       # кнопка считать
       self.btn_start_reading = QPushButton("Считать")
       self.btn_start_reading.clicked.connect(self.start_read)
       self.layout_main_window.addWidget(self.btn_start_reading)
       
       # таблица
       self.table = QTableWidget()
       self.table.setColumnCount(9)
       self.table.setRowCount(13500)
       self.table.setHorizontalHeaderLabels(['№', 'Дата и время', 
           'БКУ', 'КЛ', 'АУ', 'Канал', 'Код события', 
           'Доп. параметр', 'Описание'])
       self.table.horizontalHeader().setStretchLastSection(True)
       self.table.horizontalHeader().setVisible(True)
       self.layout_main_window.addWidget(self.table)
       # индикаторы
       self.tx = QLabel()
       self.tx.setStyleSheet(rx_tx_yellow_stylesheet)
       self.rx = QLabel()
       self.rx.setStyleSheet(rx_tx_yellow_stylesheet)
       self.layout_main_window.addWidget(self.rx)
       self.layout_main_window.addWidget(self.tx)

   def start_read(self):
       # конфигурация доп потоков
       self.table_thread = QtCore.QThread()
       self.fill_in_table_widget = FillTableWorker()
       self.fill_in_table_widget.moveToThread(self.table_thread)
       self.fill_in_table_widget.new_table_row.connect(self.add_row_in_table)
       self.table_thread.started.connect(self.fill_in_table_widget.run)
      
       self.table.clearContents() # очищаем таблицу от старых данных
       self.table_thread.start()  # запускаем поток записи данных в таблицу

# !!! vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv   
   @QtCore.pyqtSlot(int, object)   
   def add_row_in_table(self, i, list):
       self.blink('Tx')                                         # !!! +++
   
       self.table.setItem(i, 0, list[0])
       self.table.setItem(i, 1, list[1])
       self.table.setItem(i, 2, list[2])
       self.table.setItem(i, 3, list[3])
       self.table.setItem(i, 4, list[4])
       self.table.setItem(i, 5, list[5])
       self.table.setItem(i, 6, list[6])
       self.table.setItem(i, 7, list[7])
       self.table.setItem(i, 8, list[8])

       QtCore.QTimer.singleShot(25, lambda: self.blink('Rx'))   # !!! +++

   def blink(self, text):
       if text == 'Rx':
           self.rx_tx_blink(rx='blink', tx='white')
           QtCore.QTimer.singleShot(50, 
               lambda: self.rx_tx_blink(rx='white', tx='blink'))
       elif text == 'Tx':
           self.rx_tx_blink(rx='white', tx='blink')       
           QtCore.QTimer.singleShot(50, 
               lambda: self.rx_tx_blink(rx='blink', tx='white'))

   def rx_tx_blink(self, rx, tx):
       if rx == 'blink':
           self.rx.setStyleSheet(rx_stylesheet)
           self.tx.setStyleSheet(rx_tx_yellow_stylesheet)
       elif tx == 'blink':
           self.tx.setStyleSheet(tx_stylesheet)
           self.rx.setStyleSheet(rx_tx_yellow_stylesheet)
# !!! ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^


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

введите сюда описание изображения

введите сюда описание изображения

введите сюда описание изображения