ipython_view.py

Go to the documentation of this file.

"""
Backend to the console plugin.
 
@author: Eitan Isaacson
@organization: IBM Corporation
@copyright: Copyright (c) 2007 IBM Corporation
@license: BSD
 
All rights reserved. This program and the accompanying materials are made
available under the terms of the BSD which accompanies this distribution, and
is available at U{http://www.opensource.org/licenses/bsd-license.php}
"""
# this file is a modified version of source code from the Accerciser project
# https://wiki.gnome.org/Apps/Accerciser
 
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import GLib
from gi.repository import Pango
 
from pkg_resources import parse_version
 
import re
import sys
import os
 
from io import StringIO
from functools import reduce
 
 
try:
  import IPython
except ImportError:
  
  IPython = None
 
 
class IterableIPShell:
  
  def __init__(self, argv=[], user_ns=None, user_global_ns=None,
               cin=None, cout=None, cerr=None, input_func=None):
    """! Initializer
 
    @param self: this object
    @param argv: Command line options for IPython
    @param user_ns: User namespace.
    @param user_global_ns: User global namespace.
    @param cin: Console standard input.
    @param cout: Console standard output.
    @param cerr: Console standard error.
    @param input_func: Replacement for builtin raw_input()
    """
    io = IPython.utils.io
    if input_func:
      IPython.terminal.interactiveshell.raw_input_original = input_func
    if IPython.version_info < (8,):
      if cin:
        io.stdin = io.IOStream(cin)
      if cout:
        io.stdout = io.IOStream(cout)
      if cerr:
        io.stderr = io.IOStream(cerr)
    else:
      if cin:
        sys.stdin = cin
      if cout:
        sys.stdout = cout
      if cerr:
        sys.stderr = cerr
 
    # This is to get rid of the blockage that occurs during
    # IPython.Shell.InteractiveShell.user_setup()
    io.raw_input = lambda x: None
 
    os.environ['TERM'] = 'dumb'
    excepthook = sys.excepthook
 
    from traitlets.config.loader import Config
    cfg = Config()
    cfg.InteractiveShell.colors = "Linux"
    cfg.Completer.use_jedi = False
 
    if IPython.version_info < (8,):
      # InteractiveShell's __init__ overwrites io.stdout,io.stderr with
      # sys.stdout, sys.stderr, this makes sure they are right
      old_stdout, old_stderr = sys.stdout, sys.stderr
      sys.stdout, sys.stderr = io.stdout.stream, io.stderr.stream
 
    # InteractiveShell inherits from SingletonConfigurable, so use instance()
    #
    self.IP = IPython.terminal.embed.InteractiveShellEmbed.instance(\
              config=cfg, user_ns=user_ns)
 
    if IPython.version_info < (8,):
      sys.stdout, sys.stderr = old_stdout, old_stderr
 
    self.IP.system = lambda cmd: self.shell(self.IP.var_expand(cmd),
                                            header='IPython system call: ')
#                                            local_ns=user_ns)
                                            #global_ns=user_global_ns)
                                            #verbose=self.IP.rc.system_verbose)
 
    self.IP.raw_input = input_func
    sys.excepthook = excepthook
    self.iter_more = 0
    self.history_level = 0
    self.complete_sep =  re.compile('[\s\{\}\[\]\‍(\‍)]')
    self.updateNamespace({'exit':lambda:None})
    self.updateNamespace({'quit':lambda:None})
    # Workaround for updating namespace with sys.modules
    #
    self.__update_namespace()
 
    # Avoid using input splitter when not really needed.
    # Perhaps it could work even before 5.8.0
    # But it definitely does not work any more with >= 7.0.0
    self.no_input_splitter = parse_version(IPython.release.version) >= parse_version('5.8.0')
    self.lines = []
    self.indent_spaces = ''
 
  def __update_namespace(self):
    """!
    Update self.IP namespace for autocompletion with sys.modules
    @return none
    """
    for k, v in list(sys.modules.items()):
        if not '.' in k:
          self.IP.user_ns.update({k:v})
 
  def execute(self):
    """!
    Executes the current line provided by the shell object.
    @return none
    """
    self.history_level = 0
 
    if IPython.version_info < (8,):
      # this is needed because some functions in IPython use 'print' to print
      # output (like 'who')
 
      orig_stdout = sys.stdout
      sys.stdout = IPython.utils.io.stdout
 
      orig_stdin = sys.stdin
      sys.stdin = IPython.utils.io.stdin
 
    self.prompt = self.generatePrompt(self.iter_more)
 
    self.IP.hooks.pre_prompt_hook()
    if self.iter_more:
        try:
            self.prompt = self.generatePrompt(True)
        except:
            self.IP.showtraceback()
        if self.IP.autoindent:
            self.IP.rl_do_indent = True
 
    try:
      line = self.IP.raw_input(self.prompt)
    except KeyboardInterrupt:
      self.IP.write('\nKeyboardInterrupt\n')
      if self.no_input_splitter:
        self.lines = []
      else:
        self.IP.input_splitter.reset()
    except:
      self.IP.showtraceback()
    else:
      if self.no_input_splitter:
        self.lines.append(line)
        (status, self.indent_spaces) = self.IP.check_complete('\n'.join(self.lines))
        self.iter_more = status == 'incomplete'
      else:
        self.IP.input_splitter.push(line)
        self.iter_more = self.IP.input_splitter.push_accepts_more()
      if not self.iter_more:
          if self.no_input_splitter:
            source_raw = '\n'.join(self.lines)
            self.lines = []
          else:
            source_raw = self.IP.input_splitter.raw_reset()
          self.IP.run_cell(source_raw, store_history=True)
          self.IP.rl_do_indent = False
      else:
          # TODO: Auto-indent
          #
          self.IP.rl_do_indent = True
          pass
      self.prompt = self.generatePrompt(self.iter_more)
 
    if IPython.version_info < (8,):
      sys.stdout = orig_stdout
      sys.stdin = orig_stdin
 
  def generatePrompt(self, is_continuation):
    """!
    Generate prompt depending on is_continuation value
 
    @param is_continuation
    @return: The prompt string representation
 
    """
 
    if is_continuation:
      prompt = '... '
    else:
      prompt = '>>> '
    return prompt
 
 
  def historyBack(self):
    """!
    Provides one history command back.
 
    @param self this object
    @return: The command string.
    """
    self.history_level -= 1
    if not self._getHistory():
      self.history_level +=1
    return self._getHistory()
 
  def historyForward(self):
    """!
    Provides one history command forward.
 
    @param self this object
    @return: The command string.
    """
    if self.history_level < 0:
      self.history_level += 1
    return self._getHistory()
 
  def _getHistory(self):
    """!
    Gets the command string of the current history level.
 
    @param self this object
    @return: Historic command string.
    """
    try:
      rv = self.IP.user_ns['In'][self.history_level].strip('\n')
    except IndexError:
      rv = ''
    return rv
 
  def updateNamespace(self, ns_dict):
    """!
    Add the current dictionary to the shell namespace.
 
    @param ns_dict: A dictionary of symbol-values.
    @return none
    """
    self.IP.user_ns.update(ns_dict)
 
  def complete(self, line):
    """!
    Returns an auto completed line and/or possibilities for completion.
 
    @param line: Given line so far.
    @return: Line completed as for as possible, and possible further completions.
    """
    split_line = self.complete_sep.split(line)
    if split_line[-1]:
      possibilities = self.IP.complete(split_line[-1])
    else:
      completed = line
      possibilities = ['', []]
    if possibilities:
      def _commonPrefix(str1, str2):
        """!
        Reduction function. returns common prefix of two given strings.
 
        @param str1: First string.
        @param str2: Second string
        @return: Common prefix to both strings.
        """
        for i in range(len(str1)):
          if not str2.startswith(str1[:i+1]):
            return str1[:i]
        return str1
      if possibilities[1]:
        common_prefix = reduce(_commonPrefix, possibilities[1]) or split_line[-1]
        completed = line[:-len(split_line[-1])]+common_prefix
      else:
        completed = line
    else:
      completed = line
    return completed, possibilities[1]
 
 
  def shell(self, cmd,verbose=0,debug=0,header=''):
    """!
    Replacement method to allow shell commands without them blocking.
 
    @param cmd: Shell command to execute.
    @param verbose: Verbosity
    @param debug: Debug level
    @param header: Header to be printed before output
    @return none
    """
    stat = 0
    if verbose or debug: print(header+cmd)
    # flush stdout so we don't mangle python's buffering
    if not debug:
      input, output = os.popen4(cmd)
      print(output.read())
      output.close()
      input.close()
 
 
class ConsoleView(Gtk.TextView):
  
  """
  Specialized text view for console-like workflow.
 
  @cvar ANSI_COLORS: Mapping of terminal control sequence values to
                     tuples containing foreground and background color names.
  @type ANSI_COLORS: dictionary
 
  @ivar text_buffer: Widget's text buffer.  @type text_buffer: Gtk.TextBuffer
  @ivar color_pat: Regex of terminal color pattern
  @type color_pat: _sre.SRE_Pattern
  @ivar mark: Scroll mark for automatic scrolling on input.
  @type mark: Gtk.TextMark
  @ivar line_start: Start of command line mark.
  @type line_start: Gtk.TextMark
  """
  ANSI_COLORS = {'0;30': ('Black', None),
                 '0;31': ('Red', None),
                 '0;32': ('Green', None),
                 '0;33': ('Brown', None),
                 '0;34': ('Blue', None),
                 '0;35': ('Purple', None),
                 '0;36': ('Cyan', None),
                 '0;37': ('LightGray', None),
                 '1;30': ('DarkGray', None),
                 '1;31': ('DarkRed', None),
                 '1;32': ('SeaGreen', None),
                 '1;33': ('Yellow', None),
                 '1;34': ('LightBlue', None),
                 '1;35': ('MediumPurple', None),
                 '1;36': ('LightCyan', None),
                 '1;37': ('White', None),
                 '38;5;124;43': ('DarkRed', 'Yellow'),
                 '38;5;241': ('Gray', None),
                 '38;5;241;43': ('Gray', 'Yellow'),
                 '39': ('Black', None),
                 '39;49': ('Red', 'White'),
                 '43': (None, 'Yellow'),
                 '49': (None, 'White')}
 
  def __init__(self):
    """
    Initialize console view.
    """
    Gtk.TextView.__init__(self)
    self.modify_font(Pango.FontDescription('Mono'))
    self.set_cursor_visible(True)
    self.text_buffer = self.get_buffer()
    self.mark = self.text_buffer.create_mark('scroll_mark',
                                             self.text_buffer.get_end_iter(),
                                             False)
    for code in self.ANSI_COLORS:
      self.text_buffer.create_tag(code,
                                  foreground=self.ANSI_COLORS[code][0],
                                  background=self.ANSI_COLORS[code][1],
                                  weight=700)
    self.text_buffer.create_tag('0')
    self.text_buffer.create_tag('notouch', editable=False)
    self.color_pat = re.compile('\x01?\x1b\[(.*?)m\x02?')
    self.line_start = \
        self.text_buffer.create_mark('line_start',
                                     self.text_buffer.get_end_iter(), True)
    self.connect('key-press-event', self.onKeyPress)
 
  def write(self, text, editable=False):
    """!
    Write given text to buffer.
 
    @param text: Text to append.
    @param editable: If true, added text is editable.
    @return none
    """
    GLib.idle_add(self._write, text, editable)
 
  def _write(self, text, editable=False):
    """!
    Write given text to buffer.
 
    @param text: Text to append.
    @param editable: If true, added text is editable.
    @return none
    """
    segments = self.color_pat.split(text)
    segment = segments.pop(0)
    start_mark = self.text_buffer.create_mark(None,
                                              self.text_buffer.get_end_iter(),
                                              True)
    self.text_buffer.insert(self.text_buffer.get_end_iter(), segment)
 
    if segments:
      ansi_tags = self.color_pat.findall(text)
      for tag in ansi_tags:
        i = segments.index(tag)
        self.text_buffer.insert_with_tags_by_name(self.text_buffer.get_end_iter(),
                                                  segments[i+1], str(tag))
        segments.pop(i)
    if not editable:
      self.text_buffer.apply_tag_by_name('notouch',
                                         self.text_buffer.get_iter_at_mark(start_mark),
                                         self.text_buffer.get_end_iter())
    self.text_buffer.delete_mark(start_mark)
    self.scroll_mark_onscreen(self.mark)
 
  def showPrompt(self, prompt):
    """!
    Prints prompt at start of line.
 
    @param prompt: Prompt to print.
    @return none
    """
    GLib.idle_add(self._showPrompt, prompt)
 
  def _showPrompt(self, prompt):
    """!
    Prints prompt at start of line.
 
    @param prompt: Prompt to print.
    @return none
    """
    self._write(prompt)
    self.text_buffer.move_mark(self.line_start,
                               self.text_buffer.get_end_iter())
 
  def changeLine(self, text):
    """!
    Replace currently entered command line with given text.
 
    @param text: Text to use as replacement.
    @return none
    """
    GLib.idle_add(self._changeLine, text)
 
  def _changeLine(self, text):
    """!
    Replace currently entered command line with given text.
 
    @param text: Text to use as replacement.
    @return none
    """
    iter = self.text_buffer.get_iter_at_mark(self.line_start)
    iter.forward_to_line_end()
    self.text_buffer.delete(self.text_buffer.get_iter_at_mark(self.line_start), iter)
    self._write(text, True)
 
  def getCurrentLine(self):
    """!
    Get text in current command line.
 
    @return Text of current command line.
    """
    rv = self.text_buffer.get_slice(
      self.text_buffer.get_iter_at_mark(self.line_start),
      self.text_buffer.get_end_iter(), False)
    return rv
 
  def showReturned(self, text):
    """!
    Show returned text from last command and print new prompt.
 
    @param text: Text to show.
    @return none
    """
    GLib.idle_add(self._showReturned, text)
 
  def _showReturned(self, text):
    """!
    Show returned text from last command and print new prompt.
 
    @param text: Text to show.
    @return none
    """
    iter = self.text_buffer.get_iter_at_mark(self.line_start)
    iter.forward_to_line_end()
    self.text_buffer.apply_tag_by_name(
      'notouch',
      self.text_buffer.get_iter_at_mark(self.line_start),
      iter)
    self._write('\n'+text)
    if text:
      self._write('\n')
    self._showPrompt(self.prompt)
    self.text_buffer.move_mark(self.line_start, self.text_buffer.get_end_iter())
    self.text_buffer.place_cursor(self.text_buffer.get_end_iter())
 
    if self.IP.rl_do_indent:
      if self.no_input_splitter:
        indentation = self.indent_spaces
      else:
        indentation = self.IP.input_splitter.indent_spaces * ' '
      self.text_buffer.insert_at_cursor(indentation)
 
  def onKeyPress(self, widget, event):
    """!
    Key press callback used for correcting behavior for console-like
    interfaces. For example 'home' should go to prompt, not to beginning of
    line.
 
    @param widget: Widget that key press accored in.
    @param event: Event object
    @return Return True if event should not trickle.
    """
    insert_mark = self.text_buffer.get_insert()
    insert_iter = self.text_buffer.get_iter_at_mark(insert_mark)
    selection_mark = self.text_buffer.get_selection_bound()
    selection_iter = self.text_buffer.get_iter_at_mark(selection_mark)
    start_iter = self.text_buffer.get_iter_at_mark(self.line_start)
    if event.keyval == Gdk.KEY_Home:
      if event.state & Gdk.ModifierType.CONTROL_MASK or event.state & Gdk.ModifierType.MOD1_MASK:
        pass
      elif event.state & Gdk.ModifierType.SHIFT_MASK:
        self.text_buffer.move_mark(insert_mark, start_iter)
        return True
      else:
        self.text_buffer.place_cursor(start_iter)
        return True
    elif event.keyval == Gdk.KEY_Left:
      insert_iter.backward_cursor_position()
      if not insert_iter.editable(True):
        return True
    elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval in [ord('L'), ord('l')]:
        # clear previous output on Ctrl+L, but remember current input line + cursor position
        cursor_offset = self.text_buffer.get_property('cursor-position')
        cursor_pos_in_line = cursor_offset - start_iter.get_offset() + len(self.prompt)
        current_input = self.text_buffer.get_text(start_iter, self.text_buffer.get_end_iter(), False)
        self.text_buffer.set_text(self.prompt + current_input)
        self.text_buffer.move_mark(self.line_start, self.text_buffer.get_iter_at_offset(len(self.prompt)))
        self.text_buffer.place_cursor(self.text_buffer.get_iter_at_offset(cursor_pos_in_line))
        return True
    elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval in [Gdk.KEY_k, Gdk.KEY_K]:
      # clear text after input cursor on Ctrl+K
      if insert_iter.editable(True):
        self.text_buffer.delete(insert_iter, self.text_buffer.get_end_iter())
      return True
    elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_C:
      # copy selection on Ctrl+C (upper-case 'C' only)
      self.text_buffer.copy_clipboard(Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD))
      return True
    elif not event.string:
      pass
    elif start_iter.compare(insert_iter) <= 0 and \
          start_iter.compare(selection_iter) <= 0:
      pass
    elif start_iter.compare(insert_iter) > 0 and \
          start_iter.compare(selection_iter) > 0:
      self.text_buffer.place_cursor(start_iter)
    elif insert_iter.compare(selection_iter) < 0:
      self.text_buffer.move_mark(insert_mark, start_iter)
    elif insert_iter.compare(selection_iter) > 0:
      self.text_buffer.move_mark(selection_mark, start_iter)
 
    return self.onKeyPressExtend(event)
 
  def onKeyPressExtend(self, event):
    """!
    For some reason we can't extend onKeyPress directly (bug #500900).
    @param event key press
    @return none
    """
    pass
 
 
class IPythonView(ConsoleView, IterableIPShell):
  
  """
  Sub-class of both modified IPython shell and L{ConsoleView} this makes
  a GTK+ IPython console.
  """
  def __init__(self):
    """
    Initialize. Redirect I/O to console.
    """
    ConsoleView.__init__(self)
    self.cout = StringIO()
    IterableIPShell.__init__(self, cout=self.cout, cerr=self.cout,
                             input_func=self.raw_input)
    self.interrupt = False
    self.execute()
    self.promptprompt = self.generatePrompt(False)
    self.cout.truncate(0)
    self.showPrompt(self.promptprompt)
 
  def raw_input(self, prompt=''):
    """!
    Custom raw_input() replacement. Gets current line from console buffer.
 
    @param prompt: Prompt to print. Here for compatibility as replacement.
    @return The current command line text.
    """
    if self.interrupt:
      self.interrupt = False
      raise KeyboardInterrupt
    return self.getCurrentLine()
 
  def onKeyPressExtend(self, event):
    """!
    Key press callback with plenty of shell goodness, like history,
    autocompletions, etc.
 
    @param event: Event object.
    @return True if event should not trickle.
    """
 
    if event.get_state() & Gdk.ModifierType.CONTROL_MASK and event.keyval == 99:
      self.interrupt = True
      self._processLine()
      return True
    elif event.keyval == Gdk.KEY_Return:
      self._processLine()
      return True
    elif event.keyval == Gdk.KEY_Up:
      self.changeLine(self.historyBack())
      return True
    elif event.keyval == Gdk.KEY_Down:
      self.changeLine(self.historyForward())
      return True
    elif event.keyval == Gdk.KEY_Tab:
      if not self.getCurrentLine().strip():
        return False
      completed, possibilities = self.complete(self.getCurrentLine())
      if len(possibilities) > 1:
        slice = self.getCurrentLine()
        self.write('\n')
        for symbol in possibilities:
          self.write(symbol+'\n')
        self.showPrompt(self.promptprompt)
      self.changeLine(completed or slice)
      return True
 
  def _processLine(self):
    """!
    Process current command line.
    @return none
    """
    self.history_pos = 0
    self.execute()
    rv = self.cout.getvalue()
    if rv: rv = rv.strip('\n')
    self.showReturned(rv)
    self.cout.truncate(0)
    self.cout.seek(0)
 
if __name__ == "__main__":
  window = Gtk.Window()
  window.set_default_size(640, 320)
  window.connect('delete-event', lambda x, y: Gtk.main_quit())
  window.add(IPythonView())
  window.show_all()
  Gtk.main()