diff options
Diffstat (limited to 'lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/recvline.py')
-rwxr-xr-x | lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/recvline.py | 329 |
1 files changed, 0 insertions, 329 deletions
diff --git a/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/recvline.py b/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/recvline.py deleted file mode 100755 index 6c8416ae..00000000 --- a/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/recvline.py +++ /dev/null @@ -1,329 +0,0 @@ -# -*- test-case-name: twisted.conch.test.test_recvline -*- -# Copyright (c) Twisted Matrix Laboratories. -# See LICENSE for details. - -""" -Basic line editing support. - -@author: Jp Calderone -""" - -import string - -from zope.interface import implements - -from twisted.conch.insults import insults, helper - -from twisted.python import log, reflect - -_counters = {} -class Logging(object): - """Wrapper which logs attribute lookups. - - This was useful in debugging something, I guess. I forget what. - It can probably be deleted or moved somewhere more appropriate. - Nothing special going on here, really. - """ - def __init__(self, original): - self.original = original - key = reflect.qual(original.__class__) - count = _counters.get(key, 0) - _counters[key] = count + 1 - self._logFile = file(key + '-' + str(count), 'w') - - def __str__(self): - return str(super(Logging, self).__getattribute__('original')) - - def __repr__(self): - return repr(super(Logging, self).__getattribute__('original')) - - def __getattribute__(self, name): - original = super(Logging, self).__getattribute__('original') - logFile = super(Logging, self).__getattribute__('_logFile') - logFile.write(name + '\n') - return getattr(original, name) - -class TransportSequence(object): - """An L{ITerminalTransport} implementation which forwards calls to - one or more other L{ITerminalTransport}s. - - This is a cheap way for servers to keep track of the state they - expect the client to see, since all terminal manipulations can be - send to the real client and to a terminal emulator that lives in - the server process. - """ - implements(insults.ITerminalTransport) - - for keyID in ('UP_ARROW', 'DOWN_ARROW', 'RIGHT_ARROW', 'LEFT_ARROW', - 'HOME', 'INSERT', 'DELETE', 'END', 'PGUP', 'PGDN', - 'F1', 'F2', 'F3', 'F4', 'F5', 'F6', 'F7', 'F8', 'F9', - 'F10', 'F11', 'F12'): - exec '%s = object()' % (keyID,) - - TAB = '\t' - BACKSPACE = '\x7f' - - def __init__(self, *transports): - assert transports, "Cannot construct a TransportSequence with no transports" - self.transports = transports - - for method in insults.ITerminalTransport: - exec """\ -def %s(self, *a, **kw): - for tpt in self.transports: - result = tpt.%s(*a, **kw) - return result -""" % (method, method) - -class LocalTerminalBufferMixin(object): - """A mixin for RecvLine subclasses which records the state of the terminal. - - This is accomplished by performing all L{ITerminalTransport} operations on both - the transport passed to makeConnection and an instance of helper.TerminalBuffer. - - @ivar terminalCopy: A L{helper.TerminalBuffer} instance which efforts - will be made to keep up to date with the actual terminal - associated with this protocol instance. - """ - - def makeConnection(self, transport): - self.terminalCopy = helper.TerminalBuffer() - self.terminalCopy.connectionMade() - return super(LocalTerminalBufferMixin, self).makeConnection( - TransportSequence(transport, self.terminalCopy)) - - def __str__(self): - return str(self.terminalCopy) - -class RecvLine(insults.TerminalProtocol): - """L{TerminalProtocol} which adds line editing features. - - Clients will be prompted for lines of input with all the usual - features: character echoing, left and right arrow support for - moving the cursor to different areas of the line buffer, backspace - and delete for removing characters, and insert for toggling - between typeover and insert mode. Tabs will be expanded to enough - spaces to move the cursor to the next tabstop (every four - characters by default). Enter causes the line buffer to be - cleared and the line to be passed to the lineReceived() method - which, by default, does nothing. Subclasses are responsible for - redrawing the input prompt (this will probably change). - """ - width = 80 - height = 24 - - TABSTOP = 4 - - ps = ('>>> ', '... ') - pn = 0 - _printableChars = set(string.printable) - - def connectionMade(self): - # A list containing the characters making up the current line - self.lineBuffer = [] - - # A zero-based (wtf else?) index into self.lineBuffer. - # Indicates the current cursor position. - self.lineBufferIndex = 0 - - t = self.terminal - # A map of keyIDs to bound instance methods. - self.keyHandlers = { - t.LEFT_ARROW: self.handle_LEFT, - t.RIGHT_ARROW: self.handle_RIGHT, - t.TAB: self.handle_TAB, - - # Both of these should not be necessary, but figuring out - # which is necessary is a huge hassle. - '\r': self.handle_RETURN, - '\n': self.handle_RETURN, - - t.BACKSPACE: self.handle_BACKSPACE, - t.DELETE: self.handle_DELETE, - t.INSERT: self.handle_INSERT, - t.HOME: self.handle_HOME, - t.END: self.handle_END} - - self.initializeScreen() - - def initializeScreen(self): - # Hmm, state sucks. Oh well. - # For now we will just take over the whole terminal. - self.terminal.reset() - self.terminal.write(self.ps[self.pn]) - # XXX Note: I would prefer to default to starting in insert - # mode, however this does not seem to actually work! I do not - # know why. This is probably of interest to implementors - # subclassing RecvLine. - - # XXX XXX Note: But the unit tests all expect the initial mode - # to be insert right now. Fuck, there needs to be a way to - # query the current mode or something. - # self.setTypeoverMode() - self.setInsertMode() - - def currentLineBuffer(self): - s = ''.join(self.lineBuffer) - return s[:self.lineBufferIndex], s[self.lineBufferIndex:] - - def setInsertMode(self): - self.mode = 'insert' - self.terminal.setModes([insults.modes.IRM]) - - def setTypeoverMode(self): - self.mode = 'typeover' - self.terminal.resetModes([insults.modes.IRM]) - - def drawInputLine(self): - """ - Write a line containing the current input prompt and the current line - buffer at the current cursor position. - """ - self.terminal.write(self.ps[self.pn] + ''.join(self.lineBuffer)) - - def terminalSize(self, width, height): - # XXX - Clear the previous input line, redraw it at the new - # cursor position - self.terminal.eraseDisplay() - self.terminal.cursorHome() - self.width = width - self.height = height - self.drawInputLine() - - def unhandledControlSequence(self, seq): - pass - - def keystrokeReceived(self, keyID, modifier): - m = self.keyHandlers.get(keyID) - if m is not None: - m() - elif keyID in self._printableChars: - self.characterReceived(keyID, False) - else: - log.msg("Received unhandled keyID: %r" % (keyID,)) - - 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 - self.terminal.write(ch) - - def handle_TAB(self): - n = self.TABSTOP - (len(self.lineBuffer) % self.TABSTOP) - self.terminal.cursorForward(n) - self.lineBufferIndex += n - self.lineBuffer.extend(' ' * n) - - def handle_LEFT(self): - if self.lineBufferIndex > 0: - self.lineBufferIndex -= 1 - self.terminal.cursorBackward() - - def handle_RIGHT(self): - if self.lineBufferIndex < len(self.lineBuffer): - self.lineBufferIndex += 1 - self.terminal.cursorForward() - - def handle_HOME(self): - if self.lineBufferIndex: - self.terminal.cursorBackward(self.lineBufferIndex) - self.lineBufferIndex = 0 - - def handle_END(self): - offset = len(self.lineBuffer) - self.lineBufferIndex - if offset: - self.terminal.cursorForward(offset) - self.lineBufferIndex = len(self.lineBuffer) - - def handle_BACKSPACE(self): - if self.lineBufferIndex > 0: - self.lineBufferIndex -= 1 - del self.lineBuffer[self.lineBufferIndex] - self.terminal.cursorBackward() - self.terminal.deleteCharacter() - - def handle_DELETE(self): - if self.lineBufferIndex < len(self.lineBuffer): - del self.lineBuffer[self.lineBufferIndex] - self.terminal.deleteCharacter() - - def handle_RETURN(self): - line = ''.join(self.lineBuffer) - self.lineBuffer = [] - self.lineBufferIndex = 0 - self.terminal.nextLine() - self.lineReceived(line) - - def handle_INSERT(self): - assert self.mode in ('typeover', 'insert') - if self.mode == 'typeover': - self.setInsertMode() - else: - self.setTypeoverMode() - - def lineReceived(self, line): - pass - -class HistoricRecvLine(RecvLine): - """L{TerminalProtocol} which adds both basic line-editing features and input history. - - Everything supported by L{RecvLine} is also supported by this class. In addition, the - up and down arrows traverse the input history. Each received line is automatically - added to the end of the input history. - """ - def connectionMade(self): - RecvLine.connectionMade(self) - - self.historyLines = [] - self.historyPosition = 0 - - t = self.terminal - self.keyHandlers.update({t.UP_ARROW: self.handle_UP, - t.DOWN_ARROW: self.handle_DOWN}) - - def currentHistoryBuffer(self): - b = tuple(self.historyLines) - return b[:self.historyPosition], b[self.historyPosition:] - - def _deliverBuffer(self, buf): - if buf: - for ch in buf[:-1]: - self.characterReceived(ch, True) - self.characterReceived(buf[-1], False) - - def handle_UP(self): - if self.lineBuffer and self.historyPosition == len(self.historyLines): - self.historyLines.append(self.lineBuffer) - if self.historyPosition > 0: - self.handle_HOME() - self.terminal.eraseToLineEnd() - - self.historyPosition -= 1 - self.lineBuffer = [] - - self._deliverBuffer(self.historyLines[self.historyPosition]) - - def handle_DOWN(self): - if self.historyPosition < len(self.historyLines) - 1: - self.handle_HOME() - self.terminal.eraseToLineEnd() - - self.historyPosition += 1 - self.lineBuffer = [] - - self._deliverBuffer(self.historyLines[self.historyPosition]) - else: - self.handle_HOME() - self.terminal.eraseToLineEnd() - - self.historyPosition = len(self.historyLines) - self.lineBuffer = [] - self.lineBufferIndex = 0 - - def handle_RETURN(self): - if self.lineBuffer: - self.historyLines.append(''.join(self.lineBuffer)) - self.historyPosition = len(self.historyLines) - return RecvLine.handle_RETURN(self) |