diff --git a/experiments/cursesio.py b/experiments/cursesio.py
new file mode 100644
index 0000000000..83b4a9f186
--- /dev/null
+++ b/experiments/cursesio.py
@@ -0,0 +1,154 @@
+import curses
+import curses.wrapper
+from curses.ascii import isprint
+
+from twisted.internet import reactor
+
+
+class CursesStdIO():
+ def __init__(self, stdscr, callback=None):
+ self.statusText = "Synapse test app -"
+ self.searchText = ''
+ self.stdscr = stdscr
+
+ self.logLine = ''
+
+ self.callback = callback
+
+ self._setup()
+
+ def _setup(self):
+ self.stdscr.nodelay(1) # Make non blocking
+
+ self.rows, self.cols = self.stdscr.getmaxyx()
+ self.lines = []
+
+ curses.use_default_colors()
+
+ self.paintStatus(self.statusText)
+ self.stdscr.refresh()
+
+ def set_callback(self, callback):
+ self.callback = callback
+
+ def fileno(self):
+ """ We want to select on FD 0 """
+ return 0
+
+ def connectionLost(self, reason):
+ self.close()
+
+ def print_line(self, text):
+ """ add a line to the internal list of lines"""
+
+ self.lines.append(text)
+ self.redraw()
+
+ def print_log(self, text):
+ self.logLine = text
+ self.redraw()
+
+ def redraw(self):
+ """ method for redisplaying lines
+ based on internal list of lines """
+
+ self.stdscr.clear()
+ self.paintStatus(self.statusText)
+ i = 0
+ index = len(self.lines) - 1
+ while i < (self.rows - 3) and index >= 0:
+ self.stdscr.addstr(self.rows - 3 - i, 0, self.lines[index],
+ curses.A_NORMAL)
+ i = i + 1
+ index = index - 1
+
+ self.printLogLine(self.logLine)
+
+ self.stdscr.refresh()
+
+ def paintStatus(self, text):
+ if len(text) > self.cols:
+ raise RuntimeError("TextTooLongError")
+
+ self.stdscr.addstr(
+ self.rows - 2, 0,
+ text + ' ' * (self.cols - len(text)),
+ curses.A_STANDOUT)
+
+ def printLogLine(self, text):
+ self.stdscr.addstr(
+ 0, 0,
+ text + ' ' * (self.cols - len(text)),
+ curses.A_STANDOUT)
+
+ def doRead(self):
+ """ Input is ready! """
+ curses.noecho()
+ c = self.stdscr.getch() # read a character
+
+ if c == curses.KEY_BACKSPACE:
+ self.searchText = self.searchText[:-1]
+
+ elif c == curses.KEY_ENTER or c == 10:
+ text = self.searchText
+ self.searchText = ''
+
+ self.print_line(">> %s" % text)
+
+ try:
+ if self.callback:
+ self.callback.on_line(text)
+ except Exception as e:
+ self.print_line(str(e))
+
+ self.stdscr.refresh()
+
+ elif isprint(c):
+ if len(self.searchText) == self.cols - 2:
+ return
+ self.searchText = self.searchText + chr(c)
+
+ self.stdscr.addstr(self.rows - 1, 0,
+ self.searchText + (' ' * (
+ self.cols - len(self.searchText) - 2)))
+
+ self.paintStatus(self.statusText + ' %d' % len(self.searchText))
+ self.stdscr.move(self.rows - 1, len(self.searchText))
+ self.stdscr.refresh()
+
+ def logPrefix(self):
+ return "CursesStdIO"
+
+ def close(self):
+ """ clean up """
+
+ curses.nocbreak()
+ self.stdscr.keypad(0)
+ curses.echo()
+ curses.endwin()
+
+
+class Callback(object):
+
+ def __init__(self, stdio):
+ self.stdio = stdio
+
+ def on_line(self, text):
+ self.stdio.print_line(text)
+
+
+def main(stdscr):
+ screen = CursesStdIO(stdscr) # create Screen object
+
+ callback = Callback(screen)
+
+ screen.set_callback(callback)
+
+ stdscr.refresh()
+ reactor.addReader(screen)
+ reactor.run()
+ screen.close()
+
+
+if __name__ == '__main__':
+ curses.wrapper(main)
|
