如何使用 Threading.Timer 每 0.05 秒精确地从 Modbus 服务器读取数据？

问题描述

我想达到什么目的？

每 0.05 秒精确地从 Modbus 服务器读取数据。

更大的图景是我正在创建一个 PyQt5 应用程序，通过它我想要保存和绘制 Modbus 数据。所以稍后我可以将它们用于 PID 自动调节。 PID 自动调谐器要求以至少 0.05 秒的精度测量数据。并且数据点也需要像这样均匀分布：M wait 0.05s then M wait 0.05s then M 而不是这样：M wait 0.08s then M wait 0.03s then M (M = measure data)。

到目前为止我尝试了什么？

我尝试实现 Threading.Timer 每 0.05 秒读取一次数据。
问题是定时器的精度太低了。

这是我用来测试 Threading.Timer 精度的代码：

from threading import Timer
import timeit

starttime = timeit.default_timer()

def f():
    Timer(0.05,f).start()

    global starttime

    print("The time difference is :",timeit.default_timer() - starttime)
    starttime = timeit.default_timer()

    #Here I would read the data

f()

代码产生以下输出：

...
The time difference is : 0.07623
The time difference is : 0.07707
The time difference is : 0.07684
The time difference is : 0.07557
...

如果我要以这种精度读取数据。从长远来看，这会造成巨大的时差。
理想情况下，代码每隔 0.05s 读取一次数据，即每秒读取 20 次。但以这种精度，它将平均每 0.07s 次读取数据并将其解释为 0.05s。
每秒的时差为 0.07 * 20 - 0.05 * 20 = 0.4 seconds，这是不可接受的，因为一分钟后时移将是 24 秒。

如何提高`Threading.Timer`对象的准确度？或者我应该使用哪些其他方法/工具？

其他信息：

我测量了使用以下代码从 Modbus 服务器读取一个值需要多长时间：

from pyModbusTCP.client import ModbusClient
import timeit

c = ModbusClient("localhost")
c.open()

for i in range(0,10):
    starttime = timeit.default_timer()
    data_Now = c.read_holding_registers(0,1)
    print("Reading 1 register takes:",timeit.default_timer() - starttime)

输出：

...
Reading 1 register takes: 0.000297100
Reading 1 register takes: 0.000307600
Reading 1 register takes: 0.000271699
...

我使用的是什么：

Windows 10
Python 3.7（32 位版本）

解决方法

你正在过度设计。只需制作一个 FPS 锁定。如果读数超过 0.05，则必须向后重新计算（未实现）。如果您快速测量，请根据您想要的 0.05 秒重新计算并等待该时间。使用此方法，您可以实现精确的 0.05 秒间隔。如果读取寄存器的时间超过您的时间，则无法使用

这是具有 FPS 锁定的工作示例。没有什么花哨。设置精度以创建假读取延迟。为你的目的设置周期（我设置 1s，你想要 0.05s）

import time
import random

def time_keeper(lost_by_reading,period=1):
    if lost_by_reading>period:
        pass # a problem because reading takes longer than period
    elif lost_by_reading<period:
        time.sleep(period-lost_by_reading)


def mock_register_reading(precision=1000):
    rand_sleep = random.randint(0,10)/precision
    time.sleep(rand_sleep)


_measure = time.time()

period = 1

for i in range(1000):

    start = time.time()
    mock_register_reading()

    ## just for log
    print('measured after',time.time()-_measure,',ERROR: ',period-(time.time()-_measure))
    _measure = time.time()
    ### end

    finish = time.time()
    reading_time = finish-start
    time_keeper(reading_time,period = period)

注意：我建议您不要在 modbus 方面使用线程。根据我自己的经验，modbus 和线程不是朋友，例如通过线程读取寄存器只会导致灾难（以防万一您有想法每 0.5 读取一次线程）

modbus multithreading multithreading python timer timer