boswatch.timer.RepeatedTimer Class Reference

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

Data Fields
	overdueCount
	lostEvents

Protected Member Functions
def	_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__()

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

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

    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()

def boswatch.timer.RepeatedTimer.start

(

self

)

Start a new timer worker thread.

    @return 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()

def boswatch.timer.RepeatedTimer.stop

(

self

)

Stop the timer worker thread.

    @return 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()

def 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()

def 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()

def 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