<s> """<STR_LIT>""" <EOL> import re <EOL> import json <EOL> from datetime import datetime <EOL> from jupyter_client import protocol_version_info <EOL> def code_to_line ( code , cursor_pos ) : <EOL> """<STR_LIT>""" <EOL> if not code : <EOL> return "<STR_LIT>" , <NUM_LIT:0> <EOL> for line in code . splitlines ( True ) : <EOL> n = len ( line ) <EOL> if cursor_pos > n : <EOL> cursor_pos -= n <EOL> else : <EOL> break <EOL> return line , cursor_pos <EOL> _match_bracket = re . compile ( r'<STR_LIT>' , re . UNICODE ) <EOL> _end_bracket = re . compile ( r'<STR_LIT>' , re . UNICODE ) <EOL> _identifier = re . compile ( r'<STR_LIT>' , re . I | re . UNICODE ) <EOL> def extract_oname_v4 ( code , cursor_pos ) : <EOL> """<STR_LIT>""" <EOL> line , _ = code_to_line ( code , cursor_pos ) <EOL> oldline = line <EOL> line = _match_bracket . sub ( u'<STR_LIT>' , line ) <EOL> while oldline != line : <EOL> oldline = line <EOL> line = _match_bracket . sub ( u'<STR_LIT>' , line ) <EOL> line = _end_bracket . sub ( '<STR_LIT>' , line ) <EOL> matches = _identifier . findall ( line ) <EOL> if matches : <EOL> return matches [ - <NUM_LIT:1> ] <EOL> else : <EOL> return '<STR_LIT>' <EOL> class Adapter ( object ) : <EOL> """<STR_LIT>""" <EOL> msg_type_map = { } <EOL> def update_header ( self , msg ) : <EOL> return msg <EOL> def update_metadata ( self , msg ) : <EOL> return msg <EOL> def update_msg_type ( self , msg ) : <EOL> header = msg [ '<STR_LIT>' ] <EOL> msg_type = header [ '<STR_LIT>' ] <EOL> if msg_type in self . msg_type_map : <EOL> msg [ '<STR_LIT>' ] = header [ '<STR_LIT>' ] = self . msg_type_map [ msg_type ] <EOL> return msg <EOL> def handle_reply_status_error ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> return msg <EOL> def __call__ ( self , msg ) : <EOL> msg = self . update_header ( msg ) <EOL> msg = self . update_metadata ( msg ) <EOL> msg = self . update_msg_type ( msg ) <EOL> header = msg [ '<STR_LIT>' ] <EOL> handler = getattr ( self , header [ '<STR_LIT>' ] , None ) <EOL> if handler is None : <EOL> return msg <EOL> if msg [ '<STR_LIT:content>' ] . get ( '<STR_LIT:status>' , None ) in { '<STR_LIT:error>' , '<STR_LIT>' } : <EOL> return self . handle_reply_status_error ( msg ) <EOL> return handler ( msg ) <EOL> def _version_str_to_list ( version ) : <EOL> """<STR_LIT>""" <EOL> v = [ ] <EOL> for part in version . split ( '<STR_LIT:.>' ) : <EOL> try : <EOL> v . append ( int ( part ) ) <EOL> except ValueError : <EOL> pass <EOL> return v <EOL> class V5toV4 ( Adapter ) : <EOL> """<STR_LIT>""" <EOL> version = '<STR_LIT>' <EOL> msg_type_map = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def update_header ( self , msg ) : <EOL> msg [ '<STR_LIT>' ] . pop ( '<STR_LIT:version>' , None ) <EOL> return msg <EOL> def kernel_info_reply ( self , msg ) : <EOL> v4c = { } <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if key in content : <EOL> v4c [ key ] = _version_str_to_list ( content [ key ] ) <EOL> if content . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' and '<STR_LIT>' in content : <EOL> v4c [ '<STR_LIT>' ] = _version_str_to_list ( content [ '<STR_LIT>' ] ) <EOL> language_info = content . get ( '<STR_LIT>' , { } ) <EOL> language = language_info . get ( '<STR_LIT:name>' , '<STR_LIT>' ) <EOL> v4c . setdefault ( '<STR_LIT>' , language ) <EOL> if '<STR_LIT:version>' in language_info : <EOL> v4c . setdefault ( '<STR_LIT>' , _version_str_to_list ( language_info [ '<STR_LIT:version>' ] ) ) <EOL> msg [ '<STR_LIT:content>' ] = v4c <EOL> return msg <EOL> def execute_request ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> content . setdefault ( '<STR_LIT>' , [ ] ) <EOL> return msg <EOL> def execute_reply ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> content . setdefault ( '<STR_LIT>' , { } ) <EOL> return msg <EOL> def complete_request ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> code = content [ '<STR_LIT:code>' ] <EOL> cursor_pos = content [ '<STR_LIT>' ] <EOL> line , cursor_pos = code_to_line ( code , cursor_pos ) <EOL> new_content = msg [ '<STR_LIT:content>' ] = { } <EOL> new_content [ '<STR_LIT:text>' ] = '<STR_LIT>' <EOL> new_content [ '<STR_LIT>' ] = line <EOL> new_content [ '<STR_LIT>' ] = None <EOL> new_content [ '<STR_LIT>' ] = cursor_pos <EOL> return msg <EOL> def complete_reply ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> cursor_start = content . pop ( '<STR_LIT>' ) <EOL> cursor_end = content . pop ( '<STR_LIT>' ) <EOL> match_len = cursor_end - cursor_start <EOL> content [ '<STR_LIT>' ] = content [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ : match_len ] <EOL> content . pop ( '<STR_LIT>' , None ) <EOL> return msg <EOL> def object_info_request ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> code = content [ '<STR_LIT:code>' ] <EOL> cursor_pos = content [ '<STR_LIT>' ] <EOL> line , _ = code_to_line ( code , cursor_pos ) <EOL> new_content = msg [ '<STR_LIT:content>' ] = { } <EOL> new_content [ '<STR_LIT>' ] = extract_oname_v4 ( code , cursor_pos ) <EOL> new_content [ '<STR_LIT>' ] = content [ '<STR_LIT>' ] <EOL> return msg <EOL> def object_info_reply ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> msg [ '<STR_LIT:content>' ] = { '<STR_LIT>' : False , '<STR_LIT>' : '<STR_LIT>' } <EOL> return msg <EOL> def stream ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> content [ '<STR_LIT:data>' ] = content . pop ( '<STR_LIT:text>' ) <EOL> return msg <EOL> def display_data ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> content . setdefault ( "<STR_LIT:source>" , "<STR_LIT>" ) <EOL> data = content [ '<STR_LIT:data>' ] <EOL> if '<STR_LIT:application/json>' in data : <EOL> try : <EOL> data [ '<STR_LIT:application/json>' ] = json . dumps ( data [ '<STR_LIT:application/json>' ] ) <EOL> except Exception : <EOL> pass <EOL> return msg <EOL> def input_request ( self , msg ) : <EOL> msg [ '<STR_LIT:content>' ] . pop ( '<STR_LIT:password>' , None ) <EOL> return msg <EOL> class V4toV5 ( Adapter ) : <EOL> """<STR_LIT>""" <EOL> version = '<STR_LIT>' <EOL> msg_type_map = { v : k for k , v in V5toV4 . msg_type_map . items ( ) } <EOL> def update_header ( self , msg ) : <EOL> msg [ '<STR_LIT>' ] [ '<STR_LIT:version>' ] = self . version <EOL> return msg <EOL> def kernel_info_reply ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if key in content : <EOL> content [ key ] = '<STR_LIT:.>' . join ( map ( str , content [ key ] ) ) <EOL> content . setdefault ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if content [ '<STR_LIT>' ] . startswith ( '<STR_LIT>' ) and '<STR_LIT>' in content : <EOL> content [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> content [ '<STR_LIT>' ] = content . pop ( '<STR_LIT>' ) <EOL> language = content . pop ( '<STR_LIT>' ) <EOL> language_info = content . setdefault ( '<STR_LIT>' , { } ) <EOL> language_info . setdefault ( '<STR_LIT:name>' , language ) <EOL> if '<STR_LIT>' in content : <EOL> language_version = '<STR_LIT:.>' . join ( map ( str , content . pop ( '<STR_LIT>' ) ) ) <EOL> language_info . setdefault ( '<STR_LIT:version>' , language_version ) <EOL> content [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return msg <EOL> def execute_request ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> user_variables = content . pop ( '<STR_LIT>' , [ ] ) <EOL> user_expressions = content . setdefault ( '<STR_LIT>' , { } ) <EOL> for v in user_variables : <EOL> user_expressions [ v ] = v <EOL> return msg <EOL> def execute_reply ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> user_expressions = content . setdefault ( '<STR_LIT>' , { } ) <EOL> user_variables = content . pop ( '<STR_LIT>' , { } ) <EOL> if user_variables : <EOL> user_expressions . update ( user_variables ) <EOL> for payload in content . get ( '<STR_LIT>' , [ ] ) : <EOL> if payload . get ( '<STR_LIT:source>' , None ) == '<STR_LIT>' and ( '<STR_LIT:text>' in payload ) : <EOL> if '<STR_LIT:data>' not in payload : <EOL> payload [ '<STR_LIT:data>' ] = { } <EOL> payload [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] = payload . pop ( '<STR_LIT:text>' ) <EOL> return msg <EOL> def complete_request ( self , msg ) : <EOL> old_content = msg [ '<STR_LIT:content>' ] <EOL> new_content = msg [ '<STR_LIT:content>' ] = { } <EOL> new_content [ '<STR_LIT:code>' ] = old_content [ '<STR_LIT>' ] <EOL> new_content [ '<STR_LIT>' ] = old_content [ '<STR_LIT>' ] <EOL> return msg <EOL> def complete_reply ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> new_content = msg [ '<STR_LIT:content>' ] = { '<STR_LIT:status>' : '<STR_LIT>' } <EOL> new_content [ '<STR_LIT>' ] = content [ '<STR_LIT>' ] <EOL> if content [ '<STR_LIT>' ] : <EOL> new_content [ '<STR_LIT>' ] = - len ( content [ '<STR_LIT>' ] ) <EOL> else : <EOL> new_content [ '<STR_LIT>' ] = None <EOL> new_content [ '<STR_LIT>' ] = None <EOL> new_content [ '<STR_LIT>' ] = { } <EOL> return msg <EOL> def inspect_request ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> name = content [ '<STR_LIT>' ] <EOL> new_content = msg [ '<STR_LIT:content>' ] = { } <EOL> new_content [ '<STR_LIT:code>' ] = name <EOL> new_content [ '<STR_LIT>' ] = len ( name ) <EOL> new_content [ '<STR_LIT>' ] = content [ '<STR_LIT>' ] <EOL> return msg <EOL> def inspect_reply ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> new_content = msg [ '<STR_LIT:content>' ] = { '<STR_LIT:status>' : '<STR_LIT>' } <EOL> found = new_content [ '<STR_LIT>' ] = content [ '<STR_LIT>' ] <EOL> new_content [ '<STR_LIT:data>' ] = data = { } <EOL> new_content [ '<STR_LIT>' ] = { } <EOL> if found : <EOL> lines = [ ] <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if content . get ( key , False ) : <EOL> lines . append ( content [ key ] ) <EOL> break <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if content . get ( key , False ) : <EOL> lines . append ( content [ key ] ) <EOL> break <EOL> if not lines : <EOL> lines . append ( "<STR_LIT>" ) <EOL> data [ '<STR_LIT>' ] = '<STR_LIT:\n>' . join ( lines ) <EOL> return msg <EOL> def stream ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> content [ '<STR_LIT:text>' ] = content . pop ( '<STR_LIT:data>' ) <EOL> return msg <EOL> def display_data ( self , msg ) : <EOL> content = msg [ '<STR_LIT:content>' ] <EOL> content . pop ( "<STR_LIT:source>" , None ) <EOL> data = content [ '<STR_LIT:data>' ] <EOL> if '<STR_LIT:application/json>' in data : <EOL> try : <EOL> data [ '<STR_LIT:application/json>' ] = json . loads ( data [ '<STR_LIT:application/json>' ] ) <EOL> except Exception : <EOL> pass <EOL> return msg <EOL> def input_request ( self , msg ) : <EOL> msg [ '<STR_LIT:content>' ] . setdefault ( '<STR_LIT:password>' , False ) <EOL> return msg <EOL> def adapt ( msg , to_version = protocol_version_info [ <NUM_LIT:0> ] ) : <EOL> """<STR_LIT>""" <EOL> header = msg [ '<STR_LIT>' ] <EOL> if '<STR_LIT:date>' not in header : <EOL> header [ '<STR_LIT:date>' ] = datetime . now ( ) . isoformat ( ) <EOL> if '<STR_LIT:version>' in header : <EOL> from_version = int ( header [ '<STR_LIT:version>' ] . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] ) <EOL> else : <EOL> from_version = <NUM_LIT:4> <EOL> adapter = adapters . get ( ( from_version , to_version ) , None ) <EOL> if adapter is None : <EOL> return msg <EOL> return adapter ( msg ) <EOL> adapters = { <EOL> ( <NUM_LIT:5> , <NUM_LIT:4> ) : V5toV4 ( ) , <EOL> ( <NUM_LIT:4> , <NUM_LIT:5> ) : V4toV5 ( ) , <EOL> } </s>
