<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , unicode_literals <EOL> import os <EOL> import codecs <EOL> import sys <EOL> from . import control as ctrl , escape as esc <EOL> if sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:2> : <EOL> str = unicode <EOL> class Stream ( object ) : <EOL> """<STR_LIT>""" <EOL> basic = { <EOL> ctrl . BEL : "<STR_LIT>" , <EOL> ctrl . BS : "<STR_LIT>" , <EOL> ctrl . HT : "<STR_LIT>" , <EOL> ctrl . LF : "<STR_LIT>" , <EOL> ctrl . VT : "<STR_LIT>" , <EOL> ctrl . FF : "<STR_LIT>" , <EOL> ctrl . CR : "<STR_LIT>" , <EOL> ctrl . SO : "<STR_LIT>" , <EOL> ctrl . SI : "<STR_LIT>" , <EOL> } <EOL> escape = { <EOL> esc . RIS : "<STR_LIT>" , <EOL> esc . IND : "<STR_LIT:index>" , <EOL> esc . NEL : "<STR_LIT>" , <EOL> esc . RI : "<STR_LIT>" , <EOL> esc . HTS : "<STR_LIT>" , <EOL> esc . DECSC : "<STR_LIT>" , <EOL> esc . DECRC : "<STR_LIT>" , <EOL> } <EOL> sharp = { <EOL> esc . DECALN : "<STR_LIT>" , <EOL> } <EOL> percent = { <EOL> esc . DEFAULT : "<STR_LIT>" , <EOL> esc . UTF8 : "<STR_LIT>" , <EOL> esc . UTF8_OBSOLETE : "<STR_LIT>" , <EOL> } <EOL> csi = { <EOL> esc . ICH : "<STR_LIT>" , <EOL> esc . CUU : "<STR_LIT>" , <EOL> esc . CUD : "<STR_LIT>" , <EOL> esc . CUF : "<STR_LIT>" , <EOL> esc . CUB : "<STR_LIT>" , <EOL> esc . CNL : "<STR_LIT>" , <EOL> esc . CPL : "<STR_LIT>" , <EOL> esc . CHA : "<STR_LIT>" , <EOL> esc . CUP : "<STR_LIT>" , <EOL> esc . ED : "<STR_LIT>" , <EOL> esc . EL : "<STR_LIT>" , <EOL> esc . IL : "<STR_LIT>" , <EOL> esc . DL : "<STR_LIT>" , <EOL> esc . DCH : "<STR_LIT>" , <EOL> esc . ECH : "<STR_LIT>" , <EOL> esc . HPR : "<STR_LIT>" , <EOL> esc . VPA : "<STR_LIT>" , <EOL> esc . VPR : "<STR_LIT>" , <EOL> esc . HVP : "<STR_LIT>" , <EOL> esc . TBC : "<STR_LIT>" , <EOL> esc . SM : "<STR_LIT>" , <EOL> esc . RM : "<STR_LIT>" , <EOL> esc . SGR : "<STR_LIT>" , <EOL> esc . DECSTBM : "<STR_LIT>" , <EOL> esc . HPA : "<STR_LIT>" <EOL> } <EOL> def __init__ ( self ) : <EOL> self . handlers = { <EOL> "<STR_LIT>" : self . _stream , <EOL> "<STR_LIT>" : self . _escape , <EOL> "<STR_LIT>" : self . _arguments , <EOL> "<STR_LIT>" : self . _sharp , <EOL> "<STR_LIT>" : self . _percent , <EOL> "<STR_LIT>" : self . _charset <EOL> } <EOL> self . listeners = [ ] <EOL> self . reset ( ) <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> self . state = "<STR_LIT>" <EOL> self . flags = { } <EOL> self . params = [ ] <EOL> self . current = "<STR_LIT>" <EOL> def consume ( self , char ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( char , str ) : <EOL> raise TypeError ( "<STR_LIT>" % self . __class__ . __name__ ) <EOL> try : <EOL> self . handlers . get ( self . state ) ( char ) <EOL> except TypeError : <EOL> pass <EOL> except KeyError : <EOL> if __debug__ : <EOL> self . flags [ "<STR_LIT:state>" ] = self . state <EOL> self . flags [ "<STR_LIT>" ] = char <EOL> self . dispatch ( "<STR_LIT>" , * self . params ) <EOL> self . reset ( ) <EOL> else : <EOL> raise <EOL> def feed ( self , chars ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( chars , str ) : <EOL> raise TypeError ( "<STR_LIT>" % self . __class__ . __name__ ) <EOL> for char in chars : <EOL> self . consume ( char ) <EOL> def attach ( self , screen , only = ( ) ) : <EOL> """<STR_LIT>""" <EOL> self . listeners . append ( ( screen , set ( only ) ) ) <EOL> def detach ( self , screen ) : <EOL> """<STR_LIT>""" <EOL> for idx , ( listener , _ ) in enumerate ( self . listeners ) : <EOL> if screen is listener : <EOL> self . listeners . pop ( idx ) <EOL> def dispatch ( self , event , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> for listener , only in self . listeners : <EOL> if only and event not in only : <EOL> continue <EOL> try : <EOL> handler = getattr ( listener , event ) <EOL> except AttributeError : <EOL> continue <EOL> if hasattr ( listener , "<STR_LIT>" ) : <EOL> listener . __before__ ( event ) <EOL> handler ( * args , ** self . flags ) <EOL> if hasattr ( listener , "<STR_LIT>" ) : <EOL> listener . __after__ ( event ) <EOL> else : <EOL> if kwargs . get ( "<STR_LIT>" , True ) : <EOL> self . reset ( ) <EOL> def _stream ( self , char ) : <EOL> """<STR_LIT>""" <EOL> if char in self . basic : <EOL> self . dispatch ( self . basic [ char ] ) <EOL> elif char == ctrl . ESC : <EOL> self . state = "<STR_LIT>" <EOL> elif char == ctrl . CSI : <EOL> self . state = "<STR_LIT>" <EOL> elif char not in [ ctrl . NUL , ctrl . DEL ] : <EOL> self . dispatch ( "<STR_LIT>" , char ) <EOL> def _escape ( self , char ) : <EOL> """<STR_LIT>""" <EOL> if char == "<STR_LIT:#>" : <EOL> self . state = "<STR_LIT>" <EOL> elif char == "<STR_LIT:%>" : <EOL> self . state = "<STR_LIT>" <EOL> elif char == "<STR_LIT:[>" : <EOL> self . state = "<STR_LIT>" <EOL> elif char in "<STR_LIT>" : <EOL> self . state = "<STR_LIT>" <EOL> self . flags [ "<STR_LIT>" ] = char <EOL> else : <EOL> self . dispatch ( self . escape [ char ] ) <EOL> def _sharp ( self , char ) : <EOL> """<STR_LIT>""" <EOL> self . dispatch ( self . sharp [ char ] ) <EOL> def _percent ( self , char ) : <EOL> """<STR_LIT>""" <EOL> self . dispatch ( self . percent [ char ] ) <EOL> def _charset ( self , char ) : <EOL> """<STR_LIT>""" <EOL> self . dispatch ( "<STR_LIT>" , char ) <EOL> def _arguments ( self , char ) : <EOL> """<STR_LIT>""" <EOL> if char == "<STR_LIT:?>" : <EOL> self . flags [ "<STR_LIT>" ] = True <EOL> elif char in [ ctrl . BEL , ctrl . BS , ctrl . HT , ctrl . LF , ctrl . VT , <EOL> ctrl . FF , ctrl . CR ] : <EOL> self . dispatch ( self . basic [ char ] , reset = False ) <EOL> elif char == ctrl . SP : <EOL> pass <EOL> elif char in [ ctrl . CAN , ctrl . SUB ] : <EOL> self . dispatch ( "<STR_LIT>" , char ) <EOL> self . state = "<STR_LIT>" <EOL> elif char . isdigit ( ) : <EOL> self . current += char <EOL> else : <EOL> self . params . append ( min ( int ( self . current or <NUM_LIT:0> ) , <NUM_LIT> ) ) <EOL> if char == "<STR_LIT:;>" : <EOL> self . current = "<STR_LIT>" <EOL> else : <EOL> self . dispatch ( self . csi [ char ] , * self . params ) <EOL> class ByteStream ( Stream ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , encodings = None ) : <EOL> encodings = encodings or [ <EOL> ( "<STR_LIT:utf-8>" , "<STR_LIT:strict>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:strict>" ) , <EOL> ( "<STR_LIT:utf-8>" , "<STR_LIT:replace>" ) <EOL> ] <EOL> self . buffer = b"<STR_LIT>" , <NUM_LIT:0> <EOL> self . decoders = [ codecs . getincrementaldecoder ( encoding ) ( errors ) <EOL> for encoding , errors in encodings ] <EOL> super ( ByteStream , self ) . __init__ ( ) <EOL> def feed ( self , chars ) : <EOL> if not isinstance ( chars , bytes ) : <EOL> raise TypeError ( <EOL> "<STR_LIT>" % self . __class__ . __name__ ) <EOL> for decoder in self . decoders : <EOL> decoder . setstate ( self . buffer ) <EOL> try : <EOL> chars = decoder . decode ( chars ) <EOL> except UnicodeDecodeError : <EOL> continue <EOL> self . buffer = decoder . getstate ( ) <EOL> return super ( ByteStream , self ) . feed ( chars ) <EOL> else : <EOL> raise <EOL> class DebugStream ( ByteStream ) : <EOL> r"""<STR_LIT>""" <EOL> def __init__ ( self , to = sys . stdout , only = ( ) , * args , ** kwargs ) : <EOL> super ( DebugStream , self ) . __init__ ( * args , ** kwargs ) <EOL> def safe_str ( chunk ) : <EOL> if isinstance ( chunk , bytes ) : <EOL> chunk = chunk . decode ( "<STR_LIT:utf-8>" ) <EOL> elif not isinstance ( chunk , str ) : <EOL> chunk = str ( chunk ) <EOL> return chunk <EOL> class Bugger ( object ) : <EOL> __before__ = __after__ = lambda * args : None <EOL> def __getattr__ ( self , event ) : <EOL> def inner ( * args , ** flags ) : <EOL> to . write ( event . upper ( ) + "<STR_LIT:U+0020>" ) <EOL> to . write ( "<STR_LIT>" . join ( map ( safe_str , args ) ) ) <EOL> to . write ( "<STR_LIT:U+0020>" ) <EOL> to . write ( "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT>" . format ( name , safe_str ( arg ) ) <EOL> for name , arg in flags . items ( ) ) ) <EOL> to . write ( os . linesep ) <EOL> return inner <EOL> self . attach ( Bugger ( ) , only = only ) </s>
