<s> """<STR_LIT>""" <EOL> import json <EOL> import logging <EOL> import Queue <EOL> import sys <EOL> import threading <EOL> import uuid <EOL> from pusherclient import Pusher <EOL> import requests <EOL> import websocket <EOL> from common import public_creds <EOL> from pi import simple_pusher <EOL> CONFIG_FILE = '<STR_LIT>' <EOL> APPENGINE_ADDRESS = '<STR_LIT>' % public_creds . appengine_app_id <EOL> LOCAL_ADDRESS = '<STR_LIT>' <EOL> EVENT_PATH = '<STR_LIT>' <EOL> AUTH_PATH = '<STR_LIT>' <EOL> def read_or_make_config ( ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> with open ( CONFIG_FILE ) as config_file : <EOL> return config_file . read ( ) . split ( '<STR_LIT:U+002C>' ) <EOL> except : <EOL> proxy_id = str ( uuid . uuid4 ( ) . get_hex ( ) ) <EOL> proxy_secret = str ( uuid . uuid4 ( ) . get_hex ( ) ) <EOL> with open ( CONFIG_FILE , '<STR_LIT:w>' ) as config_file : <EOL> config_file . write ( '<STR_LIT>' % ( proxy_id , proxy_secret ) ) <EOL> return ( proxy_id , proxy_secret ) <EOL> class PushRPC ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , callback , args ) : <EOL> self . _proxy_id , self . _proxy_secret = read_or_make_config ( ) <EOL> logging . info ( '<STR_LIT>' , self . _proxy_id ) <EOL> self . _args = args <EOL> self . _exiting = False <EOL> self . _events = Queue . Queue ( ) <EOL> self . _events_thread = threading . Thread ( target = self . _post_events_loop ) <EOL> self . _events_thread . daemon = True <EOL> self . _events_thread . start ( ) <EOL> self . _callback = callback <EOL> if args . local : <EOL> self . _websocket_connection = None <EOL> self . _websocket_thread = threading . Thread ( target = self . _local_websocket ) <EOL> self . _websocket_thread . start ( ) <EOL> else : <EOL> self . _pusher = Pusher ( public_creds . pusher_key , <EOL> auth_callback = self . _pusher_auth_callback , <EOL> log_level = logging . ERROR ) <EOL> self . _pusher . connection . bind ( <EOL> '<STR_LIT>' , <EOL> self . _connect_handler ) <EOL> self . _pusher . connect ( ) <EOL> def _local_websocket ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _websocket_connection = websocket . create_connection ( <EOL> "<STR_LIT>" % simple_pusher . WEBSOCKET_PORT ) <EOL> request = json . dumps ( { '<STR_LIT>' : '<STR_LIT>' % self . _proxy_id } ) <EOL> self . _websocket_connection . send ( request ) <EOL> while True : <EOL> result = self . _websocket_connection . recv ( ) <EOL> self . _callback_handler ( result ) <EOL> def _pusher_auth_callback ( self , socket_id , channel_name ) : <EOL> params = { '<STR_LIT>' : socket_id , '<STR_LIT>' : channel_name } <EOL> response = self . _make_request ( APPENGINE_ADDRESS , AUTH_PATH , params = params ) <EOL> response = response . json ( ) <EOL> return response [ '<STR_LIT>' ] <EOL> def _make_request ( self , server , path , method = '<STR_LIT:GET>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> response = requests . request ( <EOL> method , server + path , <EOL> auth = ( self . _proxy_id , self . _proxy_secret ) , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' } , <EOL> ** kwargs ) <EOL> response . raise_for_status ( ) <EOL> return response <EOL> def _connect_handler ( self , _ ) : <EOL> channel_name = '<STR_LIT>' % self . _proxy_id <EOL> channel = self . _pusher . subscribe ( channel_name ) <EOL> channel . bind ( '<STR_LIT>' , self . _callback_handler ) <EOL> def _callback_handler ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> events = json . loads ( data ) <EOL> except ValueError : <EOL> logging . error ( '<STR_LIT>' , exc_info = sys . exc_info ( ) ) <EOL> return <EOL> try : <EOL> self . _callback ( events ) <EOL> except Exception : <EOL> logging . error ( '<STR_LIT>' , exc_info = sys . exc_info ( ) ) <EOL> def send_event ( self , event ) : <EOL> self . _events . put ( event ) <EOL> def _get_batch_of_events ( self , max_size = <NUM_LIT:20> ) : <EOL> """<STR_LIT>""" <EOL> events = [ ] <EOL> while len ( events ) < max_size : <EOL> try : <EOL> block = len ( events ) == <NUM_LIT:0> <EOL> event = self . _events . get ( block ) <EOL> if event is None : <EOL> return None <EOL> events . append ( event ) <EOL> except Queue . Empty : <EOL> break <EOL> assert events <EOL> return events <EOL> def _post_events_loop ( self ) : <EOL> """<STR_LIT>""" <EOL> logging . info ( '<STR_LIT>' ) <EOL> while not self . _exiting : <EOL> events = self . _get_batch_of_events ( ) <EOL> if events is None : <EOL> break <EOL> try : <EOL> self . _post_events_once ( events ) <EOL> except Exception : <EOL> logging . error ( '<STR_LIT>' , <EOL> exc_info = sys . exc_info ( ) ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> def _post_events_once ( self , events ) : <EOL> """<STR_LIT>""" <EOL> logging . info ( '<STR_LIT>' , len ( events ) ) <EOL> try : <EOL> server_address = LOCAL_ADDRESS if self . _args . local else APPENGINE_ADDRESS <EOL> self . _make_request ( server_address , EVENT_PATH , <EOL> method = '<STR_LIT:POST>' , data = json . dumps ( events ) ) <EOL> except : <EOL> logging . error ( '<STR_LIT>' , exc_info = sys . exc_info ( ) ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _exiting = True <EOL> self . _events . put ( None ) <EOL> self . _events_thread . join ( ) <EOL> if self . _args . local : <EOL> self . _websocket_connection . close ( ) <EOL> self . _websocket_thread . join ( ) <EOL> else : <EOL> self . _pusher . disconnect ( ) </s>
