Python threading.timer - повторять функцию каждые 'n' секунд

У меня проблемы с таймером Python и буду очень признателен за советы или помощь:D

Я не слишком хорошо знаю, как работают потоки, но я просто хочу запускать функцию каждые 0,5 секунды и иметь возможность запускать, останавливать и сбрасывать таймер.

Тем не менее, я продолжаю получать RuntimeError: threads can only be started once когда я выполню threading.timer.start() дважды. Есть ли обходной путь для этого? Я пытался подать заявку threading.timer.cancel() перед каждым стартом.

Псевдокод:

t=threading.timer(0.5,function)
while True:
    t.cancel()
    t.start()

16 ответов

Решение

Лучший способ - запустить поток таймера один раз. В вашей ветке таймера вы должны написать следующее

class MyThread(Thread):
    def __init__(self, event):
        Thread.__init__(self)
        self.stopped = event

    def run(self):
        while not self.stopped.wait(0.5):
            print("my thread")
            # call a function

В коде, который запустил таймер, вы можете затем set остановленное событие, чтобы остановить таймер.

stopFlag = Event()
thread = MyThread(stopFlag)
thread.start()
# this will stop the timer
stopFlag.set()

Немного улучшив Ганса Затем ответим, мы можем просто создать подкласс функции Timer. Следующее становится нашим полным кодом "таймера повтора", и его можно использовать в качестве замены для threading.Timer со всеми одинаковыми аргументами:

from threading import Timer

class RepeatTimer(Timer):
    def run(self):
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)

Пример использования:

def dummyfn(msg="foo"):
    print(msg)

timer = RepeatTimer(1, dummyfn)
timer.start()
time.sleep(5)
timer.cancel()

производит следующий вывод:

foo
foo
foo
foo

а также

timer = RepeatTimer(1, dummyfn, args=("bar",))
timer.start()
time.sleep(5)
timer.cancel()

производит

bar
bar
bar
bar

Из Эквивалента setInterval в python:

import threading

def setInterval(interval):
    def decorator(function):
        def wrapper(*args, **kwargs):
            stopped = threading.Event()

            def loop(): # executed in another thread
                while not stopped.wait(interval): # until stopped
                    function(*args, **kwargs)

            t = threading.Thread(target=loop)
            t.daemon = True # stop if the program exits
            t.start()
            return stopped
        return wrapper
    return decorator

Использование:

@setInterval(.5)
def function():
    "..."

stop = function() # start timer, the first call is in .5 seconds
stop.set() # stop the loop
stop = function() # start new timer
# ...
stop.set() 

Или вот та же функциональность, но в качестве отдельной функции вместо декоратора:

cancel_future_calls = call_repeatedly(60, print, "Hello, World")
# ...
cancel_future_calls() 

Вот как это сделать без использования потоков.

Используя таймеры

from threading import Timer,Thread,Event


class perpetualTimer():

   def __init__(self,t,hFunction):
      self.t=t
      self.hFunction = hFunction
      self.thread = Timer(self.t,self.handle_function)

   def handle_function(self):
      self.hFunction()
      self.thread = Timer(self.t,self.handle_function)
      self.thread.start()

   def start(self):
      self.thread.start()

   def cancel(self):
      self.thread.cancel()

def printer():
    print 'ipsem lorem'

t = perpetualTimer(5,printer)
t.start()

это может быть остановлено t.cancel()

В интересах предоставления правильного ответа, используя Timer в соответствии с запросом OP, я улучшу ответ swapnil jariwala:

from threading import Timer
import time


class InfiniteTimer():
    """A Timer class that does not stop, unless you want it to."""

    def __init__(self, seconds, target):
        self._should_continue = False
        self.is_running = False
        self.seconds = seconds
        self.target = target
        self.thread = None

    def _handle_target(self):
        self.is_running = True
        self.target()
        self.is_running = False
        self._start_timer()

    def _start_timer(self):
        if self._should_continue: # Code could have been running when cancel was called.
            self.thread = Timer(self.seconds, self._handle_target)
            self.thread.start()

    def start(self):
        if not self._should_continue and not self.is_running:
            self._should_continue = True
            self._start_timer()
        else:
            print("Timer already started or running, please wait if you're restarting.")

    def cancel(self):
        if self.thread is not None:
            self._should_continue = False # Just in case thread is running and cancel fails.
            self.thread.cancel()
        else:
            print("Timer never started or failed to initialize.")


def tick():
    print('ipsem lorem')

# Example Usage
t = InfiniteTimer(0.5, tick)
t.start()

Время импорта является необязательным без использования примера.

Я изменил некоторый код в коде swapnil-jariwala, чтобы сделать немного консольных часов.

from threading import Timer, Thread, Event
from datetime import datetime


class perpetualTimer():

    def __init__(self, t, hFunction):
        self.t = t
        self.hFunction = hFunction
        self.thread = Timer(self.t, self.handle_function)

    def handle_function(self):
        self.hFunction()
        self.thread = Timer(self.t, self.handle_function)
        self.thread.start()

    def start(self):
        self.thread.start()

    def cancel(self):
        self.thread.cancel()


def printer():
    tempo = datetime.today()
    print("{}:{}:{}".format(tempo.hour, tempo.minute, tempo.second))


t = perpetualTimer(1, printer)
t.start()

ВЫХОД

>>> 11:39:11
11:39:12
11:39:13
11:39:14
11:39:15
11:39:16
...

Таймер с графическим интерфейсом tkinter

Этот код помещает таймер часов в маленькое окно с tkinter

from threading import Timer, Thread, Event
from datetime import datetime
import tkinter as tk

app = tk.Tk()
lab = tk.Label(app, text="Timer will start in a sec")
lab.pack()


class perpetualTimer():

    def __init__(self, t, hFunction):
        self.t = t
        self.hFunction = hFunction
        self.thread = Timer(self.t, self.handle_function)

    def handle_function(self):
        self.hFunction()
        self.thread = Timer(self.t, self.handle_function)
        self.thread.start()

    def start(self):
        self.thread.start()

    def cancel(self):
        self.thread.cancel()


def printer():
    tempo = datetime.today()
    clock = "{}:{}:{}".format(tempo.hour, tempo.minute, tempo.second)
    try:
        lab['text'] = clock
    except RuntimeError:
        exit()


t = perpetualTimer(1, printer)
t.start()
app.mainloop()

Пример игры с карточками (вроде)

from threading import Timer, Thread, Event
from datetime import datetime


class perpetualTimer():

    def __init__(self, t, hFunction):
        self.t = t
        self.hFunction = hFunction
        self.thread = Timer(self.t, self.handle_function)

    def handle_function(self):
        self.hFunction()
        self.thread = Timer(self.t, self.handle_function)
        self.thread.start()

    def start(self):
        self.thread.start()

    def cancel(self):
        self.thread.cancel()


x = datetime.today()
start = x.second


def printer():
    global questions, counter, start
    x = datetime.today()
    tempo = x.second
    if tempo - 3 > start:
        show_ans()
    #print("\n{}:{}:{}".format(tempo.hour, tempo.minute, tempo.second), end="")
    print()
    print("-" + questions[counter])
    counter += 1
    if counter == len(answers):
        counter = 0


def show_ans():
    global answers, c2
    print("It is {}".format(answers[c2]))
    c2 += 1
    if c2 == len(answers):
        c2 = 0


questions = ["What is the capital of Italy?",
             "What is the capital of France?",
             "What is the capital of England?",
             "What is the capital of Spain?"]

answers = "Rome", "Paris", "London", "Madrid"

counter = 0
c2 = 0
print("Get ready to answer")
t = perpetualTimer(3, printer)
t.start()

выход:

Get ready to answer
>>> 
-What is the capital of Italy?
It is Rome

-What is the capital of France?
It is Paris

-What is the capital of England?
...

Я немного опоздал, но вот мои два цента:

Вы можете повторно использовать threading.Timerобъект, вызвав его .run()метод несколько раз, например:

      class SomeClassThatNeedsATimer:
    def __init__(...):
        self.timer = threading.Timer(interval, self.on_timer)
        self.timer.start()

    def on_timer(self):
        print('On timer')
        self.timer.run()

Я должен был сделать это для проекта. В итоге я запустил отдельный поток для функции

t = threading.Thread(target =heartbeat, args=(worker,))
t.start()

**** сердцебиение - моя функция, работник - один из моих аргументов ****

внутри моей функции сердцебиения:

def heartbeat(worker):

    while True:
        time.sleep(5)
        #all of my code

Поэтому, когда я запускаю поток, функция будет несколько раз ждать 5 секунд, запускать весь мой код и делать это бесконечно. Если вы хотите убить процесс, просто убейте поток.

В дополнение к приведенным выше отличным ответам с использованием потоков, если вам нужно использовать свой основной поток или предпочитаете асинхронный подход - я обернул короткий класс вокруг класса таймера aio_timers (чтобы включить повторение)

       import asyncio
from aio_timers import Timer

class RepeatingAsyncTimer():
    def __init__(self, interval, cb, *args, **kwargs):
        self.interval = interval
        self.cb = cb
        self.args = args
        self.kwargs = kwargs
        self.aio_timer = None
        self.start_timer()
    
    def start_timer(self):
        self.aio_timer = Timer(delay=self.interval, 
                               callback=self.cb_wrapper, 
                               callback_args=self.args, 
                               callback_kwargs=self.kwargs
                              )
    
    def cb_wrapper(self, *args, **kwargs):
        self.cb(*args, **kwargs)
        self.start_timer()


from time import time
def cb(timer_name):
    print(timer_name, time())

print(f'clock starts at: {time()}')
timer_1 = RepeatingAsyncTimer(interval=5, cb=cb, timer_name='timer_1')
timer_2 = RepeatingAsyncTimer(interval=10, cb=cb, timer_name='timer_2')

часы начинаются с: 1602438840.9690785

timer_ 1 1602438845.980087

таймер_ 2 1602438850.9806316

таймер_ 1 1602438850.9808934

таймер_ 1 1602438855.9863033

таймер_ 2 1602438860.9868324

таймер_ 1 1602438860.9876585

Я придумал другое решение с классом SingleTon. Подскажите пожалуйста, есть ли здесь утечка памяти.

import time,threading

class Singleton:
  __instance = None
  sleepTime = 1
  executeThread = False

  def __init__(self):
     if Singleton.__instance != None:
        raise Exception("This class is a singleton!")
     else:
        Singleton.__instance = self

  @staticmethod
  def getInstance():
     if Singleton.__instance == None:
        Singleton()
     return Singleton.__instance


  def startThread(self):
     self.executeThread = True
     self.threadNew = threading.Thread(target=self.foo_target)
     self.threadNew.start()
     print('doing other things...')


  def stopThread(self):
     print("Killing Thread ")
     self.executeThread = False
     self.threadNew.join()
     print(self.threadNew)


  def foo(self):
     print("Hello in " + str(self.sleepTime) + " seconds")


  def foo_target(self):
     while self.executeThread:
        self.foo()
        print(self.threadNew)
        time.sleep(self.sleepTime)

        if not self.executeThread:
           break


sClass = Singleton()
sClass.startThread()
time.sleep(5)
sClass.getInstance().stopThread()

sClass.getInstance().sleepTime = 2
sClass.startThread()

В интересах предоставления правильного ответа с использованием Timer по запросу OP я улучшу /questions/25587178/python-threadingtimer-povtoryat-funktsiyu-kazhdyie-n-sekund/25587209#25587209

      from threading import Timer


class InfiniteTimer():
  """A Timer class that does not stop, unless you want it to."""

  def __init__(self, seconds, target, args=[], kwargs=dict()):
    self._should_continue = False
    self.is_running = False
    self.seconds = seconds
    self.target = target
    self.args = args
    self.kwargs = kwargs
    self.thread = None

  def _handle_target(self):
    self.is_running = True
    self.target(*self.args, **self.kwargs)
    self.is_running = False
    self._start_timer()

  def _start_timer(self):
    if self._should_continue:  # Code could have been running when cancel was called.
      self.thread = Timer(
        self.seconds,
        self._handle_target,
      )
      self.thread.start()

  def start(self):
    if not self._should_continue and not self.is_running:
      self._should_continue = True
      self._start_timer()
    else:
      print(
        "Timer already started or running, please wait if you're restarting.")

  def cancel(self):
    if self.thread is not None:
      self._should_continue = False  # Just in case thread is running and cancel fails.
      self.thread.cancel()
    else:
      print("Timer never started or failed to initialize.")


def tick(i):
  print('ipsem lorem', i)


# Example Usage
t = InfiniteTimer(0.5, tick, kwargs=dict(i="i"))
t.start()

Мне нравится ответ right2clicky, особенно в том, что он не требует разрушения потока и создания нового каждый раз, когда срабатывает таймер. Кроме того, легко переопределить создание класса с обратным вызовом таймера, который периодически вызывается. Это мой обычный вариант использования:

class MyClass(RepeatTimer):
    def __init__(self, period):
        super().__init__(period, self.on_timer)

    def on_timer(self):
        print("Tick")


if __name__ == "__main__":
    mc = MyClass(1)
    mc.start()
    time.sleep(5)
    mc.cancel()

Я реализовал класс, который работает как таймер.

Я оставляю ссылку здесь на тот случай, если она кому-нибудь понадобится: https://github.com/ivanhalencp/python/tree/master/xTimer

from threading import Timer
def TaskManager():
    #do stuff
    t = Timer( 1, TaskManager )
    t.start()

TaskManager()

Вот небольшой пример, он поможет лучше понять, как он работает. Функция taskManager() в конце создает отложенный вызов функции самостоятельно.

Попробуйте изменить переменную "dalay", и вы сможете увидеть разницу

from threading import Timer, _sleep

# ------------------------------------------
DATA = []
dalay = 0.25 # sec
counter = 0
allow_run = True
FIFO = True

def taskManager():

    global counter, DATA, delay, allow_run
    counter += 1

    if len(DATA) > 0:
        if FIFO:
            print("["+str(counter)+"] new data: ["+str(DATA.pop(0))+"]")
        else:
            print("["+str(counter)+"] new data: ["+str(DATA.pop())+"]")

    else:
        print("["+str(counter)+"] no data")

    if allow_run:
        #delayed method/function call to it self
        t = Timer( dalay, taskManager )
        t.start()

    else:
        print(" END task-manager: disabled")

# ------------------------------------------
def main():

    DATA.append("data from main(): 0")
    _sleep(2)
    DATA.append("data from main(): 1")
    _sleep(2)


# ------------------------------------------
print(" START task-manager:")
taskManager()

_sleep(2)
DATA.append("first data")

_sleep(2)
DATA.append("second data")

print(" START main():")
main()
print(" END main():")

_sleep(2)
DATA.append("last data")

allow_run = False

Это альтернативная реализация, использующая функцию вместо класса. Вдохновленный @Andrew Wilkins выше.

Поскольку ожидание более точно, чем сон (учитывается время выполнения функции):

import threading

PING_ON = threading.Event()

def ping():
  while not PING_ON.wait(1):
    print("my thread %s" % str(threading.current_thread().ident))

t = threading.Thread(target=ping)
t.start()

sleep(5)
PING_ON.set()

Это пример кода для непрерывной работы таймера. Просто создайте новый таймер при истощении и вызовите ту же функцию. Не лучший способ сделать это, но можно сделать и так.

      import threading
import time


class ContinousTimer():
    def __init__(self):
        self.timer = None

    def run(self, msg='abc'):
        print(msg)

        self.timer = threading.Timer(interval=2, function=self.run, args=(msg, ))
        self.timer.start()


if __name__ == "__main__":
    t = ContinousTimer()
    try:
        t.run(msg="Hello")
        while True:
            time.sleep(0.1)
    except KeyboardInterrupt:
        # Cancel Timer
        t.timer.cancel()
Другие вопросы по тегам