Я хочу многократно выполнять функцию в Python каждые 60 секунд навсегда (просто как NSTimer в Objective C). Этот код будет запускаться как демон и эффективно, как вызов python script каждую минуту, используя cron, но не требуя, чтобы его настраивал пользователь.
В этом вопросе о cron, реализованном в Python, решение кажется эффективно sleep ( ) за х секунд. Мне не нужны такие расширенные функции, поэтому возможно что-то вроде этого будет работать
while True:
# Code executed here
time.sleep(60)
Есть ли какие-либо предсказуемые проблемы с этим кодом?
Используйте модуль sched, который реализует планировщик событий общего назначения.
import sched, time
s = sched.scheduler(time.time, time.sleep)
def do_something(sc):
print "Doing stuff..."
# do your stuff
s.enter(60, 1, do_something, (sc,))
s.enter(60, 1, do_something, (s,))
s.run()
Просто заблокируйте петлю времени на системные часы. Легко.
import time
starttime=time.time()
while True:
print "tick"
time.sleep(60.0 - ((time.time() - starttime) % 60.0))
Возможно, вы захотите рассмотреть Twisted, которая является сетевой библиотекой Python, которая реализует шаблон Reactor.
from twisted.internet import task, reactor
timeout = 60.0 # Sixty seconds
def doWork():
#do work here
pass
l = task.LoopingCall(doWork)
l.start(timeout) # call every sixty seconds
reactor.run()
Хотя "while True: sleep (60)", вероятно, будет работать Twisted, вероятно, уже реализует многие функции, которые вам в конечном итоге понадобятся (демонанизация, ведение журнала или обработка исключений, как указано в bobince) и, вероятно, будет более надежным решением
Если вам нужен неблокирующий способ выполнять вашу функцию периодически, вместо блокирующего бесконечного цикла я бы использовал поточный таймер. Таким образом, ваш код может продолжать работать и выполнять другие задачи, и ваша функция вызывается каждые n секунд. Я часто использую эту технику для печати информации о ходе выполнения длительных задач с CPU/Disk/Network.
Здесь код, который я опубликовал в аналогичном вопросе, с помощью функции start() и stop():
from threading import Timer
class RepeatedTimer(object):
def __init__(self, interval, function, *args, **kwargs):
self._timer = None
self.interval = interval
self.function = function
self.args = args
self.kwargs = kwargs
self.is_running = False
self.start()
def _run(self):
self.is_running = False
self.start()
self.function(*self.args, **self.kwargs)
def start(self):
if not self.is_running:
self._timer = Timer(self.interval, self._run)
self._timer.start()
self.is_running = True
def stop(self):
self._timer.cancel()
self.is_running = False
Использование:
from time import sleep
def hello(name):
print "Hello %s!" % name
print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
sleep(5) # your long-running job goes here...
finally:
rt.stop() # better in a try/finally block to make sure the program ends!
Особенности:
start() и stop() безопасны для вызова несколько раз, даже если таймер уже запущен/остановленinterval в любое время, это будет эффективно после следующего прогона. То же самое для args, kwargs и даже function!Чем проще я верю:
import time
def executeSomething():
#code here
time.sleep(60)
while True:
executeSomething()
Таким образом ваш код будет выполнен, тогда он ждет 60 секунд, после чего он снова запускается, ждет, выполняет и т.д. Не нужно усложнять вещи: D
Здесь обновляется код из MestreLion, который позволяет избежать дрейфации с течением времени:
import threading
import time
class RepeatedTimer(object):
def __init__(self, interval, function, *args, **kwargs):
self._timer = None
self.interval = interval
self.function = function
self.args = args
self.kwargs = kwargs
self.is_running = False
self.next_call = time.time()
self.start()
def _run(self):
self.is_running = False
self.start()
self.function(*self.args, **self.kwargs)
def start(self):
if not self.is_running:
self.next_call += self.interval
self._timer = threading.Timer(self.next_call - time.time(), self._run)
self._timer.start()
self.is_running = True
def stop(self):
self._timer.cancel()
self.is_running = False
import time, traceback
def every(delay, task):
next_time = time.time() + delay
while True:
time.sleep(max(0, next_time - time.time()))
try:
task()
except Exception:
traceback.print_exc()
# in production code you might want to have this instead of course:
# logger.exception("Problem while executing repetitive task.")
# skip tasks if we are behind schedule:
next_time += (time.time() - next_time) // delay * delay + delay
def foo():
print("foo", time.time())
every(5, foo)
Если вы хотите сделать это, не блокируя свой оставшийся код, вы можете использовать его, чтобы он запускался в своем собственном потоке:
import threading
threading.Thread(target=lambda: every(5, foo)).start()
Это решение сочетает в себе несколько функций, которые редко встречаются в других решениях:
threading.Timer или что-то еще), это приведет к прекращению цепочки. Тогда дальнейших казней не произойдет, даже если причина проблемы уже исправлена. Простой цикл и ожидание с простым sleep() гораздо более надежны в сравнении.next_time += delay.Я столкнулся с аналогичной проблемой некоторое время назад. Может быть http://cronus.readthedocs.org может помочь?
Для v0.2 работает следующий фрагмент
import cronus.beat as beat
beat.set_rate(2) # 2 Hz
while beat.true():
# do some time consuming work here
beat.sleep() # total loop duration would be 0.5 sec
Основное различие между этим и cron заключается в том, что исключение будет убивать демона навсегда. Возможно, вы захотите обернуть узлом исключения и логгером.
Один возможный ответ:
import time
t=time.time()
while True:
if time.time()-t>10:
#run your task here
t=time.time()
Я использую это, чтобы вызывать 60 событий в час, при этом большинство событий происходит через такое же количество секунд после целой минуты:
import math
import time
import random
TICK = 60 # one minute tick size
TICK_TIMING = 59 # execute on 59th second of the tick
TICK_MINIMUM = 30 # minimum catch up tick size when lagging
def set_timing():
now = time.time()
elapsed = now - info['begin']
minutes = math.floor(elapsed/TICK)
tick_elapsed = now - info['completion_time']
if (info['tick']+1) > minutes:
wait = max(0,(TICK_TIMING-(time.time() % TICK)))
print ('standard wait: %.2f' % wait)
time.sleep(wait)
elif tick_elapsed < TICK_MINIMUM:
wait = TICK_MINIMUM-tick_elapsed
print ('minimum wait: %.2f' % wait)
time.sleep(wait)
else:
print ('skip set_timing(); no wait')
drift = ((time.time() - info['begin']) - info['tick']*TICK -
TICK_TIMING + info['begin']%TICK)
print ('drift: %.6f' % drift)
info['tick'] = 0
info['begin'] = time.time()
info['completion_time'] = info['begin'] - TICK
while 1:
set_timing()
print('hello world')
#random real world event
time.sleep(random.random()*TICK_MINIMUM)
info['tick'] += 1
info['completion_time'] = time.time()
В зависимости от реальных условий вы можете получить тики длины:
60,60,62,58,60,60,120,30,30,60,60,60,60,60...etc.
но в конце 60 минут у вас будет 60 тиков; и большинство из них произойдет с правильным смещением в минуту, который вы предпочитаете.
В моей системе я получаю типичный дрейф < 1/20 секунд, пока не возникнет необходимость в коррекции.
Преимуществом этого метода является разрешение тактового дрейфа; которые могут вызвать проблемы, если вы делаете такие вещи, как добавление одного элемента за галочку, и вы ожидаете, что в течение часа будет добавлено 60 элементов. Неспособность учитывать дрейф может вызвать вторичные признаки, такие как скользящие средние, чтобы рассматривать данные слишком глубоко в прошлом, что приводит к ошибочному выводу.
например, Отобразить текущее местное время
import datetime
import glib
import logger
def get_local_time():
current_time = datetime.datetime.now().strftime("%H:%M")
logger.info("get_local_time(): %s",current_time)
return str(current_time)
def display_local_time():
logger.info("Current time is: %s", get_local_time())
return True
# call every minute
glib.timeout_add(60*1000, display_local_time)
Я использую метод Tkinter after(), который не "крадет игру" (например, модуль sched, который был представлен ранее), то есть позволяет другим вещам работать параллельно:
import Tkinter
def do_something1():
global n1
n1 += 1
if n1 == 6: # (Optional condition)
print "* do_something1() is done *"; return
# Do your stuff here
# ...
print "do_something1() "+str(n1)
tk.after(1000, do_something1)
def do_something2():
global n2
n2 += 1
if n2 == 6: # (Optional condition)
print "* do_something2() is done *"; return
# Do your stuff here
# ...
print "do_something2() "+str(n2)
tk.after(500, do_something2)
tk = Tkinter.Tk();
n1 = 0; n2 = 0
do_something1()
do_something2()
tk.mainloop()
do_something1() и do_something2() могут выполняться параллельно и на любой временной скорости. Здесь второй будет выполняться в два раза быстрее. Обратите внимание, что я использовал простой счетчик в качестве условия для завершения любой функции. Вы можете использовать любое другое ограничение, которое вам нравится, или нет, если вы выполняете функцию, пока программа не завершится (например, часы).
''' tracking number of times it prints'''
import threading
global timeInterval
count=0
def printit():
threading.Timer(timeInterval, printit).start()
print( "Hello, World!")
global count
count=count+1
print(count)
printit
if __name__ == "__main__":
timeInterval= int(input('Enter Time in Seconds:'))
printit()
Здесь адаптированная версия к коду из MestreLion. В дополнение к исходной функции этот код:
1) добавить first_interval, используемый для запуска таймера в определенное время (вызывающему нужно вычислить first_interval и пройти)
2) решить условие гонки в исходном коде. В исходном коде, если управляющий поток не смог отменить текущий таймер ("Остановить таймер и отменить выполнение действия таймеров". Это будет работать, только если таймер все еще находится на стадии ожидания ". Цитата из https://docs.python.org/2/library/threading.html), таймер будет работать бесконечно.
class RepeatedTimer(object):
def __init__(self, first_interval, interval, func, *args, **kwargs):
self.timer = None
self.first_interval = first_interval
self.interval = interval
self.func = func
self.args = args
self.kwargs = kwargs
self.running = False
self.is_started = False
def first_start(self):
try:
# no race-condition here because only control thread will call this method
# if already started will not start again
if not self.is_started:
self.is_started = True
self.timer = Timer(self.first_interval, self.run)
self.running = True
self.timer.start()
except Exception as e:
log_print(syslog.LOG_ERR, "timer first_start failed %s %s"%(e.message, traceback.format_exc()))
raise
def run(self):
# if not stopped start again
if self.running:
self.timer = Timer(self.interval, self.run)
self.timer.start()
self.func(*self.args, **self.kwargs)
def stop(self):
# cancel current timer in case failed it still OK
# if already stopped doesn't matter to stop again
if self.timer:
self.timer.cancel()
self.running = False