Raspberry Pi Christmas Tree Light Show; Controlling GPIO pins over the web
I made a Raspberry Pi powered Christmas Tree light show that can be controlled via the web by any device on the same WiFi network. For this project I used:
- (1) Raspberry Pi Model 3
- (1) 8 Channel SSR Relay
- (2) 7 Channel Power Distribution Bus
- (7) extension cords
- Female to Female Jumper Wires and 14 gauge wire
- Raspberry Pi GPIO Pin Diagram
There will be a few truncated code examples below. You can see all the code on my Github.
How it all works
I am not going to go deep into explaining exactly what each portion of the code does, but rest assured there are plenty of resources out there if you want to learn more about Python, Flask, JavaScript, and the GPIO . I have a list of resource links at the bottom for those who are interested. In retrospect, a lot of my time on this project was spent reading about multiprocessing, threading, and electricity. Note: I only shocked myself one time.
The Raspberry Pi is running a Flask server that executes GPIO functions when the server receives specific GET requests. I can access the server via a web page as long as I am on the same wireless network as the Pi. The server has 2 different types of Flask routes: 1 main index route that returns an HTML page, and n number of other routes that activate GPIO functions. I ended up with 4, but could add more if I wanted (to an extent).
Each GPIO function is a separate Python thread that will activate when the route is requested. When the Flask route corresponding to the GPIO function receives a GET request, the thread defined in its function starts and/or resumes after pausing all other threads. There is also a “shutdown” request that pauses all threads. The requests are sent via Ajax. Therefore, I never leave the index page.
Flask Server
app.py
The Flask server is very basic. In fact, its only a little bit longer than the “Hello World” example on the Flask website. For brevity, I am showing shortened versions of the code below.
After my imports, I am instantiating a Flask app and defining my routes. In Flask, routes are decorators that bind functions to URLs. For example, when I visit my localhost, Flask will run the index function, and in my case, return my index.html page. The name of the route function does not need to correlate to the URL (I don’t know you wouldn’t though). I could name my homepage function homepage() instead of index() for example and everything would still work exactly the same. Check out the Flask quick start guide for more information.
from flask import Flask, render_template
import os# Load the env variables
if os.path.exists('.env'):
print('Importing environment from .env...')
for line in open('.env'):
var = line.strip().split('=')
if len(var) == 2:
os.environ[var[0]] = var[1]app = Flask(__name__)@app.route("/")
def index():
return render_template("index.html")
The only route that returns a view-able page is the index route. This function will return the index.html template which contains the buttons (lightshow, blink, etc.) to send the various Ajax requests. This is the only HTML page of the application. Each subsequent route pauses all currently running threads and starts its corresponding thread. Remember that each thread is a function that executes a function that uses the GPIO.
Each GPIO function route does the same thing:
- Pause all active threads
- Start the target thread if it has not been started
- Resume the target thread (since it starts paused by default)
For example, the route will turn on all the pins, sleep, then turn them off. Basically it will cause all the lights to blink (hence its name)
@app.route("/blink", methods=['GET'])
def blink_view():
# Pause any running threads
any(thread.pause() for thread in threads)# Start the target thread if it is not running
if not blink_thread.isAlive():
blink_thread.start()
# Unpause the thread and thus execute its function
blink_thread.resume()
return "blink started"
For future reference, the return string here (“blink started”) will be what I see in the JavaScript console when requesting this route via Ajax (assuming the request I received a 200 response).
Skipping down to the bottom of the Flask code, I am instantiating my threads, collecting them in a list, and running the Flask server. More on the RaspberryThread class in a bit.
I have passed a couple of keyword arguments to app.run. First, since this app will never go to “production”, I have turned on the always helpful debug mode. Second, I have declared my Raspberry Pi’s IP address (replaced with zeros here) as the host instead of the default localhost. This will allow anyone on the same wireless network to see the website by entering this IP address and port into their browser.. This accomplishes my goal of impressing my Christmas party guests by turning my Christmas tree lights on and off with my phone.
Finally, I am using Flask’s built in threading capabilities by setting threaded=True. Flask threading allows the server to handle multiple requests at the same time. I am not 100% sure this is necessary (is this necessary?), but since I am using threads to execute other functions I figure it couldn’t hurt. Also, if a multiple people pull up the page at the same time they should all be able to send requests without issue.
if __name__ == '__main__':
# Create threads
blink_thread = RaspberryThread(function=blink_all)
lightshow_thread = RaspberryThread(function=lightshow)
russian_xmas_thread = RaspberryThread(function=russian_xmas)# collect threads
threads = [
blink_thread,
lightshow_thread,
russian_xmas_thread,
]# Run server
app.run(debug=True, host="000.000.00.00", port=5000, threaded=True)
Sending Requests
script.js
Each button click on my index page will send an ajax request to the corresponding Flask route/URL. For example, clinking on the “Blink” button will execute the blink() JavaScript function which sends a get request to the relative URL “/blink”.
The button HTML has a JavaScript function bound to it that executes when the button is clicked:
<button onclick="ajaxRequest('/blink')">Blink</button>The JavaScript blink function. If you recall from the return value from the blink Flask route (“blink started”), that is the value will show up here as the xhr.responseText.
function ajaxRequest(url) {
var xhr = new XMLHttpRequest();
xhr.open('GET', url);xhr.onload = function() {
if (xhr.status === 200) {
console.log('Response Text: ' + xhr.responseText);
document.getElementById("message").innerHTML = "the " + url.slice(1) + " thread is active";
} else {
console.log('Request failed. Returned status of: ' + xhr.status);
}
};
xhr.send();
}
When localhost/blink receives a GET request the route will execute a corresponding GPIO function:
def blink(*, pin_numbers: list, iterations=1, sleep=0.5) -> None:
"""Turn all pins on, sleep, turn all pins off"""
while iterations > 0:
any(on(pin_number) for pin_number in pin_numbers)
time.sleep(sleep)
any(off(pin_number) for pin_number in pin_numbers)
time.sleep(sleep)
iterations -= 1A note on the use of the any() function: I wanted to run my on/off functions on a list of GPIO pin numbers without returning anything and without using a for loop. Map() and List comprehensions return lists. Any() returns False, but still executes the function on each item in the iterator.
GPIO — Setup
light_functions.py
GPIO or General Purpose Input/Output pins can be utilized to control various Raspberry Pi peripherals and other electronics. In this case, I will be using the GPIO to interact with a power relay. There’s a lot to learn about the GPIO that I am not going to go into here. If you have never worked with it before I suggest checking out a few YouTube videos. It is a lot of fun and the possibilities are endless. The GPIO package is specific to the Raspberry Pi and does not work on other devices i.e my Macbook. You can fake it, but in my opinion, its not worth the effort. It is best just to work “on” the Pi. I will touch on my particular workflow later.
Before I can start interacting with the GPIO pins, I need to do some setup. Due to the fact that each power bus has 7 channels (I am using them in conjunction) and 1 of those is needed to power the bus itself, I will only manipulating 6 of the 8 relays on my relay board. Therefore, I need to prepare 6 pins which will control 6 separate strands of lights. I will be using GPIO pins 5, 6, 13, 19, 26, & 16. Whenever I work with the GPIO I always have a tab open with the pin diagram for reference. I should really keep one of the GPIO cheat sheet cards that comes with just about every Raspberry Pi. Maybe next time…
I prefer referring to the pins by their GPIO number, not pin number. For example, GPIO 5 is actually pin 29 on the board. I find it easier to remember the GPIO number and can tell my program to map the pins as such using the setmode() method. This is a little confusing at first, but once you work with it a bit it will click.
import RPi.GPIO as GPIO# Turn off warnings
GPIO.setwarnings(False)# Set pin mapping to board, use GPIO numbers not pin numbers
GPIO.setmode(GPIO.BCM)
To reference my pins and their numbers, I’ve made a pin class and a few helper variables. Jumper color refers the color of the jumper wire connecting the board to the relay.
class Pin(object):
def __init__(self, pin_number, jumper_color, relay_number):
self.pin_number = pin_number
self.jumper_color = jumper_color
self.relay_number = relay_numberpin1 = Pin(pin_number=5, jumper_color="green", relay_number=1)
pin2 = Pin(pin_number=6, jumper_color="orange", relay_number=2)
pin3 = Pin(pin_number=13, jumper_color="purple", relay_number=3)
pin4 = Pin(pin_number=19, jumper_color="blue", relay_number=4)
pin5 = Pin(pin_number=26, jumper_color="white", relay_number=5)
pin6 = Pin(pin_number=16, jumper_color="brown", relay_number=6)pins = [
pin1, pin2, pin3,
pin4, pin5, pin6,
]pin_numbers = [pin.pin_number for pin in pins]
relay_numbers = [pin.relay_number for pin in pins]
relay_pin_map = dict(zip(relay_numbers, pin_numbers))
Finally, to complete the setup I need to run GPIO.setup([PIN_NUMBER], GPIO.OUT) on each pin. I could use my any() method trick here, but I feel the for loop is more explicit.
for pin in pin_numbers:
GPIO.setup(pin, GPIO.OUT)Now I am ready to start programming my GPIO pins.
GPIO — Functions
light_functions.py
Since all I am really doing is turning on and off strings of Christmas lights I need to do that in an interesting way. Here I am making a few simple “helper” functions, then combining them to create a main function called lightshow. Lightshow will execute a random “helper” function on a random pin configuration for a random number of iterations and sleeping for a random time between ons and offs for between .1 and .5 seconds.
def blink(*, pin_numbers: list, iterations=1, sleep=0.5) -> None:
"""Turn all pins on, sleep, turn all pins off"""
while iterations > 0:
any(on(pin_number) for pin_number in pin_numbers)
time.sleep(sleep)
any(off(pin_number) for pin_number in pin_numbers)
time.sleep(sleep)
iterations -= 1def step(*, pin_numbers: list, iterations=1, sleep=0.5) -> None:
"""Turn a pin on then off then move onto the next pin"""
while iterations > 0:
for pin in pin_numbers:
on(pin)
time.sleep(sleep)
off(pin)
iterations -= 1def climb(*, pin_numbers: list, iterations=1, sleep=0.5) -> None:
"""Turn the pins on in order then off in reverse"""
climb_number = len(pin_numbers)
reversed_pin_numbers = reversed(pin_numbers)
while iterations > 0:
for index, pin in enumerate(pin_numbers):
if index <= climb_number:
on(pin)
time.sleep(sleep)
for pin in reversed_pin_numbers:
off(pin)
time.sleep(sleep)
iterations -= 1def lightshow():
"""
Choose a random simple function
Execute it for a random number of iterations between 1-5
Sleep for a random time during function between .1 and .5 seconds
"""
all_pins = pin_numbers
even_pins = pin_numbers[0:][::2]
odd_pins = pin_numbers[1:][::2]
random_pin = random.choice(pin_numbers)
first_half_pins = pin_numbers[:int(len(pin_numbers) / 2)]
second_half_pins = pin_numbers[int(len(pin_numbers) / 2):]
random_sleep = float(str(random.uniform(.1, .5))[:4])
random_iterations = random.choice(list(range(5))) functions = [blink, step, climb]
pin_configs = [
all_pins, even_pins, odd_pins,
first_half_pins, second_half_pins, random_pin
] random.choice(functions)(
pin_numbers=random.choice(pin_configs),
iterations=random_iterations,
sleep=random_sleep)
I apologize for any strange formatting Medium might do. It makes more sense to see it in action.
Threads
I am giving my RaspberryThread class a function on instantiation which it will execute while the thread is alive and not paused. This was illustrated above as:
# from above
# blink_thread = RaspberryThread(function=blink_all)When a new thread object is created a new thread spins up, but defaults to a paused state. While the thread is paused, its target function will not execute. To start the target function, I am just resuming the thread.
class RaspberryThread(threading.Thread):
def __init__(self, function):
self.paused = True
self.state = threading.Condition()
self.function = function
super(RaspberryThread, self).__init__() def start(self):
super(RaspberryThread, self).start() def run(self):
# self.resume() #unpause self
while True:
with self.state:
if self.paused:
self.state.wait() #block until notifed
while not self.paused:
# Call function
self.function() def resume(self):
with self.state:
self.paused = False
self.state.notify() def pause(self):
with self.state:
self.paused = True
There isn’t much else I can say about my Thread class because frankly I am still learning a lot about threading and multiprocessing in general. The class is relied heavy on this Stack Overflow post.
Wiring, Pi Configuration, Workflow, etc.
Wiring extension cords into the relay is pretty simple. Just make sure you keep track of the hot wire (the one that leads to the smaller prong on the plug) when stripping the cords. I based my wiring and use of power buses on this instructables post. I also borrowed some of their code for the light sequence function (Mad Russian’s Christmas).
I don’t want to have to SSH into the Pi in order to start the server every time I want to use the Christmas lights. I would prefer the server fired up as soon as the Pi turns on. No problem. I just need to add the appropriate commands to the bottom of my /etc/profile file. Now, when my Pi boots, the flask server will start and I can start using my lights. Automatically running a Python script on boot.
Finally, I discovered a really helpful terminal command to expedite my workflow. When working on Raspberry Pi projects I like to write code on my laptop and push code to the Pi over SSH using the secure copy (scp) command. This works fine, but overwrites files even if they are unchanged. There are also some extra arguments required (that I can never remember) to send a whole folder. I came across the command rsync which can be used to sync folders/files between local and remote directories. Rsync is great because it will only transfer changed files and can resume transfers if the connection is interrupted. I highly recommend it if you work over SSH frequently.
All Together Now
(The clicking is the relays turning on and off)
Thank you for reading and please email me if you have any questions.
