onoff/common.py

   1 import enum
   2 import logging
   3
   4 from .gobject import ScheduledFunction
   5
   6 logger = logging.getLogger("onoff.common")
   7
   8
   9 class OnoffState(enum.Flag):
  10     inactive = 0
  11     active = 1
  12     deactivating = 2
  13     activating = 3
  14     transition = 2
  15
  16
  17 class OnoffDevice(object):
  18     """A device is a thing with two states, that can be asked to transition
  19     from either state to the other. It can signal state changes to interested
  20     parties.
  21
  22     @type notify: {int -> None}
  23     @ivar notify: A set of functions taking a changed state.
  24
  25     @type state: int
  26     @ivar state: is a read-only attribute to retrieve the current state
  27     """
  28     def __init__(self):
  29         self.notify = set()
  30
  31     def changestate(self, state):
  32         """Tell interested parties that the state has changed to the given
  33         state."""
  34         for func in self.notify:
  35             func(state)
  36
  37     def activate(self):
  38         """Ask the device to power on."""
  39         raise NotImplementedError
  40
  41     def deactivate(self):
  42         """Ask the device to power off."""
  43         raise NotImplementedError
  44
  45     def close(self):
  46         """Release resources acquired by the constructor."""
  47         pass
  48
  49 class InvertedDevice(OnoffDevice):
  50     """A device that swaps active and inactive states of a give device."""
  51     def __init__(self, device):
  52         OnoffDevice.__init__(self)
  53         self.device = device
  54         self.device.activate()
  55         self.device.notify.add(self.changestate)
  56
  57     def changestate(self, state):
  58         OnoffDevice.changestate(self, state ^ OnoffState.active)
  59
  60     @property
  61     def state(self):
  62         return self.device.state ^ OnoffState.active
  63
  64     def activate(self):
  65         self.device.deactivate()
  66
  67     def deactivate(self):
  68         self.device.activate()
  69
  70     def close(self):
  71         self.device.notify.remove(self.changestate)
  72         self.device.close()
  73
  74 class ThrottledDevice(OnoffDevice):
  75     """A device that delays the activation signal and the actual deactivation
  76     by a fixed amounts of time. This limits the rate of state transitions to
  77     two per offdelay."""
  78     def __init__(self, device, ondelay, offdelay):
  79         """
  80         @type device: OnoffDevice
  81         @type ondelay: int or float
  82         @param ondelay: delay the report of OnoffState.active by this many
  83                         seconds
  84         @type offdelay: int or float
  85         @param offdelay: delay the actual deactivation by this many seconds
  86         """
  87         OnoffDevice.__init__(self)
  88         self.device = device
  89         self.ondelay = ondelay
  90         self.offdelay = offdelay
  91         self.desired_state = OnoffState.inactive
  92         self.transition = None
  93         self.device.notify.add(self.changestate)
  94
  95     @property
  96     def state(self):
  97         if self.transition is None:
  98             return self.device.state
  99         return self.desired_state | OnoffState.transition
 100
 101     def _schedule_transition(self, delay, func):
 102         assert self.transition is None
 103         self.transition = ScheduledFunction(delay, func)
 104
 105     def _cancel_transition(self):
 106         assert self.transition is not None
 107         self.transition.cancel()
 108         self.transition = None
 109
 110     def changestate(self, state):
 111         if state != OnoffState.active:
 112             OnoffDevice.changestate(self, state)
 113         else:
 114             if self.desired_state == OnoffState.inactive:
 115                 logger.warning("device became active but we want inactive," +
 116                                "suppresing signal")
 117             elif self.transition is None:
 118                 logger.debug("scheduling report of activation in %.1fs",
 119                              self.ondelay)
 120                 self._schedule_transition(self.ondelay, self._report_active)
 121             else:
 122                 logger.debug("suppressing duplicate activation signal")
 123
 124     def _report_active(self):
 125         assert self.desired_state == OnoffState.active
 126         assert self.transition is not None
 127         self.transition = None
 128         logger.debug("delivering activation signal")
 129         OnoffDevice.changestate(self, OnoffState.active)
 130
 131     def activate(self):
 132         if self.desired_state == OnoffState.inactive and \
 133                 self.transition is not None:
 134             logger.debug("cancelling pending deactivation")
 135             self._cancel_transition()
 136             curstate = self.device.state
 137             if curstate == OnoffState.active:
 138                 self.desired_state = OnoffState.active
 139                 OnoffDevice.changestate(self, OnoffState.active)
 140                 return
 141             logger.warning("device should be active during delayed " +
 142                            "deactivation, but is in state %s", curstate.name)
 143         self.desired_state = OnoffState.active
 144         self.device.activate()
 145         self.changestate(self.device.state)
 146
 147     def _do_stop(self):
 148         assert self.desired_state == OnoffState.inactive
 149         assert self.transition is not None
 150         self.transition = None
 151         logger.debug("actually deactivating")
 152         self.device.deactivate()
 153
 154     def deactivate(self):
 155         if self.desired_state == OnoffState.active and \
 156                 self.transition is not None:
 157             logger.debug("cancelling pending activation report")
 158             self._cancel_transition()
 159         self.desired_state = OnoffState.inactive
 160         if self.transition is None:
 161             logger.debug("scheduling actual deactivate in %.1fs",
 162                          self.offdelay)
 163             self._schedule_transition(self.offdelay, self._do_stop)
 164             self.changestate(self.state)
 165         else:
 166             logger.debug("not issuing deactivate due to pending deactivate")
 167
 168     def close(self):
 169         if self.transition is not None:
 170             logger.info("cancelling pending transition")
 171             self._cancel_transition()
 172             if self.desired_state == OnoffState.inactive:
 173                 logger.info("invoking pending deactivate early during close")
 174                 self.device.deactivate()
 175         self.device.notify.remove(self.changestate)
 176         self.device.close()