Shut down GPIO threads nicely

The motion sensor queue doesn't shut down properly at script end at the
moment and prevents the interpreter shutting down. This is because it's
a non-daemon thread so `__del__` never gets run and so on.

This is a bit of a major PR - I can split it up if you want. Firstly it
makes a common base class called `GPIODevice` for both `InputDevice` and
`OutputDevice`. This just takes care of the read-only pin stuff. Next it
makes a `GPIOThread` class that ensures its a daemon thread, and which
also ensures proper cleanup on shutdown.

Finally, it fixes `MotionSensor` to use the new `GPIOThread` class
(tested this time! Works nicely) and adds the `blink` method to the
`LED` class (which also works nicely this time).
This commit is contained in:
Dave Jones
2015-09-18 11:19:13 +01:00
parent c810730cf5
commit e2ddad6fea
4 changed files with 118 additions and 37 deletions

View File

@@ -1,9 +1,15 @@
from RPi import GPIO
from .devices import GPIODeviceError, GPIODevice, GPIOThread
class OutputDevice(object):
def __init__(self, pin):
self.pin = pin
class OutputDeviceError(GPIODeviceError):
pass
class OutputDevice(GPIODevice):
def __init__(self, pin=None):
super(OutputDevice, self).__init__(pin)
GPIO.setup(pin, GPIO.OUT)
def on(self):
@@ -14,7 +20,36 @@ class OutputDevice(object):
class LED(OutputDevice):
pass
def __init__(self, pin=None):
super(LED, self).__init__(pin)
self._blink_thread = None
def blink(self, on_time, off_time):
self._stop_blink()
self._blink_thread = GPIOThread(target=self._blink_led, args=(on_time, off_time))
self._blink_thread.start()
def _stop_blink(self):
if self._blink_thread:
self._blink_thread.stop()
self._blink_thread = None
def _blink_led(self, on_time, off_time):
while True:
super(LED, self).on()
if self._blink_thread.stopping.wait(on_time):
break
super(LED, self).off()
if self._blink_thread.stopping.wait(off_time):
break
def on(self):
self._stop_blink()
super(LED, self).on()
def off(self):
self._stop_blink()
super(LED, self).off()
class Buzzer(OutputDevice):