4 from .gobject import ScheduledFunction
6 logger = logging.getLogger("onoff.common")
9 class OnoffState(enum.Flag):
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
22 @type notify: {int -> None}
23 @ivar notify: A set of functions taking a changed state.
26 @ivar state: is a read-only attribute to retrieve the current state
31 def changestate(self, state):
32 """Tell interested parties that the state has changed to the given
34 for func in self.notify:
38 """Ask the device to power on."""
39 raise NotImplementedError
42 """Ask the device to power off."""
43 raise NotImplementedError
46 """Release resources acquired by the constructor."""
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)
54 self.device.activate()
55 self.device.notify.add(self.changestate)
57 def changestate(self, state):
58 OnoffDevice.changestate(self, state ^ OnoffState.active)
62 return self.device.state ^ OnoffState.active
65 self.device.deactivate()
68 self.device.activate()
71 self.device.notify.remove(self.changestate)
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
78 def __init__(self, device, ondelay, offdelay):
80 @type device: OnoffDevice
81 @type ondelay: int or float
82 @param ondelay: delay the report of OnoffState.active by this many
84 @type offdelay: int or float
85 @param offdelay: delay the actual deactivation by this many seconds
87 OnoffDevice.__init__(self)
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)
97 if self.transition is None:
98 return self.device.state
99 return self.desired_state | OnoffState.transition
101 def _schedule_transition(self, delay, func):
102 assert self.transition is None
103 self.transition = ScheduledFunction(delay, func)
105 def _cancel_transition(self):
106 assert self.transition is not None
107 self.transition.cancel()
108 self.transition = None
110 def changestate(self, state):
111 if state != OnoffState.active:
112 OnoffDevice.changestate(self, state)
114 if self.desired_state == OnoffState.inactive:
115 logger.warning("device became active but we want inactive," +
117 elif self.transition is None:
118 logger.debug("scheduling report of activation in %.1fs",
120 self._schedule_transition(self.ondelay, self._report_active)
122 logger.debug("suppressing duplicate activation signal")
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)
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)
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)
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()
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",
163 self._schedule_transition(self.offdelay, self._do_stop)
164 self.changestate(self.state)
166 logger.debug("not issuing deactivate due to pending deactivate")
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)