-"""
-Defined states:
- * 0: The device is inactive.
- * ST_ACTIVE|ST_TRANSITION: The device is transitioning from inactive to active.
- * ST_ACTIVE: The device is active.
- * ST_TRANSITION: The device is transitioning from active to inactive.
-"""
-
-ST_ACTIVE = 1
-ST_TRANSITION = 2
+import enum
+import logging
+
+from .gobject import ScheduledFunction
+
+logger = logging.getLogger("onoff.common")
+
+
+class OnoffState(enum.Flag):
+ inactive = 0
+ active = 1
+ deactivating = 2
+ activating = 3
+ transition = 2
+
+
+class OnoffDevice(object):
+ """A device is a thing with two states, that can be asked to transition
+ from either state to the other. It can signal state changes to interested
+ parties.
+
+ @type notify: {int -> None}
+ @ivar notify: A set of functions taking a changed state.
+
+ @type state: int
+ @ivar state: is a read-only attribute to retrieve the current state
+ """
+ def __init__(self):
+ self.notify = set()
+
+ def changestate(self, state):
+ """Tell interested parties that the state has changed to the given
+ state."""
+ for func in self.notify:
+ func(state)
+
+ def activate(self):
+ """Ask the device to power on."""
+ raise NotImplementedError
+
+ def deactivate(self):
+ """Ask the device to power off."""
+ raise NotImplementedError
+
+ def close(self):
+ """Release resources acquired by the constructor."""
+ pass
+
+class InvertedDevice(OnoffDevice):
+ """A device that swaps active and inactive states of a give device."""
+ def __init__(self, device):
+ OnoffDevice.__init__(self)
+ self.device = device
+ self.device.activate()
+ self.device.notify.add(self.changestate)
+
+ def changestate(self, state):
+ OnoffDevice.changestate(self, state ^ OnoffState.active)
+
+ @property
+ def state(self):
+ return self.device.state ^ OnoffState.active
+
+ def activate(self):
+ self.device.deactivate()
+
+ def deactivate(self):
+ self.device.activate()
+
+ def close(self):
+ self.device.notify.remove(self.changestate)
+ self.device.close()
+
+class ThrottledDevice(OnoffDevice):
+ """A device that delays the activation signal and the actual deactivation
+ by a fixed amounts of time. This limits the rate of state transitions to
+ two per offdelay."""
+ def __init__(self, device, ondelay, offdelay):
+ """
+ @type device: OnoffDevice
+ @type ondelay: int or float
+ @param ondelay: delay the report of OnoffState.active by this many
+ seconds
+ @type offdelay: int or float
+ @param offdelay: delay the actual deactivation by this many seconds
+ """
+ OnoffDevice.__init__(self)
+ self.device = device
+ self.ondelay = ondelay
+ self.offdelay = offdelay
+ self.desired_state = OnoffState.inactive
+ self.transition = None
+ self.device.notify.add(self.changestate)
+
+ @property
+ def state(self):
+ if self.transition is None:
+ return self.device.state
+ return self.desired_state | OnoffState.transition
+
+ def _schedule_transition(self, delay, func):
+ assert self.transition is None
+ self.transition = ScheduledFunction(delay, func)
+
+ def _cancel_transition(self):
+ assert self.transition is not None
+ self.transition.cancel()
+ self.transition = None
+
+ def changestate(self, state):
+ if state != OnoffState.active:
+ OnoffDevice.changestate(self, state)
+ else:
+ if self.desired_state == OnoffState.inactive:
+ logger.warning("device became active but we want inactive," +
+ "suppresing signal")
+ elif self.transition is None:
+ logger.debug("scheduling report of activation in %.1fs",
+ self.ondelay)
+ self._schedule_transition(self.ondelay, self._report_active)
+ else:
+ logger.debug("suppressing duplicate activation signal")
+
+ def _report_active(self):
+ assert self.desired_state == OnoffState.active
+ assert self.transition is not None
+ self.transition = None
+ logger.debug("delivering activation signal")
+ OnoffDevice.changestate(self, OnoffState.active)
+
+ def activate(self):
+ if self.desired_state == OnoffState.inactive and \
+ self.transition is not None:
+ logger.debug("cancelling pending deactivation")
+ self._cancel_transition()
+ curstate = self.device.state
+ if curstate == OnoffState.active:
+ self.desired_state = OnoffState.active
+ OnoffDevice.changestate(self, OnoffState.active)
+ return
+ logger.warning("device should be active during delayed " +
+ "deactivation, but is in state %s", curstate.name)
+ self.desired_state = OnoffState.active
+ self.device.activate()
+ self.changestate(self.device.state)
+
+ def _do_stop(self):
+ assert self.desired_state == OnoffState.inactive
+ assert self.transition is not None
+ self.transition = None
+ logger.debug("actually deactivating")
+ self.device.deactivate()
+
+ def deactivate(self):
+ if self.desired_state == OnoffState.active and \
+ self.transition is not None:
+ logger.debug("cancelling pending activation report")
+ self._cancel_transition()
+ self.desired_state = OnoffState.inactive
+ if self.transition is None:
+ logger.debug("scheduling actual deactivate in %.1fs",
+ self.offdelay)
+ self._schedule_transition(self.offdelay, self._do_stop)
+ self.changestate(self.state)
+ else:
+ logger.debug("not issuing deactivate due to pending deactivate")
+
+ def close(self):
+ if self.transition is not None:
+ logger.info("cancelling pending transition")
+ self._cancel_transition()
+ if self.desired_state == OnoffState.inactive:
+ logger.info("invoking pending deactivate early during close")
+ self.device.deactivate()
+ self.device.notify.remove(self.changestate)
+ self.device.close()
projects / ~helmut / onoff.git / blobdiff