-import random, itertools
+import random
+import itertools
from functools import wraps
from termcolor import *
from pyfiglet import figlet_format
+from functools import reduce
-grey = lambda x: colored(x, 'grey', attrs=['bold'])
-red = lambda x: colored(x, 'red', attrs=['bold'])
-green = lambda x: colored(x, 'green', attrs=['bold'])
-yellow = lambda x: colored(x, 'yellow', attrs=['bold'])
-blue = lambda x: colored(x, 'blue', attrs=['bold'])
+grey = lambda x: colored(x, 'grey', attrs=['bold'])
+red = lambda x: colored(x, 'red', attrs=['bold'])
+green = lambda x: colored(x, 'green', attrs=['bold'])
+yellow = lambda x: colored(x, 'yellow', attrs=['bold'])
+blue = lambda x: colored(x, 'blue', attrs=['bold'])
magenta = lambda x: colored(x, 'magenta', attrs=['bold'])
-cyan = lambda x: colored(x, 'cyan', attrs=['bold'])
-white = lambda x: colored(x, 'white', attrs=['bold'])
+cyan = lambda x: colored(x, 'cyan', attrs=['bold'])
+white = lambda x: colored(x, 'white', attrs=['bold'])
-on_grey = lambda x: colored(x, 'white', 'on_grey', attrs=['bold'])
-on_red = lambda x: colored(x, 'white', 'on_red', attrs=['bold'])
-on_green = lambda x: colored(x, 'white', 'on_green', attrs=['bold'])
-on_yellow = lambda x: colored(x, 'white', 'on_yellow', attrs=['bold'])
-on_blue = lambda x: colored(x, 'white', 'on_blue', attrs=['bold'])
+on_grey = lambda x: colored(x, 'white', 'on_grey', attrs=['bold'])
+on_red = lambda x: colored(x, 'white', 'on_red', attrs=['bold'])
+on_green = lambda x: colored(x, 'white', 'on_green', attrs=['bold'])
+on_yellow = lambda x: colored(x, 'white', 'on_yellow', attrs=['bold'])
+on_blue = lambda x: colored(x, 'white', 'on_blue', attrs=['bold'])
on_magenta = lambda x: colored(x, 'white', 'on_magenta', attrs=['bold'])
-on_cyan = lambda x: colored(x, 'white', 'on_cyan', attrs=['bold'])
-on_white = lambda x: colored(x, 'white', 'on_white', attrs=['bold'])
+on_cyan = lambda x: colored(x, 'white', 'on_cyan', attrs=['bold'])
+on_white = lambda x: colored(x, 'white', 'on_white', attrs=['bold'])
colors_shuffle = [grey, red, green, yellow, blue, magenta, cyan]
-background_shuffle = [on_grey, on_red, on_green, on_yellow, on_blue, on_magenta, on_cyan]
+background_shuffle = [
+ on_grey,
+ on_red,
+ on_green,
+ on_yellow,
+ on_blue,
+ on_magenta,
+ on_cyan]
cyc = itertools.cycle(colors_shuffle[1:])
"""
Print a string with ordered color with each character
"""
- c = [colors_shuffle[i%7](s[i]) for i in xrange(len(s))]
- return reduce(lambda x,y: x+y, c)
+ c = [colors_shuffle[i % 7](s[i]) for i in xrange(len(s))]
+ return reduce(lambda x, y: x + y, c)
+
def random_rainbow(s):
"""
Print a string with random color with each character
"""
c = [random.choice(colors_shuffle)(i) for i in s]
- return reduce(lambda x,y: x+y, c)
+ return reduce(lambda x, y: x + y, c)
+
def Memoize(func):
"""
Memoize decorator
"""
cache = {}
+
@wraps(func)
def wrapper(*args):
if args not in cache:
return cache[args]
return wrapper
+
@Memoize
def cycle_color(s):
"""
"""
return next(cyc)(s)
+
def ascii_art():
"""
Draw the Ascii Art
"""
fi = figlet_format('Rainbow Stream', font='doom')
print('\n'.join(
- [next(cyc)(i) for i in fi.split('\n')]
- )
+ [next(cyc)(i) for i in fi.split('\n')]
+ )
)
-
from __future__ import print_function
-import os, os.path, argparse
+import os
+import os.path
+import argparse
from twitter.stream import TwitterStream, Timeout, HeartbeatTimeout, Hangup
from twitter.oauth import OAuth, read_token_file
from twitter.util import printNicely
from dateutil import parser
-from colors import *
-from config import *
+from .colors import *
+from .config import *
+
def draw(t):
"""
# Format info
user = cycle_color(name + ' ' + '@' + screen_name + ' ')
- clock = grey('['+ time + ']')
+ clock = grey('[' + time + ']')
tweet = text.split()
- tweet = map(lambda x: grey(x) if x=='RT' else x, tweet)
- tweet = map(lambda x: cycle_color(x) if x[0]=='@' else x, tweet)
- tweet = map(lambda x: cyan(x) if x[0:7]=='http://' else x, tweet)
+ tweet = map(lambda x: grey(x) if x == 'RT' else x, tweet)
+ tweet = map(lambda x: cycle_color(x) if x[0] == '@' else x, tweet)
+ tweet = map(lambda x: cyan(x) if x[0:7] == 'http://' else x, tweet)
tweet = ' '.join(tweet)
# Draw rainbow
line1 = u"{u:>{uw}}:".format(
- u = user,
- uw = len(user) + 2,
+ u=user,
+ uw=len(user) + 2,
)
line2 = u"{c:>{cw}}".format(
- c = clock,
- cw = len(clock) + 2,
+ c=clock,
+ cw=len(clock) + 2,
)
line3 = ' ' + tweet
line4 = '\n'
parser = argparse.ArgumentParser(description=__doc__ or "")
- parser.add_argument('-to', '--timeout', help='Timeout for the stream (seconds).')
- parser.add_argument('-ht', '--heartbeat-timeout', help='Set heartbeat timeout.', default=90)
- parser.add_argument('-nb', '--no-block', action='store_true', help='Set stream to non-blocking.')
- parser.add_argument('-tt', '--track-keywords', help='Search the stream for specific text.')
+ parser.add_argument(
+ '-to',
+ '--timeout',
+ help='Timeout for the stream (seconds).')
+ parser.add_argument(
+ '-ht',
+ '--heartbeat-timeout',
+ help='Set heartbeat timeout.',
+ default=90)
+ parser.add_argument(
+ '-nb',
+ '--no-block',
+ action='store_true',
+ help='Set stream to non-blocking.')
+ parser.add_argument(
+ '-tt',
+ '--track-keywords',
+ help='Search the stream for specific text.')
return parser.parse_args()
ascii_art()
# When using rainbow stream you must authorize.
- twitter_credential = os.environ.get('HOME', os.environ.get('USERPROFILE', '')) + os.sep + '.rainbow_oauth'
+ twitter_credential = os.environ.get(
+ 'HOME',
+ os.environ.get(
+ 'USERPROFILE',
+ '')) + os.sep + '.rainbow_oauth'
if not os.path.exists(twitter_credential):
oauth_dance("Rainbow Stream",
CONSUMER_KEY,
CONSUMER_SECRET,
twitter_credential)
oauth_token, oauth_token_secret = read_token_file(twitter_credential)
- auth = OAuth(oauth_token, oauth_token_secret, CONSUMER_KEY, CONSUMER_SECRET)
+ auth = OAuth(
+ oauth_token,
+ oauth_token_secret,
+ CONSUMER_KEY,
+ CONSUMER_SECRET)
# These arguments are optional:
stream_args = dict(
query_args['track'] = args.track_keywords
# Get stream
- stream = TwitterStream(auth=auth, domain='userstream.twitter.com', **stream_args)
+ stream = TwitterStream(
+ auth=auth,
+ domain='userstream.twitter.com',
+ **stream_args)
tweet_iter = stream.user(**query_args)
# Iterate over the sample stream.
elif tweet is Hangup:
printNicely("-- Hangup --")
elif tweet.get('text'):
- line1, line2, line3, line4 = draw(t = tweet)
+ line1, line2, line3, line4 = draw(t=tweet)
printNicely(line1)
printNicely(line2)
printNicely(line3)