Вы можете использовать threading.Timer
. Вы должны сделать некоторую математику, чтобы рассчитать время следующего запуска. datetime
имеет удобный метод replace
для этого.
from datetime import datetime, timedelta
from threading import Timer
def get_sleep_time():
now = datetime.now()
next_run = now.replace(minute=int(now.minute / 5) * 5, second=0, microsecond=0) + timedelta(minutes=5)
return (next_run - now).total_seconds()
def dowork():
now = datetime.now()
print('Doing some work at', now)
schedule_next_run()
def schedule_next_run():
sleep_time = get_sleep_time()
print(f'sleeping for {sleep_time} seconds')
t = Timer(sleep_time, dowork)
t.daemon = True
t.start()
print('Starting work schedule')
schedule_next_run()
input('Doing work every 5 minutes. Press enter to exit')
В моей системе функция срабатывает в течение половины миллисекунды от целевого времени
Обратите внимание, что расчет времени округляется в меньшую сторонуи затем добавляет timedelta
, чтобы тщательно обернуть вокруг конца каждого часа. Возможно, вы захотите подумать, как это будет происходить при переходе на летнее время.
Предложение : переместите всю эту логику в класс, чтобы очистить ее.