实现 sched

问题描述

我已经为 sched.scheduler 编写了一个基本的包装器，但我得到了：

  File "/usr/local/anaconda3/lib/python3.7/sched.py",line 149,in run
    delayfunc(time - Now)
OSError: [Errno 22] Invalid argument

这个错误的原因是什么？我正在关注https://docs.python.org/3/library/sched.html

代码：

#!/usr/local/anaconda3/bin/python

from time import time,sleep
from datetime import datetime,timezone
from sched import scheduler

class Scheduler:
    # specify one or t'other  e.g. "2021-03-17 02:29:00"
    def __init__(self,event_time_utc:str='',local_unix_timestamp_override:float=0):
        self.event_local_timestamp = Scheduler.to_local_unix_timestamp(event_time_utc) \
            if event_time_utc else local_unix_timestamp_override

        # self.seconds_until_event = event_local_timestamp - time()

        print(self.event_local_timestamp - time())

        self.sch = scheduler()

    def run(self,blocking=False):
        self.sch.run(blocking)


    @staticmethod
    def to_local_unix_timestamp(time_utc: str):
        dt_utc = datetime.strptime(time_utc,"%Y-%m-%d %H:%M:%s")
        dt_local = dt_utc.replace(tzinfo=timezone.utc)
        return dt_local.timestamp()


    @staticmethod
    def to_human_readable(unix_timestamp):
        return datetime.fromtimestamp(unix_timestamp).strftime('%Y-%m-%d %H:%M:%s.%f')


    def schedule(self,offset_from_event_time_s:float,callback):
        print('Event scheduled for local unix timestamp:',self.event_local_timestamp + offset_from_event_time_s)

        self.sch.enterabs(
            time = self.event_local_timestamp + offset_from_event_time_s,priority = 1,action = callback
            )


if __name__ == '__main__':
    sch = Scheduler(local_unix_timestamp_override=time()+5)

    print(f'Started at time: {Scheduler.to_human_readable(time())}')

    def callback():
        print(f'Callback at time: {Scheduler.to_human_readable(time())}')

    sch.schedule(offset_from_event_time_s=1,callback=callback)

    sch.run(blocking=True)

输出：

> ./scheduler.py 
4.999997854232788
Started at time: 2021-03-17 11:00:03.424191
f 1615953609.424139
Traceback (most recent call last):
  File "./scheduler.py",line 55,in <module>
    sch.run(blocking=True)
  File "./scheduler.py",line 20,in run
    self.sch.run(blocking)
  File "/usr/local/anaconda3/lib/python3.7/sched.py",in run
    delayfunc(time - Now)
OSError: [Errno 22] Invalid argument

解决方法

问题的根源在于您假设 sched 模块使用 time.time() 作为其计时功能。这是一个错误的假设。它默认使用 time.monotonic()，它返回自程序启动以来的时间。这将是一个很小的数字，比如 6。当你试图延迟到 1615644884.1231558 时，这会让它变得疯狂。

改变这个：

        self.sch = scheduler()

为此：

        self.sch = scheduler(time)

我尝试调试该问题，发现 time.sleep() 中允许的浮点值的最大长度为 8（我找不到任何关于为什么只有 8 的资源）但由于您使用的是时间戳，因此该值超过了.您可以在您的机器上尝试以下代码并检查：

import time

# Will work fine
time.sleep(16156448)
    
# Will result in error
time.sleep(1615644884.1231558)

invalid-argument python scheduler