Greg Sandstrom / SOARS Status and Alerting

1
Download as
Browse Code ยป

Commit c27f3d786e5b6d00b8c300d20f5e25bf1ef655af

Authored by Greg Sandstrom
1 parent 22a44ea124
Exists in main

always try to turn off a channel when stopping blink even if there is no running thread

Showing 1 changed file with 1 additions and 1 deletions Inline Diff

ssa/alarm/lightstack.py View file @ c27f3d7
""" 1 1 """
lightstack.py 2 2 lightstack.py
Control program for buzzer and 5-light stack. 3 3 Control program for buzzer and 5-light stack.
4 4
LTA 5 5 LTA
""" 6 6 """
import threading 7 7 import threading
import functools 8 8 import functools
import os 9 9 import os
10 10
TEST_MODE = True if os.environ.get("TEST_GPIO") == "True" else False 11 11 TEST_MODE = True if os.environ.get("TEST_GPIO") == "True" else False
12 12
if TEST_MODE == True: 13 13 if TEST_MODE == True:
import fakeGPIO as GPIO 14 14 import fakeGPIO as GPIO
else: 15 15 else:
import RPi.GPIO as GPIO 16 16 import RPi.GPIO as GPIO
17 17
18 18
class Channel: 19 19 class Channel:
# mappings for channel name (e.g. "Buzzer", "Red", "Green") to pin 20 20 # mappings for channel name (e.g. "Buzzer", "Red", "Green") to pin
# number using the BCM number (e.g. GPIO 17, as opposed to BOARD 21 21 # number using the BCM number (e.g. GPIO 17, as opposed to BOARD
# header numbering) 22 22 # header numbering)
Buzzer = 17 23 23 Buzzer = 17
White = 27 24 24 White = 27
Blue = 22 25 25 Blue = 22
Green = 23 26 26 Green = 23
Yellow = 24 27 27 Yellow = 24
Red = 25 28 28 Red = 25
29 29
30 30
# decorator function 31 31 # decorator function
# A lot of complication just so that we can decorate 32 32 # A lot of complication just so that we can decorate
# other class methods with @_with_lock so that it is 33 33 # other class methods with @_with_lock so that it is
# guarded by self.lock 34 34 # guarded by self.lock
def _with_lock(func): 35 35 def _with_lock(func):
@functools.wraps(func) 36 36 @functools.wraps(func)
def wrapper(self, *args): 37 37 def wrapper(self, *args):
with self.lock: 38 38 with self.lock:
func(self, *args) 39 39 func(self, *args)
40 40
return wrapper 41 41 return wrapper
42 42
43 43
class LightStack: 44 44 class LightStack:
ON = GPIO.HIGH 45 45 ON = GPIO.HIGH
OFF = GPIO.LOW 46 46 OFF = GPIO.LOW
47 47
def __init__(self): 48 48 def __init__(self):
# use BCM pin numbering 49 49 # use BCM pin numbering
GPIO.setmode(GPIO.BCM) 50 50 GPIO.setmode(GPIO.BCM)
# I guess make this threadsafe 51 51 # I guess make this threadsafe
self.lock = threading.RLock() 52 52 self.lock = threading.RLock()
# dicts to store state and threads 53 53 # dicts to store state and threads
self.channel_states = dict() 54 54 self.channel_states = dict()
self.channel_threads = dict() 55 55 self.channel_threads = dict()
# initialize states 56 56 # initialize states
for channel in LightStack.enumerate_channels(): 57 57 for channel in LightStack.enumerate_channels():
print(f"setup channel {channel}") 58 58 print(f"setup channel {channel}")
GPIO.setup(channel, GPIO.OUT) 59 59 GPIO.setup(channel, GPIO.OUT)
GPIO.output(channel, GPIO.LOW) 60 60 GPIO.output(channel, GPIO.LOW)
self.channel_states[channel] = LightStack.OFF 61 61 self.channel_states[channel] = LightStack.OFF
print("LighStack initialized") 62 62 print("LighStack initialized")
63 63
@_with_lock 64 64 @_with_lock
def on(self, channel): 65 65 def on(self, channel):
GPIO.output(channel, LightStack.ON) 66 66 GPIO.output(channel, LightStack.ON)
self.channel_states[channel] = LightStack.ON 67 67 self.channel_states[channel] = LightStack.ON
68 68
@_with_lock 69 69 @_with_lock
def off(self, channel): 70 70 def off(self, channel):
GPIO.output(channel, LightStack.OFF) 71 71 GPIO.output(channel, LightStack.OFF)
self.channel_states[channel] = LightStack.OFF 72 72 self.channel_states[channel] = LightStack.OFF
73 73
@_with_lock 74 74 @_with_lock
def toggle(self, channel): 75 75 def toggle(self, channel):
if self.channel_states[channel] == LightStack.OFF: 76 76 if self.channel_states[channel] == LightStack.OFF:
self.on(channel) 77 77 self.on(channel)
else: 78 78 else:
self.off(channel) 79 79 self.off(channel)
80 80
@_with_lock 81 81 @_with_lock
def start_blink(self, interval, channel): 82 82 def start_blink(self, interval, channel):
if self.channel_threads.get(channel, None) is None: 83 83 if self.channel_threads.get(channel, None) is None:
new_thread = _Blinker(interval, self, channel) 84 84 new_thread = _Blinker(interval, self, channel)
self.channel_threads[channel] = new_thread 85 85 self.channel_threads[channel] = new_thread
new_thread.start() 86 86 new_thread.start()
87 87
@_with_lock 88 88 @_with_lock
def stop_blink(self, channel): 89 89 def stop_blink(self, channel):
thread = self.channel_threads.get(channel, None) 90 90 thread = self.channel_threads.get(channel, None)
if thread is not None: 91 91 if thread is not None:
thread.cancel() 92 92 thread.cancel()
thread.join() 93 93 thread.join()
self.channel_threads[channel] = None 94 94 self.channel_threads[channel] = None
self.off(channel) 95 95 self.off(channel)
96 96
def get(self, channel): 97 97 def get(self, channel):
return self.channel_states.get(channel, None) 98 98 return self.channel_states.get(channel, None)
99 99
@staticmethod 100 100 @staticmethod
def enumerate_channels(): 101 101 def enumerate_channels():
for channel_name in dir(Channel): 102 102 for channel_name in dir(Channel):
if channel_name[0] != "_": 103 103 if channel_name[0] != "_":
yield getattr(Channel, channel_name) 104 104 yield getattr(Channel, channel_name)
105 105