edit
Wiki Currently Under Major Revision
This is not perfect documentation. This is a shared space for us to keep the most helpful and accurate things each of us knows or learns about our tools. We need your help! If you are willing to share what you learn with others, please request editor access in the mfx-wiki channel on slack. No special skills or experience required.
Alpha LED Signs
Jump to navigation
Jump to search
from threading import Thread
from time import sleep
import time
import web
import mmap
#import RPi.GPIO as GPIO
from twython import TwythonStreamer
# Search terms
TERMS = '@makerfx'
lastTweet = 'Startup...'
tweetFile = 'tweets.txt'
# GPIO pin number of LED
#LED = 22
# Twitter application authentication
APP_KEY = 'xsHnwnvrI92vZxeHo6efTGqnU'
APP_SECRET = 'RNxSJTgI1AUqLDf2ZI0zMh3XXVIL0CxZJSttYgH4zDUy4hEnPX'
OAUTH_TOKEN = '735519883136512000-F2Dop77nJLUcimdRbyTXUYpzkPv8vJD'
OAUTH_TOKEN_SECRET = 'KqDUy7CetAuWWFRWbtQ8nQxVUYCBcd0qFLAoVH4gne8XG'
# Setup callbacks from Twython Streamer
class BlinkyStreamer(TwythonStreamer):
def on_success(self, data):
lastTweet = 'StreamerLoaded...'
if 'text' in data:
lastTweet = data['text'].encode('utf-8')
#print data['text'].encode('utf-8')
print lastTweet
with open(tweetFile, "a") as myfile:
myfile.write(lastTweet)
myfile.write('\n')
myfile.close()
#GPIO.output(LED, GPIO.HIGH)
#web stuff
urls = (
'/', 'index',
'/makerfx', 'makerfx'
)
app = web.application(urls, globals())
def start_streamer(arg):
# Create streamer
stream = BlinkyStreamer(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
stream.statuses.filter(track=TERMS)
class index:
def GET (self):
return 'Hello'
class makerfx:
def GET(self):
with open (tweetFile) as myfile:
head="error, couldn't read file"
#head = myfile.readlines()[0:1]
mapping=mmap.mmap(myfile.fileno(),0, prot=mmap.PROT_READ)
head = mapping[mapping.rfind(b'\n', 0, -1)+1:]
# print tweetFile
# with open(tweetFile, "a") as myfile:
# myfile.write("ack\n")
# myfile.close()
return head
if __name__ == "__main__":
thread = Thread(target=start_streamer, args=(10, ))
thread.start()
#thread.join
#web stuff
urls = (
'/', 'index',
'/makerfx', 'makerfx'
)
app = web.application(urls, globals())
def start_streamer(arg):
# Create streamer
stream = BlinkyStreamer(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
stream.statuses.filter(track=TERMS)
class index:
def GET (self):
return 'Hello'
class makerfx:
def GET(self):
with open (tweetFile) as myfile:
head="error, couldn't read file"
#head = myfile.readlines()[0:1]
mapping=mmap.mmap(myfile.fileno(),0, prot=mmap.PROT_READ)
head = mapping[mapping.rfind(b'\n', 0, -1)+1:]
# print tweetFile
# with open(tweetFile, "a") as myfile:
# myfile.write("ack\n")
# myfile.close()
return head
if __name__ == "__main__":
thread = Thread(target=start_streamer, args=(10, ))
thread.start()
#thread.join
app.run()