Raspberry Pi Twitter Monitor

Contributors: Shawn Hymel

Dissecting the Code

We could just give you the code to copy-and-paste into your Pi (which we did). However, the Python script is a bit more complicated than just blinking an LED, so we should take a moment to discuss what is going on.

At the top of the code, we import our necessary Python modules. We need "time" to call our 1/2 second delay when we blink our LED. "RPi.GPIO" is a pre-build Python package for the Raspberry Pi that gives us access to the GPIO pins. This is why we recommend the Raspbian build of Linux. Finally, "twython" is the package that connects us to Twitter allowing us to monitor the stream, search, and post.

language:python
import time
import RPi.GPIO as GPIO
from twython import TwythonStreamer

Next, we add our global constants. TERMS holds a string (or strings) of things we want to search for on Twitter. LED contains the pin number for the GPIO header. Note that this is the pin number of the header and not the GPIO number. The Twitter authentication constants hold the token strings that we copied from the dev.twitter.com page. Remember: you need to change this section to your specific Twitter token strings.

language:python
# Search terms
TERMS = '#yes'

# GPIO pin number of LED
LED = 22

# Twitter application authentication
APP_KEY = ‘<Your Consumer Key>’
APP_SECRET = ‘<Your Consumer Secret>’
OAUTH_TOKEN = ‘<Your Access Token>’
OAUTH_TOKEN_SECRET = '<Your Access Token Secret>’'

The next section is the crux of the Twitter monitor. If you have never dealt with class inheritance or callbacks before, we recommend brushing up on those concepts in order to really understand this section of code.

We create a new class that inherits TwythonStreamer (a class within Twython). While TwythonStreamer handles a bunch of stuff in the background (including communicating with the Twitter API), we are only concerned with the callback on_success(). This is a specially named method that gets called whenever a Tweet appears matching our search terms criteria. The important thing to understand is that we never call this method from within our script. An outside force (e.g. thread, program) calls on_success() for us. We just have to define what happens when that method is called. To use it, we create a BlinkyStreamer object and Twython handles the rest.

Inside of on_success() is what we want to happen whenever a successful Twitter hit occurs. In this case, we make sure there is a 'text' field within the Tweet (meaning that the Tweet contains actual text from a user), and we print the Tweet's text to the console and flash the LED for 1/2 second.

language:python
# Setup callbacks from Twython Streamer
class BlinkyStreamer(TwythonStreamer):
        def on_success(self, data):
                if 'text' in data:
                        print data['text'].encode('utf-8')
                        print
                        GPIO.output(LED, GPIO.HIGH)
                        time.sleep(0.5)
                        GPIO.output(LED, GPIO.LOW)

For those of you that are code-savvy, you might say, "Hey, wait a minute! That sleep function is blocking! You can't receive tweets while flashing the LED." And you would be correct. This is just a plain simple way to monitor Tweets. To "properly" do this, you would want to set up a separate thread in Python that monitors Tweets and puts them into a queue. Another thread would read that queue and flash the LED as quickly as Tweets are put into the queue. Threading is a bit too much for this tutorial, but if you would like to learn how, feel free to check out Threading in Python.

The GPIO setup section should look familiar if you have ever played with an Arduino or other microcontrollers. We need to tell the Raspberry Pi that we will be referencing its GPIO headers by board number (pin 1, pin 2, pin 3, etc.), that we want our LED pin to be an output, and initialize the LED to off (logic LOW).

language:python
# Setup GPIO as output
GPIO.setmode(GPIO.BOARD)
GPIO.setup(LED, GPIO.OUT)
GPIO.output(LED, GPIO.LOW)

Finally, we create an instance of our BlinkyStreamer class and let it do its thing. When we create a TwythonStreamer object (remember: our BlinkyStreamer inherits TwythonStreamer), we need to feed it our Twitter authentication information that we set earlier. Twython uses this information to connect to the Twitter servers. We provide the filter() method within TwythonStreamer some search terms and the streamer begins to monitor Twitter. Whenever one of the terms appears in the Twitter stream, the on_success() method (defined above) is called.

language:python
# Create streamer
try:
        stream = BlinkyStreamer(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
        stream.statuses.filter(track=TERMS)
except KeyboardInterrupt:
        GPIO.cleanup()

Because the filter() method is blocking, the program will run forever until we manually kill it. We can terminate the program by pressing ctrl+c keys. We wrap the streamer calls in a try/catch statement so that the GPIO pins can be released when we tell Python to exit. If we do not do this, Python will issue a stern warning that the GPIO pins are in use by another program.