-
Notifications
You must be signed in to change notification settings - Fork 0
/
interactive_counter.py
132 lines (108 loc) · 4.14 KB
/
interactive_counter.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
# -*- coding: utf-8 -*-
"""
Created on Sat May 25 20:46:25 2013
@author: leftbigtoe
"""
from tweepy import StreamListener
import tweepy, time, sys, json
from thread import start_new_thread
import numpy as np
import serial
class SListener(StreamListener):
def __init__(self, api = None):
# internal parameters
self.nBins = 60
self.maxWPM = 180
self.CLICK = 200
self.api = api or API()
# intialize variables
self.counter = 0
self.buffer = np.zeros(60, dtype = int)
self.wpm = 0
# open arduino connection
self.ardu = serial.Serial("COM4", 9600, timeout = 3)
# start buffer manager in seperate thread
start_new_thread(self.bufferManager,(0,))
def bufferManager(self,dummy):
"""
busy waiting thread that sums up received keywords in one second
writes it to the buffer every second to the buffer, calculates the
average and pushes that to the arduino
"""
# intialize index
index = int(time.clock()%60)
# endless loop
while(True):
time.sleep(0.01)
# if new bin is entered
if (not(index == int(time.clock()%self.nBins))):
# fill count of last second in buffer
self.wpm += self.counter - self.buffer[index]
self.buffer[index] = self.counter
# increment index to next bin
index = int(time.clock()%self.nBins)
# reset counter
self.counter = 0
# push to arduino
start_new_thread(self.arduPusherWPM,())
def on_data(self, status):
""" everytime a tweet comes in, counter is incremented """
self.counter += 1
start_new_thread(self.arduPusherClick,(1,))
# start new thread to print out the text
start_new_thread(self.printerTweet, (status,))
return
def printerTweet(self, tweet):
""" loads tweet as json and prints text to command line """
#print tweet
try:
text = json.loads(tweet)["text"]
print text.encode('cp850', errors='replace') + '\n'
except:
print 'could not get text of tweet'
def arduPusherWPM(self):
""" pushes current wpm to the arduino """
tmp = self.wpm if self.wpm < self.maxWPM else self.maxWPM
# send number as char
self.ardu.write(chr(int(tmp)))
def arduPusherClick(self, dummy = 0):
""" pushes signal a click signal to the arduino """
self.ardu.write(chr(self.CLICK))
def reset(self):
""" resets the counter and the buffer to start with new word """
self.buffer = np.zeros(self.nBins)
self.counter = 0
def main():
# load consumer token & secret key
f = open('auth.txt','r')
c_token, c_secret, a_token, a_secret = (line.strip() for line in f)
f.close()
# create OAuth handler and get access
auth = tweepy.OAuthHandler(c_token, c_secret)
auth.set_access_token(a_token, a_secret)
# set up listener
api = tweepy.API(auth)
listener = SListener(api)
# loop to enable changing the filter word
while 1:
try:
# get word to search for
track = raw_input("Look for word: ")
# leave whole program for quit_now
if track == 'quit_now':
sys.exit(0)
# start up stream
filterFor = [track]
stream = tweepy.Stream(auth, listener)
stream.filter(track = filterFor)
print "Streaming started..."
except KeyboardInterrupt:
# close stream, reset listener
stream.disconnect()
listener.reset()
except:
print "error!"
stream.disconnect()
break
if __name__ == '__main__':
main()