一个 pyqt 应用程序中可以有多少个 QTimer?

问题描述 投票:0回答:2

我正在开发一个控制工厂的 pyqt 5 应用程序。 我们有很多:
做 a x 秒然后做 b y 分钟

其中每一个都是用(大部分)单次 QTimers 实现的。目前大约有110个这样的。 其中大部分不会同时运行,而且亚秒级的计时精度并不重要。

我们现在需要在此基础上实施安全检查,例如
如果输入 c 太高太久,做点什么。

从运营团队返回的列表有 250 多个条件,并且可能会增加。
坚持目前代码的编写方式,我会用另一个 QTimer 实现每一个。

代码针对系统的不同部分设置了不同的对象,条件将从设置文件中读取,因此使用大量 QTimer 实现安全系统是可行的,但这是一个好主意吗?

一个应用程序中有 300-400 个 QTimers 是否合理?如果它引起问题,症状会是什么?
有不同的方法吗?
来自现实世界经验的答案会很棒。

python pyqt pyqt5
2个回答
0
投票

没有真正的或绝对的限制:它仅取决于 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定时器实现整个过程。

技巧可能是使用“事件队列”,基于一个简单的字典,该字典按超时“分组”元素:

  • 键是“常见的”基于准确性的超时:它们的项目是包含最终调用超时函数的“计时器对象”的列表;
  • 独特的计时器调用一个通用的更新函数,它检查队列的每个sorted键(超时),然后:
    • 如果按键低于给定的参考值,触发超时功能;
    • 如果不是,则使用新的超时更新密钥,减少准确量;
    • 最后,将所有之前“触发”的元素添加到新的超时键中;
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%。


0
投票

这是我根据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()
© www.soinside.com 2019 - 2024. All rights reserved.