git.maemo.org Git - theonering/blob - src/gvoice/state_machine.py

   1 #!/usr/bin/env python
   2
   3 import Queue
   4 import threading
   5 import logging
   6
   7 import gobject
   8
   9 import util.algorithms as algorithms
  10 import util.coroutines as coroutines
  11
  12 _moduleLogger = logging.getLogger("gvoice.state_machine")
  13
  14
  15 def _to_milliseconds(**kwd):
  16         if "milliseconds" in kwd:
  17                 return kwd["milliseconds"]
  18         elif "seconds" in kwd:
  19                 return kwd["seconds"] * 1000
  20         elif "minutes" in kwd:
  21                 return kwd["minutes"] * 1000 * 60
  22         raise KeyError("Unknown arg: %r" % kwd)
  23
  24
  25 class StateMachine(object):
  26
  27         STATE_ACTIVE = "active"
  28         STATE_IDLE = "idle"
  29         STATE_DND = "dnd"
  30
  31         _ACTION_UPDATE = "update"
  32         _ACTION_RESET = "reset"
  33         _ACTION_STOP = "stop"
  34
  35         _INITIAL_ACTIVE_PERIOD = int(_to_milliseconds(seconds=5))
  36         _FINAL_ACTIVE_PERIOD = int(_to_milliseconds(minutes=2))
  37         _IDLE_PERIOD = int(_to_milliseconds(minutes=10))
  38         _INFINITE_PERIOD = -1
  39
  40         _IS_DAEMON = True
  41
  42         def __init__(self, initItems, updateItems):
  43                 self._initItems = initItems
  44                 self._updateItems = updateItems
  45
  46                 self._actions = Queue.Queue()
  47                 self._state = self.STATE_ACTIVE
  48                 self._timeoutId = None
  49                 self._thread = None
  50                 self._currentPeriod = self._INITIAL_ACTIVE_PERIOD
  51                 self._set_initial_period()
  52
  53         def start(self):
  54                 assert self._thread is None
  55                 self._thread = threading.Thread(target=self._run)
  56                 self._thread.setDaemon(self._IS_DAEMON)
  57                 self._thread.start()
  58
  59         def stop(self):
  60                 if self._thread is not None:
  61                         self._actions.put(self._ACTION_STOP)
  62                         self._thread = None
  63                 else:
  64                         _moduleLogger.info("Stopping an already stopped state machine")
  65
  66         def set_state(self, state):
  67                 self._state = state
  68                 self.reset_timers()
  69
  70         def get_state(self):
  71                 return self._state
  72
  73         def reset_timers(self):
  74                 self._actions.put(self._ACTION_RESET)
  75
  76         @coroutines.func_sink
  77         def request_reset_timers(self, args):
  78                 self.reset_timers()
  79
  80         def _run(self):
  81                 for item in self._initItems:
  82                         try:
  83                                 item.update()
  84                         except Exception:
  85                                 _moduleLogger.exception("Initial update failed for %r" % item)
  86
  87                 # empty the task queue
  88                 actions = list(algorithms.itr_available(self._actions, initiallyBlock = False))
  89                 self._schedule_update()
  90                 if len(self._updateItems) == 0:
  91                         self.stop()
  92
  93                 while True:
  94                         # block till we get a task, or get all the tasks if we were late
  95                         actions = list(algorithms.itr_available(self._actions, initiallyBlock = True))
  96
  97                         if self._ACTION_STOP in actions:
  98                                 self._stop_update()
  99                                 break
 100                         elif self._ACTION_RESET in actions:
 101                                 self._reset_timers()
 102                         elif self._ACTION_UPDATE in actions:
 103                                 for item in self._updateItems:
 104                                         try:
 105                                                 item.update(force=True)
 106                                         except Exception:
 107                                                 _moduleLogger.exception("Update failed for %r" % item)
 108                                 self._schedule_update()
 109
 110         def _set_initial_period(self):
 111                 self._currentPeriod = self._INITIAL_ACTIVE_PERIOD / 2 # We will double it later
 112
 113         def _reset_timers(self):
 114                 self._stop_update()
 115                 self._set_initial_period()
 116                 self._schedule_update()
 117
 118         def _schedule_update(self):
 119                 nextTimeout = self._calculate_step(self._state, self._currentPeriod)
 120                 nextTimeout = int(nextTimeout)
 121                 if nextTimeout != self._INFINITE_PERIOD:
 122                         self._timeoutId = gobject.timeout_add(nextTimeout, self._on_timeout)
 123                 self._currentPeriod = nextTimeout
 124
 125         def _stop_update(self):
 126                 if self._timeoutId is None:
 127                         return
 128                 gobject.source_remove(self._timeoutId)
 129                 self._timeoutId = None
 130
 131         def _on_timeout(self):
 132                 self._actions.put(self._ACTION_UPDATE)
 133                 return False # do not continue
 134
 135         @classmethod
 136         def _calculate_step(cls, state, period):
 137                 if state == cls.STATE_ACTIVE:
 138                         return min(period * 2, cls._FINAL_ACTIVE_PERIOD)
 139                 elif state == cls.STATE_IDLE:
 140                         return cls._IDLE_PERIOD
 141                 elif state == cls.STATE_DND:
 142                         return cls._INFINITE_PERIOD
 143                 else:
 144                         raise RuntimeError("Unknown state: %r" % (state, ))