boswatch.timer.RepeatedTimer Class Reference

Public Member Functions
	__init__ (self, interval, targetFunction, *args, **kwargs)
	Create a new instance of the RepeatedTimer.
	start (self)
	Start a new timer worker thread.
	stop (self)
	Stop the timer worker thread.
	isRunning (self)
	Property for repeatedTimer running state.
	restTime (self)
	Property to get remaining time till next call.

Data Fields
	overdueCount
	lostEvents
	restTime

Protected Member Functions
	_target (self)
	Runs the target function with his arguments in own thread.

Protected Attributes
	_interval
	_function
	_args
	_kwargs
	_start
	_isRunning
	_event
	_thread

Constructor & Destructor Documentation

__init__()

boswatch.timer.RepeatedTimer.__init__	(		self,
			interval,
			targetFunction,
		*	args,
		**	kwargs
	)

Create a new instance of the RepeatedTimer.

Parameters

interval	interval in sec. to recall target function
targetFunction	function to call on timer event
*args	arguments for the called function
*kwargs	keyword arguments for the called function

    def __init__(self, interval, targetFunction, *args, **kwargs):
        r"""!Create a new instance of the RepeatedTimer
 
        @param interval: interval in sec. to recall target function
        @param targetFunction: function to call on timer event
        @param *args: arguments for the called function
        @param *kwargs: keyword arguments for the called function
        """
        self._interval = interval
        self._function = targetFunction
        self._args = args
        self._kwargs = kwargs
        self._start = 0
        self.overdueCount = 0
        self.lostEvents = 0
        self._isRunning = False
        self._event = Event()
        self._thread = None
 

Member Function Documentation

start()

boswatch.timer.RepeatedTimer.start

(

self

)

Start a new timer worker thread.

Returns

True or False

    def start(self):
        r"""!Start a new timer worker thread
 
        @return True or False"""
        if self._thread is None:
            self._event.clear()
            self._thread = Thread(target=self._target)
            self._thread.name = "RepTim(" + str(self._interval) + ")"
            self._thread.daemon = True  # start as daemon (thread dies if main program ends)
            self._thread.start()
            logging.debug("start repeatedTimer: %s", self._thread.name)
            return True
        logging.debug("repeatedTimer always started")
        return True
 

stop()

boswatch.timer.RepeatedTimer.stop

(

self

)

Stop the timer worker thread.

Returns

True or False

    def stop(self):
        r"""!Stop the timer worker thread
 
        @return True or False"""
        if self._thread is not None:
            logging.debug("stop repeatedTimer: %s", self._thread.name)
            self._event.set()
            if self._thread is not None:
                self._thread.join()
            return True
        logging.warning("repeatedTimer always stopped")
        return True
 

_target()

boswatch.timer.RepeatedTimer._target (   self )

protected

Runs the target function with his arguments in own thread.

    def _target(self):
        r"""!Runs the target function with his arguments in own thread"""
        self._start = time.time()
        while not self._event.wait(self.restTime):
            logging.debug("work")
            startTime = time.time()
 
            try:
                self._function(*self._args, **self._kwargs)
            except:  # pragma: no cover
                logging.exception("target throws an exception")
 
            runTime = time.time() - startTime
            if runTime < self._interval:
                logging.debug("ready after: %0.3f sec. - next call in: %0.3f sec.", runTime, self.restTime)
            else:
                lostEvents = int(runTime / self._interval)
                logging.warning("timer overdue! interval: %0.3f sec. - runtime: %0.3f sec. - "
                                "%d events lost - next call in: %0.3f sec.", self._interval, runTime, lostEvents, self.restTime)
                self.lostEvents += lostEvents
                self.overdueCount += 1
        logging.debug("repeatedTimer thread stopped: %s", self._thread.name)
        self._thread = None  # set to none after leave teh thread (running recognize)
 

isRunning()

boswatch.timer.RepeatedTimer.isRunning

(

self

)

Property for repeatedTimer running state.

    def isRunning(self):
        r"""!Property for repeatedTimer running state"""
        if self._thread:
            return True
        return False
 

restTime()

boswatch.timer.RepeatedTimer.restTime

(

self

)

Property to get remaining time till next call.

    def restTime(self):
        r"""!Property to get remaining time till next call"""
        return self._interval - ((time.time() - self._start) % self._interval)

Field Documentation

_interval

boswatch.timer.RepeatedTimer._interval

protected

_function

boswatch.timer.RepeatedTimer._function

protected

_args

boswatch.timer.RepeatedTimer._args

protected

_kwargs

boswatch.timer.RepeatedTimer._kwargs

protected

_start

boswatch.timer.RepeatedTimer._start

protected

overdueCount

boswatch.timer.RepeatedTimer.overdueCount

lostEvents

boswatch.timer.RepeatedTimer.lostEvents

_isRunning

boswatch.timer.RepeatedTimer._isRunning

protected

_event

boswatch.timer.RepeatedTimer._event

protected

_thread

boswatch.timer.RepeatedTimer._thread

protected

restTime

boswatch.timer.RepeatedTimer.restTime