Send commands to Arduino using Python from Twitter
I wanted to figure out how to send commands to the Arduino from Twitter. For example, if someone used a specific hashtag the Arduino would communicate to an LED turning it on for a few seconds then turning it off.
I managed to get this right using Python, the Twitter Stream API and Serial communication to my Arduino Uno.
Here is the result:
I created a small prototype of this to show to a client using this code and constructing a box
To be totally honest it's an extremely simple exercise due to the amount of exceptional libraries available on Github.
I decided to connect to the Twitter API using Python 2.7.10 since my Arduino is connected to . To be fair I'm sure I could condense this project to run completely on the Raspberry Pi itself but I only just completed this first iteration now and perhaps that will be a future project.
I first needed to connect to Twitter by creating an application in the Twitter Developer portal. I learned very quickly that using the REST API would result in rate limiting and wasn't an ideal solution of what I wanted to achieve. I ended up looking at the Twitter Stream API which could keep a connection open to Twitter without causing issues with rate limiting.
After some Google searching I came across this fantastic library which handled all the heavy lifting for me called TwitterAPI by Jonas Geduldig. Using the library I was able to connect with OAUTH and create a track for a specific hashtag.
Once that was done I just needed to send the command to the Arduino using PySerial and that would complete the Python scripting requirements.
On the Arduino side I needed to write some script to handle the commands sent from Python which would finally toggle the LED to turn on and off. Nothing special about this but it should give you a good indication about how to get this running and how to hopefully expand on this.
Arduino requirements
- Arduino Uno
- Breadboard
- 220 Ohm Resistor
- LED Light
- USB Cable to connect Arduino board to computer
- Arduino IDE
- 3 wires to connect to power, ground and the digital pin for the LED
Arduino circuit setup
Arduino sketch
With the Arduino sketch, I'm just flashing the LED whenever a Serial message is sent from Python otherwise the LED be off. I've added a simple debugging switch to show what behaviour to expect from the LED once the Serial message is received from Python. Upload this sketch to your Arduino board with the debugging value set to 1 to check if your circuitry is connected correctly and that your LED is connected to the correct pin. I'm using pin A1 for the LED connection on the board.
/*
* Author: Nicholas Bester
* Title: Twitter Mention Test
* Version: 0.1
*/
// Debuging variables
int const DEBUG = 0; // Test LED without Serial feedback
// LED control
int ledPin = A1;
// Value sent from Python
int signalState;
void setup() {
// Transistor pin connection on board
pinMode(ledPin, OUTPUT);
// Enabling communication
Serial.begin(9600);
// Test breadboard setup to LED
if (DEBUG) {
tweetReceived();
}
}
void loop() {
if (!DEBUG) {
if (Serial.available()) {
byte receivedValue = Serial.read() - '0';
signalState = receivedValue;
if (signalState == 1) {
tweetReceived();
}
else if (signalState == 0) {
ledToggle(false);
}
Serial.flush();
}
}
}
// Flash the light when tweet is received
void tweetReceived() {
for (int i = 0; i < 10; i++) {
ledToggle(true);
delay(100);
ledToggle(false);
delay(100);
}
}
// turn LED on and off
void ledToggle(boolean value) {
if (value) {
analogWrite(ledPin, 1023);
} else {
analogWrite(ledPin, 0);
}
}
Python script
View the updated script at the bottom of the page
The Python script listens to communications from Twitter using the Twitter Streaming API. As soon as a tweet is made matching the pattern specified in the track, Python will send a serial message to the Arduino using the PySerial library.
As an added extra I've made Python output the user for each tweet received which I'll probably extra to displaying on an LCD screen on the Arduino Uno.
Another thing to note is that the functionality doesn't really cater for popular tracked strings. Ideally if there were a lot of tweets coming back they should be managed in a way so we can create a queue of commands to send to the Arduino.
Remember to insert your Twitter access token and secret
# Author: Nicholas Bester
# Title: Twitter Stream connection
# Description: Tracks a string using the Twitter Streaming API
# Send commands using PySerial to an Arduino
# Version: 1.0
# imports
import time
from time import sleep
from TwitterAPI import TwitterAPI
import struct
import os
from serial import Serial
# Pretty console colours
class bcolors:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
# Variables
availableArduino = True # Debugging without an Arduino
testSerial = False # Debugging without Twitter connection
arduinoPort = '[Insert USB port]' # USB port address for the Arduino
arduinoBaud = '[Insert BAUD]' # Baud for serial communication
arduinoWaitTime = 3 # The length of time Python wait before attemping to issue commands to the Arduino
stringToTrack = "[Enter a track term]" # Change this to the search term you wish to track from Twitter
# Access requirements for Twitter API connection
access_token = '[insert token here]'
access_token_secret = '[insert secret here]'
consumer_key = '[insert key here]'
consumer_secret = '[insert secret here]'
# Clearing the screen for aesthetic display of console statements
os.system('cls' if os.name == 'nt' else 'clear')
print bcolors.WARNING + "Initialising Twitter Stream Application" + bcolors.ENDC
print bcolors.OKGREEN + "Initialisation OK!" + bcolors.ENDC
print bcolors.WARNING + 'Initialising Arduino Board through Serial' + bcolors.ENDC
# Arduino serial communication
if availableArduino:
ser = Serial(arduinoPort, arduinoBaud, timeout=3)
# Gives the Arduino board time to initialise
sleep(arduinoWaitTime)
# Testing serial send to Arduino (ensure available Arduino is True)
if testSerial:
print "On"
ser.write(bytes(1))
sleep(arduinoWaitTime)
print "Off"
ser.write(bytes(0))
sleep(arduinoWaitTime)
else:
print bcolors.OKGREEN + "Initialisation OK!" + bcolors.ENDC
print bcolors.WARNING + 'Initialising Twitter Stream API Authorisation' + bcolors.ENDC
# Setting up a connection to Twitter using the Twitter API
api = TwitterAPI(consumer_key, consumer_secret, access_token, access_token_secret)
print bcolors.OKGREEN + "Initialisation OK!" + bcolors.ENDC
print bcolors.OKBLUE + 'Twitter Stream Request running' + bcolors.ENDC
# A request object which opens a stream to Twitter tracking the hashtag in question
r = api.request('statuses/filter', {'track':stringToTrack})
# Checks if text within the stream item is populated and issues a command to the Arduino
for item in r.get_iterator():
if 'text' in item:
print item['user']['screen_name'].encode('utf-8') + ' tweeted: ' + item['text'].encode('utf-8')# Print screen name and the tweet text
# It is possible to check the tweets for further commands using regular expressions to send multiple commands to the Arduino but thats not illustrated in this code.
if availableArduino:
print "Arduino turning on the LED"
ser.write(bytes(1)) # The command is a simple byte intepretation of the integer 1
sleep(arduinoWaitTime) # Wait before sending next command
ser.write(bytes(0))
sleep(arduinoWaitTime) # Wait before sending next command
Update 1:
I expanded on this project by moving the Python script onto a Raspberry Pi to increase portability. I'll look at moving it all over to the Arduino Uno next using a WIFI shield removing Python from the equation.
Update 2:
The Python script can return an unexpected error resulting in a broken connection. When this happens it's best to just ignore the error and have the script continue accepting commands however all of this is due to a broad track term causing the script to be overwhelmed with tweets making the script fall behind on capturing. Here's the updated script resolving this:
# Author: Nicholas Bester
# Title: Twitter Stream connection
# Description: Tracks a string using the Twitter API and sends a Serial command to an Arduino once a Tweet matching the tracked string is found
# Version: 1.1
# imports
import time
from time import sleep
from TwitterAPI import TwitterAPI
import struct
import os
from serial import Serial
import httplib
from httplib import IncompleteRead
# Pretty console colours
class bcolors:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
# Variables
availableArduino = False # Debugging without an Arduino
testSerial = False # Debugging without Twitter connection
arduinoPort = '[insert here]' # USB port address for the Arduino
arduinoBaud = '9600' # Baud for serial communication
arduinoWaitTime = 3 # The length of time Python wait before attemping to issue commands to the Arduino
stringToTrack = "[insert here]" # Change this to the search term you wish to track from Twitter
# Access requirements for Twitter API connection
access_token = '[insert here]'
access_token_secret = '[insert here]'
consumer_key = '[insert here]'
consumer_secret = '[insert here]'
# Clearing the screen for aesthetic display of console statements
os.system('cls' if os.name == 'nt' else 'clear')
print bcolors.WARNING + "Initialising Twitter Stream Application" + bcolors.ENDC
print bcolors.OKGREEN + "Initialisation OK!" + bcolors.ENDC
print bcolors.WARNING + 'Initialising Arduino Board through Serial' + bcolors.ENDC
# Arduino serial communication
if availableArduino:
ser = Serial(arduinoPort, arduinoBaud, timeout=3)
# Gives the Arduino board time to initialise
sleep(arduinoWaitTime)
# Testing serial send to Arduino (ensure available Arduino is True)
if testSerial:
print "On"
ser.write(bytes(1))
sleep(arduinoWaitTime)
print "Off"
ser.write(bytes(0))
sleep(arduinoWaitTime)
else:
print bcolors.OKGREEN + "Initialisation OK!" + bcolors.ENDC
print bcolors.WARNING + 'Initialising Twitter Stream API Authorisation' + bcolors.ENDC
try:
# Setting up a connection to Twitter using the Twitter API
api = TwitterAPI(consumer_key, consumer_secret, access_token, access_token_secret)
print bcolors.OKGREEN + "Initialisation OK!" + bcolors.ENDC
print bcolors.OKBLUE + 'Twitter Stream Request running' + bcolors.ENDC
# A request object which opens a stream to Twitter tracking the hashtag in question
r = api.request('statuses/filter', {'track':stringToTrack})
# Checks if text within the stream item is populated and issues a command to the Arduino
for item in r.get_iterator():
if 'text' in item:
print item['user']['screen_name'].encode('utf-8') + ' tweeted: ' + item['text'].encode('utf-8')# Print screen name and the tweet text
# It is possible to check the tweets for further commands using regular expressions to send multiple commands to the Arduino
if availableArduino:
print "Arduino turning on the LED"
ser.write(bytes(1)) # The command is a simple byte intepretation of the integer 1
sleep(arduinoWaitTime) # Wait before sending next command
ser.write(bytes(0))
sleep(arduinoWaitTime) # Wait before sending next command
except IncompleteRead:
# Oh well, reconnect and keep trucking
print "IncompleteRead occurred"
except KeyboardInterrupt:
# Or however you want to exit this loop
api.disconnect()
exit()