По некоторым соображениям решил продолжить тему. У предложенного таймера есть недостаток - он “плывёт”. То есть, накапливает погрешности, вызванные несовершенством ОС и, возможно, также вызванные GIL. Так как (на первый взгляд) это плавание легко удалить, то я сделаю это.
import threading
import time
class Timer:
def __init__(self, function):
self._func = function
self._isRunning = False
def start(self, period, startWithJerk=False):
if self._isRunning == False:
self._currentTime = time.time()
self._start_new_period(period)
if startWithJerk: self._func()
self._period = period
self._isRunning = True
def _start_new_period(self, period):
self._timer = threading.Timer(period, function=self._on_timeout)
self._timer.start()
def _on_timeout(self):
if self._isRunning:
self._currentTime += self._period
period = self._period - (time.time() - self._currentTime)
while period <= 0:
period += self._period
self._start_new_period(period)
self._func()
def is_running(self):
return self._isRunning
def stop(self):
if self._isRunning:
self._timer.cancel()
self._timer.join()
self._timer = None
self._isRunning = False
if __name__ == '__main__':
import wx
import datetime
class MyFrame(wx.Frame):
def __init__(self):
wx.Frame.__init__(self, None, title="Timer with GUI",
size=(300, 600))
panel = wx.Panel(self)
startBtn = wx.Button(panel, -1, "Start timer")
stopBtn = wx.Button(panel, -1, "Stop timer")
clearBtn = wx.Button(panel, -1, "Clear")
self.log = wx.TextCtrl(panel, -1, "",
style=wx.TE_RICH|wx.TE_MULTILINE)
inner = wx.BoxSizer(wx.HORIZONTAL)
inner.Add(startBtn, 0, wx.RIGHT, 15)
inner.Add(stopBtn, 0, wx.RIGHT, 15)
inner.Add(clearBtn, 0, wx.RIGHT, 15)
main = wx.BoxSizer(wx.VERTICAL)
main.Add(inner, 0, wx.ALL, 5)
main.Add(self.log, 1, wx.EXPAND|wx.ALL, 5)
panel.SetSizer(main)
self.Bind(wx.EVT_BUTTON, self.on_start_button, startBtn)
self.Bind(wx.EVT_BUTTON, self.on_stop_button, stopBtn)
self.Bind(wx.EVT_BUTTON, self.on_clear_button, clearBtn)
self.Bind(wx.EVT_CLOSE, self.on_close_window)
self.timer = Timer(self.ret_func)
def on_start_button(self, event):
self.timer.start(0.1)
def on_stop_button(self, event):
self.timer.stop()
def on_clear_button(self, event):
self.log.Clear()
def ret_func(self):
wx.CallAfter(self._ret_func)
def _ret_func(self):
if self.log.GetNumberOfLines() > 50:
self.log.Clear()
self.log.AppendText("timer: %s\n" %
datetime.datetime(2000, 1, 1).now())
def on_close_window(self, event):
self.timer.stop()
self.Destroy()
app = wx.PySimpleApp()
frm = MyFrame()
frm.Show()
app.MainLoop()
В моих тестах “плавание” прекратилось. Но проверка была на одном компьютере и с одним таймером. Пока мне этого достаточно.
Вместо циклического создания threading.Timer можно использовать один поток, в котором будет цикл со sleep. Но у threading.Timer есть полезный метод cancel(), позволяющий быстро остановить таймер. С join не выйдет быстро остановить таймер - надо будет дождаться окончания sleep.
Цикл со sleep можно использовать в multiprocessing.Process, если реализовывать таймер на нём. Вероятно, там уместно будет использовать terminate(). Но процессы не разделяют данные, потому будет трудно передать в качестве функции процесса метод из какого-нибудь объекта родительского процесса. Как я понимаю, асинхронности не получится.