可变延时定时器

Question

我想要一个计时器（目前使用Python 3.8）首先检查系统时间或海军天文台时钟，这样它就可以随时启动并与00秒同步。

我只对系统时钟或海军天文台的秒数感兴趣。在每一分钟的顶部，即当秒数 = 00 时，我需要将数据写入 DataFrame 或数据库，然后再睡 60 秒。

我首先检查了系统时间，确定了从 00 秒开始的时间，并设置了第一个延迟时间。之后它应该延迟或休眠 60 秒，然后再次运行。数据在不断变化，但此时我只需要每 60 秒写入一次数据，也希望它能够使用其他时间范围，如 5 分钟、15 分钟等，但是一旦第一个完成，另一个时间框架会很容易。

这是我的蹩脚尝试，它运行几次迭代然后退出，我确信它不是很有效

def time_delay():
    sec = int(time.strftime('%S'))
    if sec != 0:
        wait_time = 60 - sec
        time.sleep(wait_time)
        sec = int(time.strftime('%S'))
        wait_time = 60 - sec
        
    elif time.sleep(60):
        time_delay()

Answer 1

这将在秒数为 0 时调用一个函数:

def time_loop(job):
    while True:
        if int(time.time()) % 60 == 0:
            job()
        time.sleep( 1 )

Answer 2

这并不像您想要的那样普遍，正如我在评论中所说，由于涉及固有的开销，不可能与时间源完全同步（但您可以非常接近）。

下面所做的是定义一个 Thread 子类，该子类可用于以固定间隔调用用户指定的函数，从系统时钟的整秒数开始。

它首先延迟到指定秒数的下一个 multiple 发生，然后从该点开始每隔 interval 秒调用一次函数，同时补偿执行指定函数的时间很长。

from threading import Thread, Event
import time


class TimedCalls(Thread):
    def __init__(self, func, interval, multiple):
        super().__init__()
        self.func = func
        self.interval = interval
        if multiple == 0:
            self.multiple = 60  # Avoid ZeroDivisionError.
        else:
            self.multiple = multiple
        self.stopped = Event()

    def cancel(self):
        self.stopped.set()

    def start(self):
        # Initially sync to a whole multiple of the interval.
        frac = self.multiple  - (time.time() % self.multiple)
        time.sleep(frac)
        super().start()

    def run(self):
        next_call = time.time()
        while not self.stopped.is_set():
            self.func(next_call)  # Target activity.
            next_call = next_call + self.interval
            # Block until beginning of next interval (unless canceled).
            self.stopped.wait(next_call - time.time())


def my_function(tm):
    print(f"{tm} → {time.strftime('%H:%M:%S', time.localtime(tm))}")


# Call function every 10 sec starting at multiple of 5 secs
timed_calls = TimedCalls(my_function, 10, 5)
timed_calls.start()
try:
    time.sleep(125)  # Let the timed calls occur.
finally:
    timed_calls.cancel()
print('done')

示例输出：

1617662925.0018559 → 15:48:45
1617662935.0018559 → 15:48:55
1617662945.0018559 → 15:49:05
1617662955.0018559 → 15:49:15
1617662965.0018559 → 15:49:25
1617662975.0018559 → 15:49:35
1617662985.0018559 → 15:49:45
1617662995.0018559 → 15:49:55
1617663005.0018559 → 15:50:05
1617663015.0018559 → 15:50:15
1617663025.0018559 → 15:50:25
1617663035.0018559 → 15:50:35
1617663045.0018559 → 15:50:45
done

可变延时定时器

Timer for variable time delay

python

variables

time

timer

delay