<s> """<STR_LIT>""" <EOL> import io <EOL> import re <EOL> import os <EOL> import sys <EOL> import tty <EOL> import pty <EOL> import time <EOL> import array <EOL> import fcntl <EOL> import select <EOL> import signal <EOL> import termios <EOL> import tempfile <EOL> import threading <EOL> MODE_NUMS = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> START_ALTERNATE_MODE = frozenset ( '<STR_LIT>' . format ( i ) . encode ( ) for i in MODE_NUMS ) <EOL> END_ALTERNATE_MODE = frozenset ( '<STR_LIT>' . format ( i ) . encode ( ) for i in MODE_NUMS ) <EOL> ALTERNATE_MODE_FLAGS = tuple ( START_ALTERNATE_MODE ) + tuple ( END_ALTERNATE_MODE ) <EOL> RE_HIDDEN = re . compile ( b'<STR_LIT>' ) <EOL> RE_COLOR = re . compile ( b'<STR_LIT>' ) <EOL> def _findfirst ( s , substrs ) : <EOL> """<STR_LIT>""" <EOL> i = len ( s ) <EOL> result = None <EOL> for substr in substrs : <EOL> pos = s . find ( substr ) <EOL> if - <NUM_LIT:1> < pos < i : <EOL> i = pos <EOL> result = substr <EOL> return i , result <EOL> def _on_main_thread ( ) : <EOL> """<STR_LIT>""" <EOL> return threading . current_thread ( ) is threading . main_thread ( ) <EOL> def _find_error_code ( e ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( e , PermissionError ) : <EOL> code = <NUM_LIT> <EOL> elif isinstance ( e , FileNotFoundError ) : <EOL> code = <NUM_LIT> <EOL> else : <EOL> code = <NUM_LIT:1> <EOL> return code <EOL> class TeePTY ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , bufsize = <NUM_LIT> , remove_color = True , encoding = '<STR_LIT:utf-8>' , <EOL> errors = '<STR_LIT:strict>' ) : <EOL> """<STR_LIT>""" <EOL> self . bufsize = bufsize <EOL> self . pid = self . master_fd = None <EOL> self . _in_alt_mode = False <EOL> self . remove_color = remove_color <EOL> self . encoding = encoding <EOL> self . errors = errors <EOL> self . buffer = io . BytesIO ( ) <EOL> self . wcode = None <EOL> self . _temp_stdin = None <EOL> def __str__ ( self ) : <EOL> return self . buffer . getvalue ( ) . decode ( encoding = self . encoding , <EOL> errors = self . errors ) <EOL> def __del__ ( self ) : <EOL> if self . _temp_stdin is not None : <EOL> self . _temp_stdin . close ( ) <EOL> self . _temp_stdin = None <EOL> def spawn ( self , argv = None , env = None , stdin = None , delay = None ) : <EOL> """<STR_LIT>""" <EOL> assert self . master_fd is None <EOL> self . _in_alt_mode = False <EOL> if not argv : <EOL> argv = [ os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> argv = self . _put_stdin_in_argv ( argv , stdin ) <EOL> pid , master_fd = pty . fork ( ) <EOL> self . pid = pid <EOL> self . master_fd = master_fd <EOL> if pid == pty . CHILD : <EOL> if self . _temp_stdin is not None : <EOL> self . _delay_for_pipe ( env = env , delay = delay ) <EOL> try : <EOL> if env is None : <EOL> os . execvp ( argv [ <NUM_LIT:0> ] , argv ) <EOL> else : <EOL> os . execvpe ( argv [ <NUM_LIT:0> ] , argv , env ) <EOL> except OSError as e : <EOL> os . _exit ( _find_error_code ( e ) ) <EOL> else : <EOL> self . _pipe_stdin ( stdin ) <EOL> on_main_thread = _on_main_thread ( ) <EOL> if on_main_thread : <EOL> old_handler = signal . signal ( signal . SIGWINCH , self . _signal_winch ) <EOL> try : <EOL> mode = tty . tcgetattr ( pty . STDIN_FILENO ) <EOL> tty . setraw ( pty . STDIN_FILENO ) <EOL> restore = True <EOL> except tty . error : <EOL> restore = False <EOL> self . _init_fd ( ) <EOL> try : <EOL> self . _copy ( ) <EOL> except ( IOError , OSError ) : <EOL> if restore : <EOL> tty . tcsetattr ( pty . STDIN_FILENO , tty . TCSAFLUSH , mode ) <EOL> _ , self . wcode = os . waitpid ( pid , <NUM_LIT:0> ) <EOL> os . close ( master_fd ) <EOL> self . master_fd = None <EOL> self . _in_alt_mode = False <EOL> if on_main_thread : <EOL> signal . signal ( signal . SIGWINCH , old_handler ) <EOL> return self . wcode <EOL> def _init_fd ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _set_pty_size ( ) <EOL> def _signal_winch ( self , signum , frame ) : <EOL> """<STR_LIT>""" <EOL> self . _set_pty_size ( ) <EOL> def _set_pty_size ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . master_fd is not None <EOL> buf = array . array ( '<STR_LIT:h>' , [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> fcntl . ioctl ( pty . STDOUT_FILENO , termios . TIOCGWINSZ , buf , True ) <EOL> fcntl . ioctl ( self . master_fd , termios . TIOCSWINSZ , buf ) <EOL> def _copy ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . master_fd is not None <EOL> master_fd = self . master_fd <EOL> bufsize = self . bufsize <EOL> while True : <EOL> try : <EOL> rfds , wfds , xfds = select . select ( [ master_fd , pty . STDIN_FILENO ] , [ ] , [ ] ) <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:4> : <EOL> continue <EOL> if master_fd in rfds : <EOL> data = os . read ( master_fd , bufsize ) <EOL> self . write_stdout ( data ) <EOL> if pty . STDIN_FILENO in rfds : <EOL> data = os . read ( pty . STDIN_FILENO , bufsize ) <EOL> self . write_stdin ( data ) <EOL> def _sanatize_data ( self , data ) : <EOL> i , flag = _findfirst ( data , ALTERNATE_MODE_FLAGS ) <EOL> if flag is None and self . _in_alt_mode : <EOL> return b'<STR_LIT>' <EOL> elif flag is not None : <EOL> if flag in START_ALTERNATE_MODE : <EOL> d0 = data [ : i ] <EOL> self . _in_alt_mode = True <EOL> d1 = self . _sanatize_data ( data [ i + len ( flag ) : ] ) <EOL> data = d0 + d1 <EOL> elif flag in END_ALTERNATE_MODE : <EOL> self . _in_alt_mode = False <EOL> data = self . _sanatize_data ( data [ i + len ( flag ) : ] ) <EOL> data = RE_HIDDEN . sub ( b'<STR_LIT>' , data ) <EOL> if self . remove_color : <EOL> data = RE_COLOR . sub ( b'<STR_LIT>' , data ) <EOL> return data <EOL> def write_stdout ( self , data ) : <EOL> """<STR_LIT>""" <EOL> os . write ( pty . STDOUT_FILENO , data ) <EOL> data = self . _sanatize_data ( data ) <EOL> if len ( data ) > <NUM_LIT:0> : <EOL> self . buffer . write ( data ) <EOL> def write_stdin ( self , data ) : <EOL> """<STR_LIT>""" <EOL> master_fd = self . master_fd <EOL> assert master_fd is not None <EOL> while len ( data ) > <NUM_LIT:0> : <EOL> n = os . write ( master_fd , data ) <EOL> data = data [ n : ] <EOL> def _stdin_filename ( self , stdin ) : <EOL> if stdin is None : <EOL> rtn = None <EOL> elif isinstance ( stdin , io . FileIO ) and os . path . isfile ( stdin . name ) : <EOL> rtn = stdin . name <EOL> elif isinstance ( stdin , ( io . BufferedIOBase , str , bytes ) ) : <EOL> self . _temp_stdin = tsi = tempfile . NamedTemporaryFile ( ) <EOL> rtn = tsi . name <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' . format ( stdin ) ) <EOL> return rtn <EOL> def _put_stdin_in_argv ( self , argv , stdin ) : <EOL> stdin_filename = self . _stdin_filename ( stdin ) <EOL> if stdin_filename is None : <EOL> return argv <EOL> argv = list ( argv ) <EOL> if argv . count ( '<STR_LIT:->' ) == <NUM_LIT:0> : <EOL> argv . append ( stdin_filename ) <EOL> else : <EOL> argv [ argv . index ( '<STR_LIT:->' ) ] = stdin_filename <EOL> return argv <EOL> def _pipe_stdin ( self , stdin ) : <EOL> if stdin is None or isinstance ( stdin , io . FileIO ) : <EOL> return None <EOL> tsi = self . _temp_stdin <EOL> bufsize = self . bufsize <EOL> if isinstance ( stdin , io . BufferedIOBase ) : <EOL> buf = stdin . read ( bufsize ) <EOL> while len ( buf ) != <NUM_LIT:0> : <EOL> tsi . write ( buf ) <EOL> tsi . flush ( ) <EOL> buf = stdin . read ( bufsize ) <EOL> elif isinstance ( stdin , ( str , bytes ) ) : <EOL> raw = stdin . encode ( ) if isinstance ( stdin , str ) else stdin <EOL> for i in range ( ( len ( raw ) // bufsize ) + <NUM_LIT:1> ) : <EOL> tsi . write ( raw [ i * bufsize : ( i + <NUM_LIT:1> ) * bufsize ] ) <EOL> tsi . flush ( ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' . format ( stdin ) ) <EOL> def _delay_for_pipe ( self , env = None , delay = None ) : <EOL> if delay is None : <EOL> delay = ( env or os . environ ) . get ( '<STR_LIT>' , - <NUM_LIT:1.0> ) <EOL> delay = float ( delay ) <EOL> if <NUM_LIT:0.0> < delay : <EOL> time . sleep ( delay ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> tpty = TeePTY ( ) <EOL> tpty . spawn ( sys . argv [ <NUM_LIT:1> : ] ) <EOL> print ( '<STR_LIT>' * <NUM_LIT:10> ) <EOL> print ( tpty . buffer . getvalue ( ) ) <EOL> print ( '<STR_LIT>' * <NUM_LIT:10> ) <EOL> print ( tpty ) <EOL> print ( '<STR_LIT>' * <NUM_LIT:10> ) <EOL> print ( '<STR_LIT>' . format ( tpty . wcode ) ) </s>
