<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> from prompt_toolkit . document import Document <EOL> from prompt_toolkit . eventloop . base import EventLoop <EOL> from prompt_toolkit . eventloop . posix_utils import PosixStdinReader <EOL> from six . moves import range <EOL> from . key_mappings import prompt_toolkit_key_to_vt100_key <EOL> from . screen import BetterScreen <EOL> from . stream import BetterStream <EOL> from . utils import set_terminal_size , pty_make_controlling_tty <EOL> import os <EOL> import resource <EOL> import signal <EOL> import sys <EOL> import time <EOL> import traceback <EOL> import datetime <EOL> __all__ = ( <EOL> '<STR_LIT>' , <EOL> ) <EOL> class Process ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , eventloop , invalidate , exec_func , bell_func = None , <EOL> done_callback = None , has_priority = None ) : <EOL> assert isinstance ( eventloop , EventLoop ) <EOL> assert callable ( invalidate ) <EOL> assert callable ( exec_func ) <EOL> assert bell_func is None or callable ( bell_func ) <EOL> assert done_callback is None or callable ( done_callback ) <EOL> assert has_priority is None or callable ( has_priority ) <EOL> self . eventloop = eventloop <EOL> self . invalidate = invalidate <EOL> self . exec_func = exec_func <EOL> self . done_callback = done_callback <EOL> self . has_priority = has_priority or ( lambda : True ) <EOL> self . pid = None <EOL> self . is_terminated = False <EOL> self . suspended = False <EOL> self . master , self . slave = os . openpty ( ) <EOL> self . _reader = PosixStdinReader ( self . master , errors = '<STR_LIT:replace>' ) <EOL> self . sx = <NUM_LIT:0> <EOL> self . sy = <NUM_LIT:0> <EOL> self . screen = BetterScreen ( self . sx , self . sy , <EOL> write_process_input = self . write_input , <EOL> bell_func = bell_func ) <EOL> self . stream = BetterStream ( self . screen ) <EOL> self . stream . attach ( self . screen ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . set_size ( <NUM_LIT> , <NUM_LIT> ) <EOL> self . _start ( ) <EOL> self . _connect_reader ( ) <EOL> self . _waitpid ( ) <EOL> @ classmethod <EOL> def from_command ( cls , eventloop , invalidate , command , done_callback , <EOL> bell_func = None , before_exec_func = None , has_priority = None ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( command , list ) <EOL> def execv ( ) : <EOL> if before_exec_func : <EOL> before_exec_func ( ) <EOL> for p in os . environ [ '<STR_LIT>' ] . split ( '<STR_LIT::>' ) : <EOL> path = os . path . join ( p , command [ <NUM_LIT:0> ] ) <EOL> if os . path . exists ( path ) and os . access ( path , os . X_OK ) : <EOL> os . execv ( path , command ) <EOL> return cls ( eventloop , invalidate , execv , <EOL> bell_func = bell_func , done_callback = done_callback , <EOL> has_priority = has_priority ) <EOL> def _start ( self ) : <EOL> """<STR_LIT>""" <EOL> pid = os . fork ( ) <EOL> if pid == <NUM_LIT:0> : <EOL> self . _in_child ( ) <EOL> elif pid > <NUM_LIT:0> : <EOL> os . close ( self . slave ) <EOL> self . slave = None <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . pid = pid <EOL> def _waitpid ( self ) : <EOL> """<STR_LIT>""" <EOL> def wait_for_finished ( ) : <EOL> "<STR_LIT>" <EOL> os . waitpid ( self . pid , <NUM_LIT:0> ) <EOL> self . eventloop . call_from_executor ( done ) <EOL> def done ( ) : <EOL> "<STR_LIT>" <EOL> os . close ( self . master ) <EOL> self . eventloop . remove_reader ( self . master ) <EOL> self . master = None <EOL> self . is_terminated = True <EOL> self . done_callback ( ) <EOL> self . eventloop . run_in_executor ( wait_for_finished ) <EOL> def set_size ( self , width , height ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( width , int ) <EOL> assert isinstance ( height , int ) <EOL> if self . master is not None : <EOL> if ( self . sx , self . sy ) != ( width , height ) : <EOL> set_terminal_size ( self . master , height , width ) <EOL> self . screen . resize ( lines = height , columns = width ) <EOL> self . screen . lines = height <EOL> self . screen . columns = width <EOL> self . sx = width <EOL> self . sy = height <EOL> def _in_child ( self ) : <EOL> "<STR_LIT>" <EOL> os . close ( self . master ) <EOL> signal . signal ( signal . SIGWINCH , <NUM_LIT:0> ) <EOL> pty_make_controlling_tty ( self . slave ) <EOL> os . dup2 ( self . slave , <NUM_LIT:0> ) <EOL> os . dup2 ( self . slave , <NUM_LIT:1> ) <EOL> os . dup2 ( self . slave , <NUM_LIT:2> ) <EOL> try : <EOL> self . _close_file_descriptors ( ) <EOL> self . exec_func ( ) <EOL> except Exception : <EOL> traceback . print_exc ( ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> os . _exit ( <NUM_LIT:1> ) <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> def _close_file_descriptors ( self ) : <EOL> max_fd = resource . getrlimit ( resource . RLIMIT_NOFILE ) [ - <NUM_LIT:1> ] <EOL> try : <EOL> os . closerange ( <NUM_LIT:3> , max_fd ) <EOL> except OverflowError : <EOL> os . closerange ( <NUM_LIT:3> , <NUM_LIT> ) <EOL> def write_input ( self , data , paste = False ) : <EOL> """<STR_LIT>""" <EOL> if paste and self . screen . bracketed_paste_enabled : <EOL> data = '<STR_LIT>' + data + '<STR_LIT>' <EOL> self . write_bytes ( data . encode ( '<STR_LIT:utf-8>' ) ) <EOL> def write_bytes ( self , data ) : <EOL> while self . master is not None : <EOL> try : <EOL> os . write ( self . master , data ) <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:4> : <EOL> continue <EOL> return <EOL> def write_key ( self , key ) : <EOL> """<STR_LIT>""" <EOL> data = prompt_toolkit_key_to_vt100_key ( <EOL> key , application_mode = self . screen . in_application_mode ) <EOL> self . write_input ( data ) <EOL> def _connect_reader ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . master is not None : <EOL> self . eventloop . add_reader ( self . master , self . _read ) <EOL> def _read ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . _reader . read ( <NUM_LIT> ) <EOL> if not self . _reader . closed : <EOL> def process ( ) : <EOL> self . stream . feed ( d ) <EOL> self . invalidate ( ) <EOL> if self . has_priority ( ) : <EOL> process ( ) <EOL> else : <EOL> self . eventloop . remove_reader ( self . master ) <EOL> def do_asap ( ) : <EOL> "<STR_LIT>" <EOL> process ( ) <EOL> self . _connect_reader ( ) <EOL> timestamp = datetime . datetime . now ( ) + datetime . timedelta ( seconds = <NUM_LIT:1> ) <EOL> self . eventloop . call_from_executor ( <EOL> do_asap , _max_postpone_until = timestamp ) <EOL> else : <EOL> self . eventloop . remove_reader ( self . master ) <EOL> def suspend ( self ) : <EOL> """<STR_LIT>""" <EOL> self . suspended = True <EOL> self . eventloop . remove_reader ( self . master ) <EOL> def resume ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . suspended and self . master is not None : <EOL> self . _connect_reader ( ) <EOL> self . suspended = False <EOL> def get_cwd ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_cwd_for_pid ( self . pid ) <EOL> def get_name ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . master is not None : <EOL> return get_name_for_fd ( self . master ) <EOL> def send_signal ( self , signal ) : <EOL> "<STR_LIT>" <EOL> assert isinstance ( signal , int ) , type ( signal ) <EOL> if self . pid and not self . is_terminated : <EOL> try : <EOL> os . kill ( self . pid , signal ) <EOL> except OSError : <EOL> pass <EOL> def create_copy_document ( self ) : <EOL> """<STR_LIT>""" <EOL> data_buffer = self . screen . data_buffer <EOL> text = [ ] <EOL> token_lists = [ ] <EOL> first_row = min ( data_buffer . keys ( ) ) <EOL> last_row = max ( data_buffer . keys ( ) ) <EOL> def token_has_no_background ( token ) : <EOL> try : <EOL> return token [ <NUM_LIT:2> ] is None <EOL> except IndexError : <EOL> return True <EOL> for lineno in range ( first_row , last_row + <NUM_LIT:1> ) : <EOL> token_list = [ ] <EOL> row = data_buffer [ lineno ] <EOL> max_column = max ( row . keys ( ) ) if row else <NUM_LIT:0> <EOL> row_data = [ row [ x ] for x in range ( <NUM_LIT:0> , max_column + <NUM_LIT:1> ) ] <EOL> while ( row_data and row_data [ - <NUM_LIT:1> ] . char . isspace ( ) and <EOL> token_has_no_background ( row_data [ - <NUM_LIT:1> ] . token ) ) : <EOL> row_data . pop ( ) <EOL> char_iter = iter ( range ( len ( row_data ) ) ) <EOL> for x in char_iter : <EOL> c = row [ x ] <EOL> text . append ( c . char ) <EOL> token_list . append ( ( c . token , c . char ) ) <EOL> if c . width == <NUM_LIT:2> : <EOL> try : <EOL> next ( char_iter ) <EOL> except StopIteration : <EOL> pass <EOL> token_lists . append ( token_list ) <EOL> text . append ( '<STR_LIT:\n>' ) <EOL> def get_tokens_for_line ( lineno ) : <EOL> try : <EOL> return token_lists [ lineno ] <EOL> except IndexError : <EOL> return [ ] <EOL> d = Document ( text = '<STR_LIT>' . join ( text ) ) <EOL> return Document ( text = d . text , <EOL> cursor_position = d . translate_row_col_to_index ( <EOL> row = self . screen . pt_screen . cursor_position . y , <EOL> col = self . screen . pt_screen . cursor_position . x ) ) , get_tokens_for_line <EOL> def get_cwd_for_pid ( pid ) : <EOL> """<STR_LIT>""" <EOL> if sys . platform in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> try : <EOL> return os . readlink ( '<STR_LIT>' % pid ) <EOL> except OSError : <EOL> pass <EOL> if sys . platform in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> def get_name_for_fd ( fd ) : <EOL> """<STR_LIT>""" <EOL> pgrp = os . tcgetpgrp ( fd ) <EOL> try : <EOL> with open ( '<STR_LIT>' % pgrp , '<STR_LIT:rb>' ) as f : <EOL> return f . read ( ) . decode ( '<STR_LIT:utf-8>' , '<STR_LIT:ignore>' ) . partition ( '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> except IOError : <EOL> pass <EOL> elif sys . platform == '<STR_LIT>' : <EOL> from . darwin import get_proc_name <EOL> def get_name_for_fd ( fd ) : <EOL> """<STR_LIT>""" <EOL> pgrp = os . tcgetpgrp ( fd ) <EOL> try : <EOL> return get_proc_name ( pgrp ) <EOL> except IOError : <EOL> pass <EOL> else : <EOL> def get_name_for_fd ( fd ) : <EOL> """<STR_LIT>""" <EOL> return </s>
