如何提高DMX控制Python脚本的计时精度

Question

我有一个 Python 脚本，可以通过 OLA 库控制 DMX 照明。该脚本从 CSV 文件读取颜色值，并以 120 毫秒的固定间隔将它们发送到 DMX 控制器。

我使用 time.perf_counter() 进行计时，但随着时间的推移，我遇到了漂移。具体来说，当脚本发送超过 1000 个值的大序列时，它在传递 100 个值后开始漂移超过 40 毫秒。我的应用需要极高的精度。

import array
import csv
import time
import warnings
from ola.ClientWrapper import ClientWrapper
from tqdm import tqdm

# Suppress DeprecationWarning
warnings.filterwarnings("ignore", category=DeprecationWarning)

def custom_sleep(duration, get_now=time.perf_counter):
    now = get_now()
    end = now + duration
    while now < end:
        now = get_now()

def read_csv(filename):
    with open(filename, newline='') as file:
        reader = csv.reader(file)
        next(reader)  # Skip the header row
        return [list(map(int, row)) for row in reader]

def DmxSent(state):
    pass

def main():
    universe = 1
    dmx_values = read_csv('dmx_values.csv')

    wrapper = ClientWrapper()
    client = wrapper.Client()

    with open('log_file.txt', 'w') as log_file:
        for values in tqdm(dmx_values, desc="Processing DMX Values"):
            start_time = time.perf_counter()
            data = array.array('B', values + [0] * (512 - len(values)))
            client.SendDmx(universe, data, DmxSent)
            custom_sleep(0.120)  # Hold for 120ms
            elapsed_time = (time.perf_counter() - start_time) * 1000
            log_file.write(f"Elapsed Time: {elapsed_time:.2f} ms\n")

if __name__ == "__main__":
    main()

我实现了一个custom_sleep函数，使用time.perf_counter()来实现更精确的计时，但仍然出现漂移。

我添加了日志记录来测量每次操作所花费的实际时间，但漂移仍然存在。我通过用摄像机以 25fps 记录光源并观察到值 100 左右的漂移来确认这一点。

我希望计时具有高精度，并且随着时间的推移漂移最小。

我正在 MacBook Air M1 上运行该脚本。我使用的 DMX 接口是 Enttec USB Pro。

我正在考虑将 C/C++ 集成到时序关键部分，但希望获得社区关于最佳方法的建议。

Answer 1

Pyglet.clock 不应随时间漂移，但在滴答之后您的函数运行仍然会有延迟。每次调用函数时，这应该相对恒定，因此在您的情况下可能不是问题。

Answer 2

问题是您必须注意通过驱动程序发送命令来控制设备所需的时间。因此，您无法在发送命令“之后”开始计时。

更具体地说，你必须调用你的 custom_sleep 并指定开始时间

custom_sleep(0.120, lambda: start_time)

它会避免漂移，但你必须小心间隙是否足够命令。

您还可以通过在计时器中包含所有操作的时间来提高精度和一致性，例如

   time_interval = 0.120
   gap = 0
   start_time = time.perf_counter()
   for values in tqdm(dmx_values, desc="Processing DMX Values"):
        data = array.array('B', values + [0] * (512 - len(values)))
        client.SendDmx(universe, data, DmxSent)
        elapsed_time = (time.perf_counter() - start_time) * 1000
        log_file.write(f"Elapsed Time: {elapsed_time:.2f} ms\n")
        gap = gap + time_interval
        custom_sleep(gap, start_time)  # Hold for 120ms

我不认为你的睡眠功能会导致漂移，但你正在使用繁忙的循环来检查时间。我就不详细说了，因为方法太多了。您可以参考执行周期性动作

对于实时应用程序，您可以考虑使用rtos。它可以像设置一个计时器并调用您正在使用的 ola 库的 api 一样简单，但这实际上取决于您的情况。