在PyQt工作线程中使用QTimer

Question

我正在使用串行设备并根据接收到的数据设置一个标志(这是全局变量)。现在我想通过使用计时器在一段时间后(例如一秒钟)重置标志。

代码如下:

class Inlet_Worker(QObject):
def __init__(self):
super(Inlet_Worker, self).__init__()
self.timer = QTimer(self)
self.timer.timeout.connect(self.Reset_Register_Barcode)
def run(self):
global Register_Barcode
while True :
if  client.read_coils(address = 0x0802).bits[0]:
Register_Barcode = True
self.timer.start(1000)


def Reset_Register_Barcode(self):
global Register_Barcode
Register_Barcode = False

但是定时器不工作。

Answer 1

我将从您的示例代码中假设您正在使用QThread，并且您也使用QObject。你的工作对象上的moveToThread。这是正确的步骤，但还有一些其他的事情你必须做，使你的计时器工作。

首先，您应该使用单镜头计时器，以避免在当前计时器处于活动状态时重新注册。其次，必须显式地处理任何挂起的事件，因为while循环将阻塞线程的事件循环。如果没有这个，计时器的超时信号将永远不会发出。第三，您应该确保在程序退出时工作线程和线程干净地关闭(这也将防止任何Qt错误消息)。最后，如果可能的话，您应该使用信号将注册更改传达给主GUI线程，而不是全局变量。

下面的演示脚本(基于您的示例)实现了所有这些。单击Start按钮后，线程将启动并定期更新注册(由复选框指示)。希望您能够看到如何使其适应您的实际应用程序:

from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
class Inlet_Worker(QObject):
barcodeRegistered = pyqtSignal(bool)
def __init__(self):
super().__init__()
self._stopped = False
self._registered = False
self.timer = QTimer(self)
self.timer.setSingleShot(True)
self.timer.timeout.connect(self.updateBarcodeRegistration)
def run(self):
count = 0
self._stopped = False
while not self._stopped:
#if client.read_coils(address = 0x0802).bits[0]:
count += 1
if count % 20 == 0 and not self._registered:
self.updateBarcodeRegistration(True)
self.timer.start(2000)
QCoreApplication.processEvents()
QThread.msleep(100)
self.updateBarcodeRegistration(False)
self.timer.stop()
print('Stopped')
def updateBarcodeRegistration(self, enable=False):
print('Register' if enable else 'Reset')
self._registered = enable
self.barcodeRegistered.emit(enable)
def stop(self):
self._stopped = True

class Window(QWidget):
def __init__(self):
super().__init__()
self.thread = QThread()
self.worker = Inlet_Worker()
self.worker.moveToThread(self.thread)
self.button = QPushButton('Start')
self.check = QCheckBox('Registered')
layout = QHBoxLayout(self)
layout.addWidget(self.button)
layout.addWidget(self.check)
self.thread.started.connect(self.worker.run)
self.button.clicked.connect(self.thread.start)
self.worker.barcodeRegistered.connect(self.check.setChecked)
def closeEvent(self, event):
self.worker.stop()
self.thread.quit()
self.thread.wait()
if __name__ == '__main__':
app = QApplication(['Test'])
window = Window()
window.setGeometry(600, 100, 200, 50)
window.show()
app.exec()