<s> """<STR_LIT>""" <EOL> import code , sys , StringIO , tokenize <EOL> from twisted . conch import recvline <EOL> from twisted . internet import defer <EOL> from twisted . python . htmlizer import TokenPrinter <EOL> class FileWrapper : <EOL> """<STR_LIT>""" <EOL> softspace = <NUM_LIT:0> <EOL> state = '<STR_LIT>' <EOL> def __init__ ( self , o ) : <EOL> self . o = o <EOL> def flush ( self ) : <EOL> pass <EOL> def write ( self , data ) : <EOL> self . o . addOutput ( data . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) ) <EOL> def writelines ( self , lines ) : <EOL> self . write ( '<STR_LIT>' . join ( lines ) ) <EOL> class ManholeInterpreter ( code . InteractiveInterpreter ) : <EOL> """<STR_LIT>""" <EOL> numDeferreds = <NUM_LIT:0> <EOL> def __init__ ( self , handler , locals = None , filename = "<STR_LIT>" ) : <EOL> code . InteractiveInterpreter . __init__ ( self , locals ) <EOL> self . _pendingDeferreds = { } <EOL> self . handler = handler <EOL> self . filename = filename <EOL> self . resetBuffer ( ) <EOL> def resetBuffer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . buffer = [ ] <EOL> def push ( self , line ) : <EOL> """<STR_LIT>""" <EOL> self . buffer . append ( line ) <EOL> source = "<STR_LIT:\n>" . join ( self . buffer ) <EOL> more = self . runsource ( source , self . filename ) <EOL> if not more : <EOL> self . resetBuffer ( ) <EOL> return more <EOL> def runcode ( self , * a , ** kw ) : <EOL> orighook , sys . displayhook = sys . displayhook , self . displayhook <EOL> try : <EOL> origout , sys . stdout = sys . stdout , FileWrapper ( self . handler ) <EOL> try : <EOL> code . InteractiveInterpreter . runcode ( self , * a , ** kw ) <EOL> finally : <EOL> sys . stdout = origout <EOL> finally : <EOL> sys . displayhook = orighook <EOL> def displayhook ( self , obj ) : <EOL> self . locals [ '<STR_LIT:_>' ] = obj <EOL> if isinstance ( obj , defer . Deferred ) : <EOL> if hasattr ( obj , "<STR_LIT:result>" ) : <EOL> self . write ( repr ( obj ) ) <EOL> elif id ( obj ) in self . _pendingDeferreds : <EOL> self . write ( "<STR_LIT>" % ( self . _pendingDeferreds [ id ( obj ) ] [ <NUM_LIT:0> ] , ) ) <EOL> else : <EOL> d = self . _pendingDeferreds <EOL> k = self . numDeferreds <EOL> d [ id ( obj ) ] = ( k , obj ) <EOL> self . numDeferreds += <NUM_LIT:1> <EOL> obj . addCallbacks ( self . _cbDisplayDeferred , self . _ebDisplayDeferred , <EOL> callbackArgs = ( k , obj ) , errbackArgs = ( k , obj ) ) <EOL> self . write ( "<STR_LIT>" % ( k , ) ) <EOL> elif obj is not None : <EOL> self . write ( repr ( obj ) ) <EOL> def _cbDisplayDeferred ( self , result , k , obj ) : <EOL> self . write ( "<STR_LIT>" % ( k , result ) , True ) <EOL> del self . _pendingDeferreds [ id ( obj ) ] <EOL> return result <EOL> def _ebDisplayDeferred ( self , failure , k , obj ) : <EOL> self . write ( "<STR_LIT>" % ( k , failure . getErrorMessage ( ) ) , True ) <EOL> del self . _pendingDeferreds [ id ( obj ) ] <EOL> return failure <EOL> def write ( self , data , async = False ) : <EOL> self . handler . addOutput ( data , async ) <EOL> CTRL_C = '<STR_LIT>' <EOL> CTRL_D = '<STR_LIT>' <EOL> CTRL_BACKSLASH = '<STR_LIT>' <EOL> CTRL_L = '<STR_LIT>' <EOL> CTRL_A = '<STR_LIT>' <EOL> CTRL_E = '<STR_LIT>' <EOL> class Manhole ( recvline . HistoricRecvLine ) : <EOL> """<STR_LIT>""" <EOL> namespace = None <EOL> def __init__ ( self , namespace = None ) : <EOL> recvline . HistoricRecvLine . __init__ ( self ) <EOL> if namespace is not None : <EOL> self . namespace = namespace . copy ( ) <EOL> def connectionMade ( self ) : <EOL> recvline . HistoricRecvLine . connectionMade ( self ) <EOL> self . interpreter = ManholeInterpreter ( self , self . namespace ) <EOL> self . keyHandlers [ CTRL_C ] = self . handle_INT <EOL> self . keyHandlers [ CTRL_D ] = self . handle_EOF <EOL> self . keyHandlers [ CTRL_L ] = self . handle_FF <EOL> self . keyHandlers [ CTRL_A ] = self . handle_HOME <EOL> self . keyHandlers [ CTRL_E ] = self . handle_END <EOL> self . keyHandlers [ CTRL_BACKSLASH ] = self . handle_QUIT <EOL> def handle_INT ( self ) : <EOL> """<STR_LIT>""" <EOL> self . pn = <NUM_LIT:0> <EOL> self . lineBuffer = [ ] <EOL> self . lineBufferIndex = <NUM_LIT:0> <EOL> self . interpreter . resetBuffer ( ) <EOL> self . terminal . nextLine ( ) <EOL> self . terminal . write ( "<STR_LIT>" ) <EOL> self . terminal . nextLine ( ) <EOL> self . terminal . write ( self . ps [ self . pn ] ) <EOL> def handle_EOF ( self ) : <EOL> if self . lineBuffer : <EOL> self . terminal . write ( '<STR_LIT>' ) <EOL> else : <EOL> self . handle_QUIT ( ) <EOL> def handle_FF ( self ) : <EOL> """<STR_LIT>""" <EOL> self . terminal . eraseDisplay ( ) <EOL> self . terminal . cursorHome ( ) <EOL> self . drawInputLine ( ) <EOL> def handle_QUIT ( self ) : <EOL> self . terminal . loseConnection ( ) <EOL> def _needsNewline ( self ) : <EOL> w = self . terminal . lastWrite <EOL> return not w . endswith ( '<STR_LIT:\n>' ) and not w . endswith ( '<STR_LIT>' ) <EOL> def addOutput ( self , bytes , async = False ) : <EOL> if async : <EOL> self . terminal . eraseLine ( ) <EOL> self . terminal . cursorBackward ( len ( self . lineBuffer ) + len ( self . ps [ self . pn ] ) ) <EOL> self . terminal . write ( bytes ) <EOL> if async : <EOL> if self . _needsNewline ( ) : <EOL> self . terminal . nextLine ( ) <EOL> self . terminal . write ( self . ps [ self . pn ] ) <EOL> if self . lineBuffer : <EOL> oldBuffer = self . lineBuffer <EOL> self . lineBuffer = [ ] <EOL> self . lineBufferIndex = <NUM_LIT:0> <EOL> self . _deliverBuffer ( oldBuffer ) <EOL> def lineReceived ( self , line ) : <EOL> more = self . interpreter . push ( line ) <EOL> self . pn = bool ( more ) <EOL> if self . _needsNewline ( ) : <EOL> self . terminal . nextLine ( ) <EOL> self . terminal . write ( self . ps [ self . pn ] ) <EOL> class VT102Writer : <EOL> """<STR_LIT>""" <EOL> typeToColor = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:string>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> normalColor = '<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> self . written = [ ] <EOL> def color ( self , type ) : <EOL> r = self . typeToColor . get ( type , '<STR_LIT>' ) <EOL> return r <EOL> def write ( self , token , type = None ) : <EOL> if token and token != '<STR_LIT:\r>' : <EOL> c = self . color ( type ) <EOL> if c : <EOL> self . written . append ( c ) <EOL> self . written . append ( token ) <EOL> if c : <EOL> self . written . append ( self . normalColor ) <EOL> def __str__ ( self ) : <EOL> s = '<STR_LIT>' . join ( self . written ) <EOL> return s . strip ( '<STR_LIT:\n>' ) . splitlines ( ) [ - <NUM_LIT:1> ] <EOL> def lastColorizedLine ( source ) : <EOL> """<STR_LIT>""" <EOL> w = VT102Writer ( ) <EOL> p = TokenPrinter ( w . write ) . printtoken <EOL> s = StringIO . StringIO ( source ) <EOL> tokenize . tokenize ( s . readline , p ) <EOL> return str ( w ) <EOL> class ColoredManhole ( Manhole ) : <EOL> """<STR_LIT>""" <EOL> def getSource ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( '<STR_LIT:\n>' . join ( self . interpreter . buffer ) + <EOL> '<STR_LIT:\n>' + <EOL> '<STR_LIT>' . join ( self . lineBuffer ) ) <EOL> def characterReceived ( self , ch , moreCharactersComing ) : <EOL> if self . mode == '<STR_LIT>' : <EOL> self . lineBuffer . insert ( self . lineBufferIndex , ch ) <EOL> else : <EOL> self . lineBuffer [ self . lineBufferIndex : self . lineBufferIndex + <NUM_LIT:1> ] = [ ch ] <EOL> self . lineBufferIndex += <NUM_LIT:1> <EOL> if moreCharactersComing : <EOL> return <EOL> if ch == '<STR_LIT:U+0020>' : <EOL> self . terminal . write ( ch ) <EOL> return <EOL> source = self . getSource ( ) <EOL> try : <EOL> coloredLine = lastColorizedLine ( source ) <EOL> except tokenize . TokenError : <EOL> self . terminal . write ( ch ) <EOL> else : <EOL> self . terminal . eraseLine ( ) <EOL> self . terminal . cursorBackward ( len ( self . lineBuffer ) + len ( self . ps [ self . pn ] ) - <NUM_LIT:1> ) <EOL> self . terminal . write ( self . ps [ self . pn ] + coloredLine ) <EOL> n = len ( self . lineBuffer ) - self . lineBufferIndex <EOL> if n : <EOL> self . terminal . cursorBackward ( n ) </s>
