我每小时运行一个脚本(计时器为3600秒(。但每次运行时,我的导出文件(脚本的结果(的时间戳都晚于上一次。几乎不可能找到最佳点并坚持大约相同的出口时间。有什么建议可以拥有一个更稳定可靠的计时器吗?
代码:
import pandas as pd
from datetime import datetime as dt
import schedule
import time
def export():
df = pd.read_excel("filepath", sheet_name='Import Data')
now =dt.now()
dt_string = now.strftime("%Y%m%d %H%M%S")
df["Minute"] = now.minute
df["Hour"] = now.hour
df["Day"] = now.day
df["Month"] = now.month
df["Year"] = now.year
df.to_excel("filepath" + dt_string + ".xlsx", sheet_name='Export Data')
schedule.every(3600).seconds.do(export)
while 1:
schedule.run_pending()
time.sleep(1)
解决这个问题的方法是根据实际时钟调整运行时间。我所说的实际时钟是指UTC时间,或者你系统上的任何时间。时区并不重要。重要的是根据时间安排,而不是根据上次运行后经过的时间。
测量经过的时间的问题是,你需要考虑进行测量所需的时间,也可能有其他因素导致延迟。
解决方案1
在我意识到schedule
库也可以做到这一点之前,这是我最初的想法
你可能想直接跳到下面的解决方案2,那更简单
下面是一个示例代码。它每5秒运行一个方法。这样,延迟不会随着时间的推移而改变,时间将保持不变。
您需要根据自己的需要进行调整。你可以修改它,使它每小时运行一次,而不是每5秒运行一次。
示例代码:
import time
from datetime import datetime as dt
import threading
from threading import Event
def export():
print("Do something every 5 seconds.")
run_once = Event()
run_once.clear()
def run_method(method_to_run):
while 1:
run_once.wait()
run_once.clear()
method_to_run()
thread = threading.Thread(target=run_method, args=(export,))
thread.start()
last_run = 0
while 1:
when_to_run = int(dt.now().second / 5)
if when_to_run > last_run:
last_run = when_to_run
run_once.set()
elif when_to_run < last_run:
last_run = when_to_run
time.sleep(1)
编辑1:如果你想每小时运行一次任务,那么你需要修改上面例子中的以下行:
# when_to_run = int(dt.now().second / 5)
when_to_run = int(dt.now().hour)
编辑2:添加解决方案2。
解决方案2
使用schedule
库的.at()
方法(感谢@itprorh66的评论。(
文件:https://schedule.readthedocs.io/en/stable/reference.html#schedule.Job.at
示例代码:
import time
import schedule
def export():
print("Do something every 5 seconds.")
schedule.every().hour.at(":00").do(export)
while 1:
schedule.run_pending()
time.sleep(1)