<s> import logging <EOL> import termios <EOL> import codecs <EOL> import struct <EOL> import fcntl <EOL> import sys <EOL> import os <EOL> import io <EOL> __all__ = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class ConsoleShell ( ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client , log = logging , stream_out = None ) : <EOL> self . client = client <EOL> self . log = log <EOL> self . stream_out = stream_out or io . StringIO ( ) <EOL> self . encoding_errors = '<STR_LIT:replace>' <EOL> self . decoder = None <EOL> def write ( self , string = u'<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . stream_out . write ( string ) <EOL> @ property <EOL> def terminal_type ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' <EOL> @ property <EOL> def xdisploc ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' <EOL> @ property <EOL> def terminal_speed ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT> , <NUM_LIT> <EOL> @ property <EOL> def terminal_width ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT> <EOL> @ property <EOL> def terminal_height ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT> <EOL> def decode ( self , input , final = False ) : <EOL> """<STR_LIT>""" <EOL> enc = self . client . encoding ( incoming = True ) <EOL> if ( self . decoder is None or enc != self . decoder . _encoding ) : <EOL> self . decoder = codecs . getincrementaldecoder ( enc ) ( <EOL> errors = self . encoding_errors ) <EOL> self . decoder . _encoding = enc <EOL> return self . decoder . decode ( input , final ) <EOL> def feed_byte ( self , byte ) : <EOL> """<STR_LIT>""" <EOL> ucs = self . decode ( byte ) <EOL> if ucs : <EOL> self . write ( ucs ) <EOL> def encode ( self , buf , errors = None ) : <EOL> """<STR_LIT>""" <EOL> errors = errors if errors is not None else self . encoding_errors <EOL> return bytes ( buf , self . client . encoding ( outgoing = True ) , errors ) <EOL> @ property <EOL> def will_echo ( self ) : <EOL> """<STR_LIT>""" <EOL> from . telopt import ECHO <EOL> return self . client . stream . remote_option . enabled ( ECHO ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> enc_in = self . client . encoding ( incoming = True ) <EOL> enc_out = self . client . encoding ( outgoing = True ) <EOL> return ( enc_in if enc_in == enc_out else <EOL> '<STR_LIT>' . format ( enc_in , enc_out ) ) <EOL> class TerminalShell ( ConsoleShell ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client , log = logging , stream_out = None ) : <EOL> self . stream_out = stream_out or sys . __stdout__ <EOL> super ( ) . __init__ ( client , log , self . stream_out ) <EOL> def write ( self , string = u'<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if string : <EOL> self . stream_out . write ( string ) <EOL> except BlockingIOError : <EOL> self . client . _loop . call_later ( <NUM_LIT> , self . write , string ) <EOL> try : <EOL> self . stream_out . flush ( ) <EOL> except BlockingIOError : <EOL> self . client . _loop . call_later ( <NUM_LIT> , self . write ) <EOL> @ property <EOL> def terminal_type ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) or super ( ) . terminal_type ) <EOL> @ property <EOL> def xdisploc ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) or super ( ) . xdisploc ) <EOL> @ property <EOL> def terminal_speed ( self ) : <EOL> """<STR_LIT>""" <EOL> if ( hasattr ( self . stream_out , '<STR_LIT>' ) <EOL> and os . isatty ( self . stream_out . fileno ( ) ) ) : <EOL> return self . _query_term_speed ( self . stream_out ) <EOL> return super ( ) . terminal_speed <EOL> @ property <EOL> def terminal_width ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if ( hasattr ( self . stream_out , '<STR_LIT>' ) <EOL> and os . isatty ( self . stream_out . fileno ( ) ) ) : <EOL> rows , cols , xpixels , ypixels = ( <EOL> self . _query_term_winsize ( self . stream_out ) ) <EOL> return cols <EOL> except io . UnsupportedOperation : <EOL> self . log . warn ( '<STR_LIT>' . format ( <EOL> self . __class__ . __name__ , self . stream_out ) ) <EOL> pass <EOL> try : <EOL> cols = int ( os . environ . get ( '<STR_LIT>' , str ( super ( ) . terminal_width ) ) ) <EOL> except ValueError : <EOL> cols = super ( ) . terminal_width <EOL> return cols <EOL> @ property <EOL> def terminal_height ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if ( hasattr ( self . stream_out , '<STR_LIT>' ) <EOL> and os . isatty ( self . stream_out . fileno ( ) ) ) : <EOL> rows , cols , xpixels , ypixels = ( <EOL> self . _query_term_winsize ( self . stream_out ) ) <EOL> return rows <EOL> except io . UnsupportedOperation : <EOL> self . log . warn ( '<STR_LIT>' . format ( <EOL> self . __class__ . __name__ , self . stream_out ) ) <EOL> pass <EOL> try : <EOL> rows = int ( os . environ . get ( '<STR_LIT>' , str ( super ( ) . terminal_height ) ) ) <EOL> except ValueError : <EOL> rows = super ( ) . terminal_height <EOL> return rows <EOL> @ staticmethod <EOL> def _query_term_speed ( tty_fd ) : <EOL> """<STR_LIT>""" <EOL> iflag , oflag , cflag , lflag , ispeed , ospeed , cc = ( <EOL> termios . tcgetattr ( tty_fd ) ) <EOL> return ispeed , ospeed <EOL> @ staticmethod <EOL> def _query_term_winsize ( tty_fd ) : <EOL> """<STR_LIT>""" <EOL> val = fcntl . ioctl ( tty_fd , termios . TIOCGWINSZ , b'<STR_LIT:\x00>' * <NUM_LIT:8> ) <EOL> return struct . unpack ( '<STR_LIT>' , val ) </s>
