Source code for bravo.stdio

# vim: set fileencoding=utf8 :

import os
import sys

from twisted.conch.insults.insults import ServerProtocol
from twisted.conch.manhole import Manhole
from twisted.internet import reactor
from twisted.internet.defer import Deferred
from twisted.internet.protocol import ClientCreator
from twisted.internet.stdio import StandardIO
from twisted.internet.task import LoopingCall
from twisted.protocols.amp import AMP
from twisted.protocols.basic import LineReceiver

from bravo.amp import Version, Worlds, RunCommand
from import fancy_console_name

    import termios
    import tty
    fancy_console = os.isatty(sys.__stdin__.fileno())
except ImportError:
    fancy_console = False

typeToColor = {
    'identifier': '\x1b[31m',
    'keyword': '\x1b[32m',
    'parameter': '\x1b[33m',
    'variable': '\x1b[1;33m',
    'string': '\x1b[35m',
    'number': '\x1b[36m',
    'op': '\x1b[37m'

normalColor = '\x1b[0m'

[docs]class AMPGateway(object): """ Wrapper around the logical implementation of a console. """ def __init__(self, host, port=25600): self.ready = False = host self.port = port = None
[docs] def connect(self): """ Connect this gateway to a remote Bravo server. Returns a Deferred that will fire when connected, or fail if the connection cannot be established. """ = ClientCreator(reactor, AMP) d =, self.port) d.addCallback(self.connected) return d
def connected(self, p): self.remote = p self.sendLine("Successfully connected to server, getting version...") d = self.remote.callRemote(Version) d.addCallback(self.version) LoopingCall(self.world_loop).start(10) def world_loop(self): self.remote.callRemote(Worlds).addCallback( lambda d: setattr(self, "worlds", d["worlds"]) ) def version(self, d): self.version = d["version"] self.sendLine("Connected to Bravo %s. Ready." % self.version) self.ready = True
[docs] def call(self, command, params): """ Run a command. This is the client-side implementation; it wraps a few things to protect the console from raw logic and the server from builtin commands. """ self.ready_deferred = Deferred() if self.ready: if command in ("exit", "quit"): # Quit. stop_console() reactor.stop() elif command == "worlds": # Print list of available worlds. self.sendLine("Worlds:") for world in self.worlds: self.sendLine(world) elif command == "select": # World selection. world = params[0] if world in self.worlds: = world self.sendLine("Selected world %s" % world) else: self.sendLine("Couldn't find world %s" % world) else: # Remote command. Do we have a world? if try: d = self.remote.callRemote(RunCommand,, command=command, parameters=params) d.addCallback(self.results) self.ready = False except: self.sendLine("Huh?") else: self.sendLine("No world selected.") if self.ready: self.ready_deferred.callback(None) return self.ready_deferred
def results(self, d): for line in d["output"]: self.sendLine(line) self.ready = True reactor.callLater(0, self.ready_deferred.callback, None) def sendLine(self, line): if isinstance(line, unicode): line = line.encode("utf8") self.print_hook(line)
class BravoInterpreter(object): def __init__(self, handler, ag): self.handler = handler = ag = self.print_hook def resetBuffer(self): pass def print_hook(self, line): # XXX #for user in self.factory.protocols: # printable = printable.replace(user, fancy_console_name(user)) self.handler.addOutput("%s\n" % line) def push(self, line): """ Handle a command. """ line = line.strip() if line: params = line.split() command = params.pop(0).lower(), params) def lastColorizedLine(self, line): s = [] for token in line.split(): try: int(token) s.append(typeToColor["number"] + token) except ValueError: if token in self.commands: s.append(typeToColor["keyword"] + token) elif token in self.factory.protocols: s.append(fancy_console_name(token)) else: s.append(normalColor + token) return normalColor + " ".join(s)
[docs]class BravoManhole(Manhole): """ A console for TTYs. """ ps = ("\x1b[1;37mBravo \x1b[0;37m>\x1b[0;0m ", "... ") def __init__(self, factory, *args, **kwargs): Manhole.__init__(self, *args, **kwargs) self.f = factory def connectionMade(self): Manhole.connectionMade(self) self.interpreter = BravoInterpreter(self, self.f) # Borrowed from ColoredManhole, this colorizes input. def characterReceived(self, ch, moreCharactersComing): if self.mode == 'insert': self.lineBuffer.insert(self.lineBufferIndex, ch) else: self.lineBuffer[self.lineBufferIndex:self.lineBufferIndex+1] = [ch] self.lineBufferIndex += 1 if moreCharactersComing: # Skip it all, we'll get called with another character in like 2 # femtoseconds. return if ch == ' ': # Don't bother to try to color whitespace self.terminal.write(ch) return source = ''.join(self.lineBuffer) # Try to write some junk try: coloredLine = self.interpreter.lastColorizedLine(source) except: # We couldn't do it. Strange. Oh well, just add the character. self.terminal.write(ch) else: # Success! Clear the source on this line. self.terminal.eraseLine() self.terminal.cursorBackward(len(self.lineBuffer) + len([]) - 1) # And write a new, colorized one. self.terminal.write([] + coloredLine) # And move the cursor to where it belongs n = len(self.lineBuffer) - self.lineBufferIndex if n: self.terminal.cursorBackward(n)
greeting = """ Welcome to Bravo! This terminal has no fancy features. """ prompt = "Bravo > "
[docs]class BravoConsole(LineReceiver): """ A console for things not quite as awesome as TTYs. This console is extremely well-suited to Win32. """ delimiter = os.linesep def __init__(self, ag): = ag ag.print_hook = self.sendLine def connectionMade(self): self.transport.write(greeting) self.transport.write(prompt) def lineReceived(self, line): line = line.strip() if line: params = line.split() command = params.pop(0).lower() d =, params) d.addCallback(lambda chaff: self.transport.write(prompt)) else: self.transport.write(prompt) # Cribbed from Twisted. This version doesn't try to start the reactor, or a # handful of other things. At some point, this may no longer even look like # Twisted code.
oldSettings = None def start_console(): ag = AMPGateway("localhost", 25600) ag.connect() if fancy_console: global oldSettings fd = sys.__stdin__.fileno() oldSettings = termios.tcgetattr(fd) tty.setraw(fd) p = ServerProtocol(BravoManhole, ag) else: p = BravoConsole(ag) StandardIO(p) return p def stop_console(): if fancy_console: fd = sys.__stdin__.fileno() termios.tcsetattr(fd, termios.TCSANOW, oldSettings) # Took me forever to figure it out. This adorable little gem is # the control sequence RIS, which resets ANSI-compatible terminals # to their initial state. In the process, of course, they nuke all # of the stuff on the screen. os.write(fd, "\r\x1bc\r")