Мигающие индикаторы PyQt5 Python
У меня есть проблема с передачей данных между потоками. Есть небольшое приложение, идея которого в следующем:
есть кнопка, таблица и два индикатора - 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_())