我正在开发一个控制工厂的 pyqt 5 应用程序。
我们有很多:
做 a x 秒然后做 b y 分钟
其中每一个都是用(大部分)单次 QTimers 实现的。目前大约有110个这样的。 其中大部分不会同时运行,而且亚秒级的计时精度并不重要。
我们现在需要在此基础上实施安全检查,例如
如果输入 c 太高太久,做点什么。
从运营团队返回的列表有 250 多个条件,并且可能会增加。
坚持目前代码的编写方式,我会用另一个 QTimer 实现每一个。
代码针对系统的不同部分设置了不同的对象,条件将从设置文件中读取,因此使用大量 QTimer 实现安全系统是可行的,但这是一个好主意吗?
一个应用程序中有 300-400 个 QTimers 是否合理?如果它引起问题,症状会是什么?
有不同的方法吗?
来自现实世界经验的答案会很棒。
没有真正的或绝对的限制:它仅取决于 CPU 的能力以及整个过程的实现方式。
下面的基本示例显示了一个带有自定义 QTableWidgetItem 子类的 QTableWidget,每个子类都有自己的 QTimer:当计时器到期时,它会更新其背景颜色。
from random import randrange
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
class TimerItem(QTableWidgetItem):
def __init__(self, text=''):
super().__init__(text)
self.timer = QTimer(singleShot=True, timeout=self.update)
self.timer.start(randrange(500, 2500))
def update(self):
self.setBackground(QColor(
randrange(255), randrange(255), randrange(255)
))
self.timer.start(randrange(500, 2500))
class UniqueTimerTable(QTableWidget):
def __init__(self, rows=50, cols=50):
super().__init__(rows, cols)
for r in range(rows):
for c in range(cols):
self.setItem(r, c, TimerItem('{} {}'.format(r + 1, c + 1)))
self.resizeColumnsToContents()
app = QApplication([])
w = UniqueTimerTable()
w.showMaximized()
app.exec()
使用上面的代码,我可以看到一个“低”的平均 CPU 使用率(大约 5%),即使我的 12 岁的电脑和窗口最大化......有 2500 个项目(和计时器)!
请注意,“高”CPU 使用率并非与计时器的数量唯一相关,但肯定是增加了 Qt 项目视图实现和更高级别(如 QTableWidget)的使用。
仍然,总有优化的空间。
你说精度并不关键,不需要亚秒级精度,意思是你大概可以用一个single定时器实现整个过程。
技巧可能是使用“事件队列”,基于一个简单的字典,该字典按超时“分组”元素:
class TimerObject(object):
def __init__(self, item):
self.item = item
def timeout(self):
self.item.setBackground(QColor(
randrange(255), randrange(255), randrange(255)
))
class CommonTimerTable(QTableWidget):
accuracy = .1
def __init__(self, rows=50, cols=50):
super().__init__(rows, cols)
self.queue = {}
for r in range(rows):
for c in range(cols):
item = QTableWidgetItem('{} {}'.format(r + 1, c + 1))
self.setItem(r, c, item)
timeout = randrange(1, 30) * self.accuracy
if timeout in self.queue:
timerList = self.queue[timeout]
else:
timerList = self.queue[timeout] = []
timerList.append(TimerObject(item))
self.resizeColumnsToContents()
self.queueTimer = QTimer(interval=1000 * self.accuracy,
timeout=self.processQueue)
self.queueTimer.start()
def processQueue(self):
renewed = []
# sorting is mandatory!!!
# it ensures that each new key is **always** below the previous value
# while still preserving the sorted order of the timeouts (the keys)
for timeout in sorted(self.queue.keys()):
if timeout <= self.accuracy:
objects = self.queue.pop(timeout)
renewed.extend(objects)
for timerObject in objects:
timerObject.timeout()
else:
self.queue[timeout - self.accuracy] = self.queue.pop(timeout)
# note: this must be done *after* updating the timeouts, because new
# objects have to be consistent with their *new* timeout
for timerObject in renewed:
timeout = randrange(1, 30) * self.accuracy
if timeout in self.queue:
timerList = self.queue[timeout]
else:
timerList = self.queue[timeout] = []
timerList.append(timerObject)
app = QApplication([])
w = CommonTimerTable()
w.showMaximized()
app.exec()
在上面的例子中,即使有相同的 2500 个元素,CPU 使用率也降低到平均 3%,即使精度为 100 毫秒。如果您对这种精度水平不感兴趣,将
_accuracy
值设置为 .5
(半秒),即使在我的旧机器上,CPU 使用率也不到 1%。
这是我根据musicamante的队列建议制作的demo。 它将安全警报添加到队列中并检查任何到期的警报。够轻了 希望能在主控软件里实现
from random import choice
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from datetime import datetime, timedelta
HIGH = 1
OK = 0
LOW = -1
WATCH = 5
alert_colors = {
HIGH: QColor("red"),
OK: QColor("white"),
LOW: QColor("lightblue"),
WATCH: QColor("bisque"),
}
TIMER_INTERVAL = 500
def random_walk_int(start=0, offsets=[-1, 0, 1]):
"""
generator to give integer value based on previous value
"""
x = start
while True:
yield (x)
dx = choice(offsets)
x = x + dx
class ItemInput(object):
def __init__(self, r, c, item):
self.r = r
self.c = c
self.id = f"{r:03}{c:03}"
self.item = item
self.max = 55
self.min = 45
self.timeout = 5 # seconds
self.gen = random_walk_int(start=50)
def value(self):
value = next(self.gen)
self.item.setText(f"{value}")
return value
def alert_action(self, alert_type):
self.item.setBackground(alert_colors[alert_type])
class CommonTimerTable(QTableWidget):
accuracy = 0.1 # seconds
def __init__(self, rows=50, cols=50):
super().__init__(rows, cols)
self.queue = {}
self.inputs = {}
self.rows = rows
self.cols = cols
for r in range(self.rows):
for c in range(self.cols):
item = QTableWidgetItem()
self.setItem(r, c, item)
input = ItemInput(r, c, item)
self.inputs[input.id] = input
self.resizeColumnsToContents()
self.alert_queue = {}
self.queueTimer = QTimer(interval=TIMER_INTERVAL, timeout=self.processQueue)
self.queueTimer.start()
def processQueue(self):
# check all inputs and update alert_queue
now = datetime.now()
for key, input in self.inputs.items():
value = input.value()
if value > input.max:
# if input not in alert_queue add it
if key not in self.alert_queue:
self.alert_queue[key] = {
"due time": now + timedelta(seconds=input.timeout),
"input": input,
"alert type": HIGH,
}
input.alert_action(WATCH)
elif value < input.min:
# if input not in alert_queue add it
if key not in self.alert_queue:
self.alert_queue[key] = {
"due time": now + timedelta(seconds=input.timeout),
"input": input,
"alert type": LOW,
}
input.alert_action(WATCH)
else:
# value is in bounds so if input in alert_queue remove it
if key in self.alert_queue:
del self.alert_queue[key]
input.alert_action(OK)
# action any due alerts
for key, alert in self.alert_queue.items():
if alert["due time"] < now:
alert["input"].alert_action(alert["alert type"])
app = QApplication([])
w = CommonTimerTable(rows=50, cols=50)
w.showMaximized()
app.exec()