<s> """<STR_LIT>""" <EOL> import sys <EOL> import time <EOL> from code import CommandCompiler <EOL> from datetime import datetime <EOL> from pprint import pprint <EOL> import zmq <EOL> from zmq . eventloop import ioloop , zmqstream <EOL> from IPython . utils . traitlets import Instance , List , Int , Dict , Set , Unicode , CBytes <EOL> from IPython . zmq . completer import KernelCompleter <EOL> from IPython . parallel . error import wrap_exception <EOL> from IPython . parallel . factory import SessionFactory <EOL> from IPython . parallel . util import serialize_object , unpack_apply_message , asbytes <EOL> def printer ( * args ) : <EOL> pprint ( args , stream = sys . __stdout__ ) <EOL> class _Passer ( zmqstream . ZMQStream ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> pass <EOL> def send ( self , * args , ** kwargs ) : <EOL> pass <EOL> send_multipart = send <EOL> class Kernel ( SessionFactory ) : <EOL> exec_lines = List ( Unicode , config = True , <EOL> help = "<STR_LIT>" ) <EOL> int_id = Int ( - <NUM_LIT:1> ) <EOL> bident = CBytes ( ) <EOL> ident = Unicode ( ) <EOL> def _ident_changed ( self , name , old , new ) : <EOL> self . bident = asbytes ( new ) <EOL> user_ns = Dict ( config = True , help = """<STR_LIT>""" ) <EOL> control_stream = Instance ( zmqstream . ZMQStream ) <EOL> task_stream = Instance ( zmqstream . ZMQStream ) <EOL> iopub_stream = Instance ( zmqstream . ZMQStream ) <EOL> client = Instance ( '<STR_LIT>' ) <EOL> shell_streams = List ( ) <EOL> compiler = Instance ( CommandCompiler , ( ) , { } ) <EOL> completer = Instance ( KernelCompleter ) <EOL> aborted = Set ( ) <EOL> shell_handlers = Dict ( ) <EOL> control_handlers = Dict ( ) <EOL> def _set_prefix ( self ) : <EOL> self . prefix = "<STR_LIT>" % self . int_id <EOL> def _connect_completer ( self ) : <EOL> self . completer = KernelCompleter ( self . user_ns ) <EOL> def __init__ ( self , ** kwargs ) : <EOL> super ( Kernel , self ) . __init__ ( ** kwargs ) <EOL> self . _set_prefix ( ) <EOL> self . _connect_completer ( ) <EOL> self . on_trait_change ( self . _set_prefix , '<STR_LIT:id>' ) <EOL> self . on_trait_change ( self . _connect_completer , '<STR_LIT>' ) <EOL> for msg_type in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] : <EOL> self . shell_handlers [ msg_type ] = getattr ( self , msg_type ) <EOL> for msg_type in [ '<STR_LIT>' , '<STR_LIT>' ] + list ( self . shell_handlers . keys ( ) ) : <EOL> self . control_handlers [ msg_type ] = getattr ( self , msg_type ) <EOL> self . _initial_exec_lines ( ) <EOL> def _wrap_exception ( self , method = None ) : <EOL> e_info = dict ( engine_uuid = self . ident , engine_id = self . int_id , method = method ) <EOL> content = wrap_exception ( e_info ) <EOL> return content <EOL> def _initial_exec_lines ( self ) : <EOL> s = _Passer ( ) <EOL> content = dict ( silent = True , user_variable = [ ] , user_expressions = [ ] ) <EOL> for line in self . exec_lines : <EOL> self . log . debug ( "<STR_LIT>" % line ) <EOL> content . update ( { '<STR_LIT:code>' : line } ) <EOL> msg = self . session . msg ( '<STR_LIT>' , content ) <EOL> self . execute_request ( s , [ ] , msg ) <EOL> def abort_queues ( self ) : <EOL> for stream in self . shell_streams : <EOL> if stream : <EOL> self . abort_queue ( stream ) <EOL> def abort_queue ( self , stream ) : <EOL> while True : <EOL> idents , msg = self . session . recv ( stream , zmq . NOBLOCK , content = True ) <EOL> if msg is None : <EOL> return <EOL> self . log . info ( "<STR_LIT>" ) <EOL> self . log . info ( str ( msg ) ) <EOL> msg_type = msg [ '<STR_LIT>' ] <EOL> reply_type = msg_type . split ( '<STR_LIT:_>' ) [ <NUM_LIT:0> ] + '<STR_LIT>' <EOL> reply_msg = self . session . send ( stream , reply_type , <EOL> content = { '<STR_LIT:status>' : '<STR_LIT>' } , parent = msg , ident = idents ) <EOL> self . log . debug ( str ( reply_msg ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> def abort_request ( self , stream , ident , parent ) : <EOL> """<STR_LIT>""" <EOL> msg_ids = parent [ '<STR_LIT:content>' ] . get ( '<STR_LIT>' , None ) <EOL> if isinstance ( msg_ids , str ) : <EOL> msg_ids = [ msg_ids ] <EOL> if not msg_ids : <EOL> self . abort_queues ( ) <EOL> for mid in msg_ids : <EOL> self . aborted . add ( str ( mid ) ) <EOL> content = dict ( status = '<STR_LIT>' ) <EOL> reply_msg = self . session . send ( stream , '<STR_LIT>' , content = content , <EOL> parent = parent , ident = ident ) <EOL> self . log . debug ( str ( reply_msg ) ) <EOL> def shutdown_request ( self , stream , ident , parent ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . abort_queues ( ) <EOL> except : <EOL> content = self . _wrap_exception ( '<STR_LIT>' ) <EOL> else : <EOL> content = dict ( parent [ '<STR_LIT:content>' ] ) <EOL> content [ '<STR_LIT:status>' ] = '<STR_LIT>' <EOL> msg = self . session . send ( stream , '<STR_LIT>' , <EOL> content = content , parent = parent , ident = ident ) <EOL> self . log . debug ( str ( msg ) ) <EOL> dc = ioloop . DelayedCallback ( lambda : sys . exit ( <NUM_LIT:0> ) , <NUM_LIT:1000> , self . loop ) <EOL> dc . start ( ) <EOL> def dispatch_control ( self , msg ) : <EOL> idents , msg = self . session . feed_identities ( msg , copy = False ) <EOL> try : <EOL> msg = self . session . unpack_message ( msg , content = True , copy = False ) <EOL> except : <EOL> self . log . error ( "<STR_LIT>" , exc_info = True ) <EOL> return <EOL> else : <EOL> self . log . debug ( "<STR_LIT>" , msg ) <EOL> header = msg [ '<STR_LIT>' ] <EOL> msg_id = header [ '<STR_LIT>' ] <EOL> handler = self . control_handlers . get ( msg [ '<STR_LIT>' ] , None ) <EOL> if handler is None : <EOL> self . log . error ( "<STR_LIT>" % msg [ '<STR_LIT>' ] ) <EOL> else : <EOL> handler ( self . control_stream , idents , msg ) <EOL> def check_dependencies ( self , dependencies ) : <EOL> if not dependencies : <EOL> return True <EOL> if len ( dependencies ) == <NUM_LIT:2> and dependencies [ <NUM_LIT:0> ] in '<STR_LIT>' . split ( ) : <EOL> anyorall = dependencies [ <NUM_LIT:0> ] <EOL> dependencies = dependencies [ <NUM_LIT:1> ] <EOL> else : <EOL> anyorall = '<STR_LIT:all>' <EOL> results = self . client . get_results ( dependencies , status_only = True ) <EOL> if results [ '<STR_LIT:status>' ] != '<STR_LIT>' : <EOL> return False <EOL> if anyorall == '<STR_LIT>' : <EOL> if not results [ '<STR_LIT>' ] : <EOL> return False <EOL> else : <EOL> if results [ '<STR_LIT>' ] : <EOL> return False <EOL> return True <EOL> def check_aborted ( self , msg_id ) : <EOL> return msg_id in self . aborted <EOL> def clear_request ( self , stream , idents , parent ) : <EOL> """<STR_LIT>""" <EOL> self . user_ns = { } <EOL> msg = self . session . send ( stream , '<STR_LIT>' , ident = idents , parent = parent , <EOL> content = dict ( status = '<STR_LIT>' ) ) <EOL> self . _initial_exec_lines ( ) <EOL> def execute_request ( self , stream , ident , parent ) : <EOL> self . log . debug ( '<STR_LIT>' % parent ) <EOL> try : <EOL> code = parent [ '<STR_LIT:content>' ] [ '<STR_LIT:code>' ] <EOL> except : <EOL> self . log . error ( "<STR_LIT>" % parent , exc_info = True ) <EOL> return <EOL> self . session . send ( self . iopub_stream , '<STR_LIT>' , { '<STR_LIT:code>' : code } , parent = parent , <EOL> ident = asbytes ( '<STR_LIT>' % self . prefix ) ) <EOL> started = datetime . now ( ) <EOL> try : <EOL> comp_code = self . compiler ( code , '<STR_LIT>' ) <EOL> if hasattr ( sys . displayhook , '<STR_LIT>' ) : <EOL> sys . displayhook . set_parent ( parent ) <EOL> sys . stdout . set_parent ( parent ) <EOL> sys . stderr . set_parent ( parent ) <EOL> exec ( comp_code , self . user_ns , self . user_ns ) <EOL> except : <EOL> exc_content = self . _wrap_exception ( '<STR_LIT>' ) <EOL> self . session . send ( self . iopub_stream , '<STR_LIT>' , exc_content , parent = parent , <EOL> ident = asbytes ( '<STR_LIT>' % self . prefix ) ) <EOL> reply_content = exc_content <EOL> else : <EOL> reply_content = { '<STR_LIT:status>' : '<STR_LIT>' } <EOL> reply_msg = self . session . send ( stream , '<STR_LIT>' , reply_content , parent = parent , <EOL> ident = ident , subheader = dict ( started = started ) ) <EOL> self . log . debug ( str ( reply_msg ) ) <EOL> if reply_msg [ '<STR_LIT:content>' ] [ '<STR_LIT:status>' ] == '<STR_LIT:error>' : <EOL> self . abort_queues ( ) <EOL> def complete_request ( self , stream , ident , parent ) : <EOL> matches = { '<STR_LIT>' : self . complete ( parent ) , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } <EOL> completion_msg = self . session . send ( stream , '<STR_LIT>' , <EOL> matches , parent , ident ) <EOL> def complete ( self , msg ) : <EOL> return self . completer . complete ( msg . content . line , msg . content . text ) <EOL> def apply_request ( self , stream , ident , parent ) : <EOL> stream . flush ( zmq . POLLOUT ) <EOL> try : <EOL> content = parent [ '<STR_LIT:content>' ] <EOL> bufs = parent [ '<STR_LIT>' ] <EOL> msg_id = parent [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> except : <EOL> self . log . error ( "<STR_LIT>" % parent , exc_info = True ) <EOL> return <EOL> sub = { '<STR_LIT>' : True , '<STR_LIT>' : self . ident , <EOL> '<STR_LIT>' : datetime . now ( ) } <EOL> try : <EOL> if hasattr ( sys . displayhook , '<STR_LIT>' ) : <EOL> sys . displayhook . set_parent ( parent ) <EOL> sys . stdout . set_parent ( parent ) <EOL> sys . stderr . set_parent ( parent ) <EOL> working = self . user_ns <EOL> prefix = "<STR_LIT:_>" + str ( msg_id ) . replace ( "<STR_LIT:->" , "<STR_LIT>" ) + "<STR_LIT:_>" <EOL> f , args , kwargs = unpack_apply_message ( bufs , working , copy = False ) <EOL> fname = getattr ( f , '<STR_LIT>' , '<STR_LIT:f>' ) <EOL> fname = prefix + "<STR_LIT:f>" <EOL> argname = prefix + "<STR_LIT:args>" <EOL> kwargname = prefix + "<STR_LIT>" <EOL> resultname = prefix + "<STR_LIT:result>" <EOL> ns = { fname : f , argname : args , kwargname : kwargs , resultname : None } <EOL> working . update ( ns ) <EOL> code = "<STR_LIT>" % ( resultname , fname , argname , kwargname ) <EOL> try : <EOL> exec ( code , working , working ) <EOL> result = working . get ( resultname ) <EOL> finally : <EOL> for key in ns . keys ( ) : <EOL> working . pop ( key ) <EOL> packed_result , buf = serialize_object ( result ) <EOL> result_buf = [ packed_result ] + buf <EOL> except : <EOL> exc_content = self . _wrap_exception ( '<STR_LIT>' ) <EOL> self . session . send ( self . iopub_stream , '<STR_LIT>' , exc_content , parent = parent , <EOL> ident = asbytes ( '<STR_LIT>' % self . prefix ) ) <EOL> reply_content = exc_content <EOL> result_buf = [ ] <EOL> if exc_content [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> sub [ '<STR_LIT>' ] = False <EOL> else : <EOL> reply_content = { '<STR_LIT:status>' : '<STR_LIT>' } <EOL> sub [ '<STR_LIT:status>' ] = reply_content [ '<STR_LIT:status>' ] <EOL> reply_msg = self . session . send ( stream , '<STR_LIT>' , reply_content , <EOL> parent = parent , ident = ident , buffers = result_buf , subheader = sub ) <EOL> sys . stdout . flush ( ) <EOL> sys . stderr . flush ( ) <EOL> def dispatch_queue ( self , stream , msg ) : <EOL> self . control_stream . flush ( ) <EOL> idents , msg = self . session . feed_identities ( msg , copy = False ) <EOL> try : <EOL> msg = self . session . unpack_message ( msg , content = True , copy = False ) <EOL> except : <EOL> self . log . error ( "<STR_LIT>" , exc_info = True ) <EOL> return <EOL> else : <EOL> self . log . debug ( "<STR_LIT>" , msg ) <EOL> header = msg [ '<STR_LIT>' ] <EOL> msg_id = header [ '<STR_LIT>' ] <EOL> if self . check_aborted ( msg_id ) : <EOL> self . aborted . remove ( msg_id ) <EOL> reply_type = msg [ '<STR_LIT>' ] . split ( '<STR_LIT:_>' ) [ <NUM_LIT:0> ] + '<STR_LIT>' <EOL> status = { '<STR_LIT:status>' : '<STR_LIT>' } <EOL> reply_msg = self . session . send ( stream , reply_type , subheader = status , <EOL> content = status , parent = msg , ident = idents ) <EOL> return <EOL> handler = self . shell_handlers . get ( msg [ '<STR_LIT>' ] , None ) <EOL> if handler is None : <EOL> self . log . error ( "<STR_LIT>" % msg [ '<STR_LIT>' ] ) <EOL> else : <EOL> handler ( stream , idents , msg ) <EOL> def start ( self ) : <EOL> if self . control_stream : <EOL> self . control_stream . on_recv ( self . dispatch_control , copy = False ) <EOL> self . control_stream . on_err ( printer ) <EOL> def make_dispatcher ( stream ) : <EOL> def dispatcher ( msg ) : <EOL> return self . dispatch_queue ( stream , msg ) <EOL> return dispatcher <EOL> for s in self . shell_streams : <EOL> s . on_recv ( make_dispatcher ( s ) , copy = False ) <EOL> s . on_err ( printer ) <EOL> if self . iopub_stream : <EOL> self . iopub_stream . on_err ( printer ) </s>
