<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> from . . keys import Keys <EOL> from . . utils import Callback <EOL> from prompt_toolkit . buffer import EditReadOnlyBuffer <EOL> import weakref <EOL> __all__ = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> class KeyPress ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , key , data ) : <EOL> self . key = key <EOL> self . data = data <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . key , self . data ) <EOL> def __eq__ ( self , other ) : <EOL> return self . key == other . key and self . data == other . data <EOL> class InputProcessor ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , registry , cli_ref ) : <EOL> self . _registry = registry <EOL> self . _cli_ref = cli_ref <EOL> self . reset ( ) <EOL> self . beforeKeyPress = Callback ( ) <EOL> self . afterKeyPress = Callback ( ) <EOL> def reset ( self ) : <EOL> self . _previous_key_sequence = None <EOL> self . _previous_handler = None <EOL> self . _process_coroutine = self . _process ( ) <EOL> self . _process_coroutine . send ( None ) <EOL> self . arg = None <EOL> def _get_matches ( self , key_presses ) : <EOL> """<STR_LIT>""" <EOL> keys = tuple ( k . key for k in key_presses ) <EOL> cli = self . _cli_ref ( ) <EOL> with_mode = [ b for b in self . _registry . get_bindings_for_keys ( keys ) if b . filter ( cli ) ] <EOL> if with_mode : <EOL> return with_mode <EOL> keys_any = tuple ( keys [ : - <NUM_LIT:1> ] + ( Keys . Any , ) ) <EOL> with_mode_any = [ b for b in self . _registry . get_bindings_for_keys ( keys_any ) if b . filter ( cli ) ] <EOL> if with_mode_any : <EOL> return with_mode_any <EOL> return [ ] <EOL> def _is_prefix_of_longer_match ( self , key_presses ) : <EOL> """<STR_LIT>""" <EOL> keys = tuple ( k . key for k in key_presses ) <EOL> cli = self . _cli_ref ( ) <EOL> filters = set ( b . filter for b in self . _registry . get_bindings_starting_with_keys ( keys ) ) <EOL> return any ( f ( cli ) for f in filters ) <EOL> def _process ( self ) : <EOL> """<STR_LIT>""" <EOL> buffer = [ ] <EOL> retry = False <EOL> while True : <EOL> if retry : <EOL> retry = False <EOL> else : <EOL> buffer . append ( ( yield ) ) <EOL> if buffer : <EOL> is_prefix_of_longer_match = self . _is_prefix_of_longer_match ( buffer ) <EOL> matches = self . _get_matches ( buffer ) <EOL> if matches and matches [ - <NUM_LIT:1> ] . eager ( self . _cli_ref ( ) ) : <EOL> is_prefix_of_longer_match = False <EOL> if not is_prefix_of_longer_match and matches : <EOL> self . _call_handler ( matches [ - <NUM_LIT:1> ] , key_sequence = buffer ) <EOL> buffer = [ ] <EOL> elif not is_prefix_of_longer_match and not matches : <EOL> retry = True <EOL> found = False <EOL> for i in range ( len ( buffer ) , <NUM_LIT:0> , - <NUM_LIT:1> ) : <EOL> matches = self . _get_matches ( buffer [ : i ] ) <EOL> if matches : <EOL> self . _call_handler ( matches [ - <NUM_LIT:1> ] , key_sequence = buffer [ : i ] ) <EOL> buffer = buffer [ i : ] <EOL> found = True <EOL> if not found : <EOL> buffer = buffer [ <NUM_LIT:1> : ] <EOL> def feed_key ( self , key_press ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( key_press , KeyPress ) <EOL> if key_press . key != Keys . CPRResponse : <EOL> self . beforeKeyPress . fire ( ) <EOL> self . _process_coroutine . send ( key_press ) <EOL> if key_press . key != Keys . CPRResponse : <EOL> self . afterKeyPress . fire ( ) <EOL> def _call_handler ( self , handler , key_sequence = None ) : <EOL> arg = self . arg <EOL> self . arg = None <EOL> is_repeat = handler == self . _previous_handler <EOL> try : <EOL> event = Event ( weakref . ref ( self ) , arg = arg , key_sequence = key_sequence , <EOL> previous_key_sequence = self . _previous_key_sequence , <EOL> is_repeat = is_repeat ) <EOL> handler . call ( event ) <EOL> self . _registry . on_handler_called . fire ( event ) <EOL> except EditReadOnlyBuffer : <EOL> pass <EOL> self . _previous_key_sequence = key_sequence <EOL> self . _previous_handler = handler <EOL> cli = self . _cli_ref ( ) <EOL> if cli and handler . invalidate_ui ( cli ) : <EOL> cli . invalidate ( ) <EOL> class Event ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , input_processor_ref , arg = None , key_sequence = None , <EOL> previous_key_sequence = None , is_repeat = False ) : <EOL> self . _input_processor_ref = input_processor_ref <EOL> self . key_sequence = key_sequence <EOL> self . previous_key_sequence = previous_key_sequence <EOL> self . is_repeat = is_repeat <EOL> self . _arg = arg <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . arg , self . key_sequence , self . is_repeat ) <EOL> @ property <EOL> def data ( self ) : <EOL> return self . key_sequence [ - <NUM_LIT:1> ] . data <EOL> @ property <EOL> def input_processor ( self ) : <EOL> return self . _input_processor_ref ( ) <EOL> @ property <EOL> def cli ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . input_processor . _cli_ref ( ) <EOL> @ property <EOL> def current_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . cli . current_buffer <EOL> @ property <EOL> def arg ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _arg or <NUM_LIT:1> <EOL> def append_to_arg_count ( self , data ) : <EOL> """<STR_LIT>""" <EOL> assert data in '<STR_LIT>' <EOL> current = self . _arg <EOL> if current is None : <EOL> if data == '<STR_LIT:->' : <EOL> data = '<STR_LIT>' <EOL> result = int ( data ) <EOL> else : <EOL> result = int ( "<STR_LIT>" % ( current , data ) ) <EOL> if int ( result ) >= <NUM_LIT> : <EOL> result = None <EOL> self . input_processor . arg = result </s>
