Timer.py

Go to the documentation of this file.

#!/usr/bin/env/python
# -*- coding: euc-jp -*-
 
 
 
 
import time
import threading
 
import OpenRTM_aist
 
 
 
class Timer:
  """
  """
 
  
  def __init__(self, interval):
    self._interval = interval
    self._running  = False
    self._runningMutex = threading.RLock()
    self._tasks = []
    self._taskMutex = threading.RLock()
    self._thread = threading.Thread(target=self.run)
    return
 
  
  def __del__(self):
    self._running = False
    try:
      self.join()
    except:
      pass
    self._thread = None
 
  def join(self):
    try:
      self._thread.join()
      self._thread = threading.Thread(target=self.run)
    except:
      pass
 
    return
 
  
  def run(self):
    while self._running:
      self.invoke()
      if self._interval.tv_sec != 0:
        time.sleep(self._interval.tv_sec)
      elif self._interval.tv_usec:
        time.sleep(self._interval.tv_usec/1000000.0)
    return 0
 
 
  
  def start(self):
    guard = OpenRTM_aist.ScopedLock(self._runningMutex)
    if not self._running:
      self._running = True
      self._thread.start()
    return
 
 
  
  def stop(self):
    guard = OpenRTM_aist.ScopedLock(self._runningMutex)
    if self._running:
      self._running = False
      self.join()
    return
 
 
  
  def invoke(self):
    guard = OpenRTM_aist.ScopedLock(self._taskMutex)
    for i in range(len(self._tasks)):
      self._tasks[i].remains = self._tasks[i].remains - self._interval
      if self._tasks[i].remains.sign() <= 0.0:
        self._tasks[i].remains = self._tasks[i].period
        self._tasks[i].listener.invoke()
    del guard
    return
 
  
  def registerListener(self, listener, tm):
    guard = OpenRTM_aist.ScopedLock(self._taskMutex)
    for i in range(len(self._tasks)):
      if self._tasks[i].listener == listener:
        self._tasks[i].period = tm
        self._tasks[i].remains = tm
        return listener
    self._tasks.append(self.Task(listener, tm))
    return listener
 
 
  
  def registerListenerObj(self, obj, cbf, tm):
    return self.registerListener(OpenRTM_aist.ListenerObject(obj, cbf), tm)
 
 
  
  def registerListenerFunc(self, cbf, tm):
    return self.registerListener(OpenRTM_aist.ListenerFunc(cbf), tm)
 
 
  
  def unregisterListener(self, id):
    guard = OpenRTM_aist.ScopedLock(self._taskMutex)
    len_ = len(self._tasks)
    for i in range(len_):
      idx = (len_ - 1) - i
      if self._tasks[idx].listener == id:
        del self._tasks[idx]
        return True
    return False
 
 
  
  class Task:
    def __init__(self, lb, tm):
      self.listener = lb
      self.period = tm
      self.remains = tm
      return