# encoding: utf-8 -*- test-case-name: # FIXME: Need to add tests. # ipython1.frontend.wx.tests.test_wx_frontend -*- """Classes to provide a Wx frontend to the IPython.kernel.core.interpreter. This class inherits from ConsoleWidget, that provides a console-like widget to provide a text-rendering widget suitable for a terminal. """ __docformat__ = "restructuredtext en" #------------------------------------------------------------------------------- # Copyright (C) 2008 The IPython Development Team # # Distributed under the terms of the BSD License. The full license is in # the file COPYING, distributed as part of this software. #------------------------------------------------------------------------------- #------------------------------------------------------------------------------- # Imports #------------------------------------------------------------------------------- # Major library imports import re import __builtin__ import sys from threading import Lock import wx from wx import stc # Ipython-specific imports. from IPython.frontend.process import PipedProcess from console_widget import ConsoleWidget, _COMPLETE_BUFFER_MARKER, \ _ERROR_MARKER, _INPUT_MARKER from IPython.frontend.prefilterfrontend import PrefilterFrontEnd #------------------------------------------------------------------------------- # Classes to implement the Wx frontend #------------------------------------------------------------------------------- class WxController(ConsoleWidget, PrefilterFrontEnd): """Classes to provide a Wx frontend to the IPython.kernel.core.interpreter. This class inherits from ConsoleWidget, that provides a console-like widget to provide a text-rendering widget suitable for a terminal. """ # Print debug info on what is happening to the console. debug = False # The title of the terminal, as captured through the ANSI escape # sequences. def _set_title(self, title): return self.Parent.SetTitle(title) def _get_title(self): return self.Parent.GetTitle() title = property(_get_title, _set_title) # The buffer being edited. # We are duplicating the definition here because of multiple # inheritence def _set_input_buffer(self, string): ConsoleWidget._set_input_buffer(self, string) self._colorize_input_buffer() def _get_input_buffer(self): """ Returns the text in current edit buffer. """ return ConsoleWidget._get_input_buffer(self) input_buffer = property(_get_input_buffer, _set_input_buffer) #-------------------------------------------------------------------------- # Private Attributes #-------------------------------------------------------------------------- # A flag governing the behavior of the input. Can be: # # 'readline' for readline-like behavior with a prompt # and an edit buffer. # 'raw_input' similar to readline, but triggered by a raw-input # call. Can be used by subclasses to act differently. # 'subprocess' for sending the raw input directly to a # subprocess. # 'buffering' for buffering of the input, that will be used # when the input state switches back to another state. _input_state = 'readline' # Attribute to store reference to the pipes of a subprocess, if we # are running any. _running_process = False # A queue for writing fast streams to the screen without flooding the # event loop _out_buffer = [] # A lock to lock the _out_buffer to make sure we don't empty it # while it is being swapped _out_buffer_lock = Lock() # The different line markers used to higlight the prompts. _markers = dict() #-------------------------------------------------------------------------- # Public API #-------------------------------------------------------------------------- def __init__(self, parent, id=wx.ID_ANY, pos=wx.DefaultPosition, size=wx.DefaultSize, style=wx.CLIP_CHILDREN|wx.WANTS_CHARS, styledef=None, *args, **kwds): """ Create Shell instance. Parameters ----------- styledef : dict, optional styledef is the dictionary of options used to define the style. """ if styledef is not None: self.style = styledef ConsoleWidget.__init__(self, parent, id, pos, size, style) PrefilterFrontEnd.__init__(self, **kwds) # Stick in our own raw_input: self.ipython0.raw_input = self.raw_input # A time for flushing the write buffer BUFFER_FLUSH_TIMER_ID = 100 self._buffer_flush_timer = wx.Timer(self, BUFFER_FLUSH_TIMER_ID) wx.EVT_TIMER(self, BUFFER_FLUSH_TIMER_ID, self._buffer_flush) if 'debug' in kwds: self.debug = kwds['debug'] kwds.pop('debug') # Inject self in namespace, for debug if self.debug: self.shell.user_ns['self'] = self # Inject our own raw_input in namespace self.shell.user_ns['raw_input'] = self.raw_input def raw_input(self, prompt=''): """ A replacement from python's raw_input. """ self.new_prompt(prompt) self._input_state = 'raw_input' if hasattr(self, '_cursor'): del self._cursor self.SetCursor(wx.StockCursor(wx.CURSOR_CROSS)) self.__old_on_enter = self._on_enter event_loop = wx.EventLoop() def my_on_enter(): event_loop.Exit() self._on_enter = my_on_enter # XXX: Running a separate event_loop. Ugly. event_loop.Run() self._on_enter = self.__old_on_enter self._input_state = 'buffering' self._cursor = wx.BusyCursor() return self.input_buffer.rstrip('\n') def system_call(self, command_string): self._input_state = 'subprocess' event_loop = wx.EventLoop() def _end_system_call(): self._input_state = 'buffering' self._running_process = False event_loop.Exit() self._running_process = PipedProcess(command_string, out_callback=self.buffered_write, end_callback = _end_system_call) self._running_process.start() # XXX: Running a separate event_loop. Ugly. event_loop.Run() # Be sure to flush the buffer. self._buffer_flush(event=None) def do_calltip(self): """ Analyse current and displays useful calltip for it. """ if self.debug: print >>sys.__stdout__, "do_calltip" separators = re.compile('[\s\{\}\[\]\(\)\= ,:]') symbol = self.input_buffer symbol_string = separators.split(symbol)[-1] base_symbol_string = symbol_string.split('.')[0] if base_symbol_string in self.shell.user_ns: symbol = self.shell.user_ns[base_symbol_string] elif base_symbol_string in self.shell.user_global_ns: symbol = self.shell.user_global_ns[base_symbol_string] elif base_symbol_string in __builtin__.__dict__: symbol = __builtin__.__dict__[base_symbol_string] else: return False try: for name in symbol_string.split('.')[1:] + ['__doc__']: symbol = getattr(symbol, name) self.AutoCompCancel() # Check that the symbol can indeed be converted to a string: symbol += '' wx.CallAfter(self.CallTipShow, self.GetCurrentPos(), symbol) except: # The retrieve symbol couldn't be converted to a string pass def _popup_completion(self, create=False): """ Updates the popup completion menu if it exists. If create is true, open the menu. """ if self.debug: print >>sys.__stdout__, "_popup_completion" line = self.input_buffer if (self.AutoCompActive() and line and not line[-1] == '.') \ or create==True: suggestion, completions = self.complete(line) if completions: offset = len(self._get_completion_text(line)) self.pop_completion(completions, offset=offset) if self.debug: print >>sys.__stdout__, completions def buffered_write(self, text): """ A write method for streams, that caches the stream in order to avoid flooding the event loop. This can be called outside of the main loop, in separate threads. """ self._out_buffer_lock.acquire() self._out_buffer.append(text) self._out_buffer_lock.release() if not self._buffer_flush_timer.IsRunning(): wx.CallAfter(self._buffer_flush_timer.Start, milliseconds=100, oneShot=True) def clear_screen(self): """ Empty completely the widget. """ self.ClearAll() self.new_prompt(self.input_prompt_template.substitute( number=(self.last_result['number'] + 1))) #-------------------------------------------------------------------------- # LineFrontEnd interface #-------------------------------------------------------------------------- def execute(self, python_string, raw_string=None): self._input_state = 'buffering' self.CallTipCancel() self._cursor = wx.BusyCursor() if raw_string is None: raw_string = python_string end_line = self.current_prompt_line \ + max(1, len(raw_string.split('\n'))-1) for i in range(self.current_prompt_line, end_line): if i in self._markers: self.MarkerDeleteHandle(self._markers[i]) self._markers[i] = self.MarkerAdd(i, _COMPLETE_BUFFER_MARKER) # Use a callafter to update the display robustly under windows def callback(): self.GotoPos(self.GetLength()) PrefilterFrontEnd.execute(self, python_string, raw_string=raw_string) wx.CallAfter(callback) def execute_command(self, command, hidden=False): """ Execute a command, not only in the model, but also in the view. """ # XXX: This method needs to be integrated in the base fronted # interface if hidden: return self.shell.execute(command) else: # XXX: we are not storing the input buffer previous to the # execution, as this forces us to run the execution # input_buffer a yield, which is not good. ##current_buffer = self.shell.control.input_buffer command = command.rstrip() if len(command.split('\n')) > 1: # The input command is several lines long, we need to # force the execution to happen command += '\n' cleaned_command = self.prefilter_input(command) self.input_buffer = command # Do not use wx.Yield() (aka GUI.process_events()) to avoid # recursive yields. self.ProcessEvent(wx.PaintEvent()) self.write('\n') if not self.is_complete(cleaned_command + '\n'): self._colorize_input_buffer() self.render_error('Incomplete or invalid input') self.new_prompt(self.input_prompt_template.substitute( number=(self.last_result['number'] + 1))) return False self._on_enter() return True def save_output_hooks(self): self.__old_raw_input = __builtin__.raw_input PrefilterFrontEnd.save_output_hooks(self) def capture_output(self): self.SetLexer(stc.STC_LEX_NULL) PrefilterFrontEnd.capture_output(self) __builtin__.raw_input = self.raw_input def release_output(self): __builtin__.raw_input = self.__old_raw_input PrefilterFrontEnd.release_output(self) self.SetLexer(stc.STC_LEX_PYTHON) def after_execute(self): PrefilterFrontEnd.after_execute(self) # Clear the wait cursor if hasattr(self, '_cursor'): del self._cursor self.SetCursor(wx.StockCursor(wx.CURSOR_CHAR)) def show_traceback(self): start_line = self.GetCurrentLine() PrefilterFrontEnd.show_traceback(self) self.ProcessEvent(wx.PaintEvent()) #wx.Yield() for i in range(start_line, self.GetCurrentLine()): self._markers[i] = self.MarkerAdd(i, _ERROR_MARKER) #-------------------------------------------------------------------------- # FrontEndBase interface #-------------------------------------------------------------------------- def render_error(self, e): start_line = self.GetCurrentLine() self.write('\n' + e + '\n') for i in range(start_line, self.GetCurrentLine()): self._markers[i] = self.MarkerAdd(i, _ERROR_MARKER) #-------------------------------------------------------------------------- # ConsoleWidget interface #-------------------------------------------------------------------------- def new_prompt(self, prompt): """ Display a new prompt, and start a new input buffer. """ self._input_state = 'readline' ConsoleWidget.new_prompt(self, prompt) i = self.current_prompt_line self._markers[i] = self.MarkerAdd(i, _INPUT_MARKER) def continuation_prompt(self, *args, **kwargs): # Avoid multiple inheritence, be explicit about which # parent method class gets called return ConsoleWidget.continuation_prompt(self, *args, **kwargs) def write(self, *args, **kwargs): # Avoid multiple inheritence, be explicit about which # parent method class gets called return ConsoleWidget.write(self, *args, **kwargs) def _on_key_down(self, event, skip=True): """ Capture the character events, let the parent widget handle them, and put our logic afterward. """ # FIXME: This method needs to be broken down in smaller ones. current_line_num = self.GetCurrentLine() key_code = event.GetKeyCode() if key_code in (ord('c'), ord('C')) and event.ControlDown(): # Capture Control-C if self._input_state == 'subprocess': if self.debug: print >>sys.__stderr__, 'Killing running process' if hasattr(self._running_process, 'process'): self._running_process.process.kill() elif self._input_state == 'buffering': if self.debug: print >>sys.__stderr__, 'Raising KeyboardInterrupt' raise KeyboardInterrupt # XXX: We need to make really sure we # get back to a prompt. elif self._input_state == 'subprocess' and ( ( key_code <256 and not event.ControlDown() ) or ( key_code in (ord('d'), ord('D')) and event.ControlDown())): # We are running a process, we redirect keys. ConsoleWidget._on_key_down(self, event, skip=skip) char = chr(key_code) # Deal with some inconsistency in wx keycodes: if char == '\r': char = '\n' elif not event.ShiftDown(): char = char.lower() if event.ControlDown() and key_code in (ord('d'), ord('D')): char = '\04' self._running_process.process.stdin.write(char) self._running_process.process.stdin.flush() elif key_code in (ord('('), 57, 53): # Calltips event.Skip() self.do_calltip() elif self.AutoCompActive() and not key_code == ord('\t'): event.Skip() if key_code in (wx.WXK_BACK, wx.WXK_DELETE): wx.CallAfter(self._popup_completion, create=True) elif not key_code in (wx.WXK_UP, wx.WXK_DOWN, wx.WXK_LEFT, wx.WXK_RIGHT, wx.WXK_ESCAPE): wx.CallAfter(self._popup_completion) else: # Up history if key_code == wx.WXK_UP and ( event.ControlDown() or current_line_num == self.current_prompt_line ): new_buffer = self.get_history_previous( self.input_buffer) if new_buffer is not None: self.input_buffer = new_buffer if self.GetCurrentLine() > self.current_prompt_line: # Go to first line, for seemless history up. self.GotoPos(self.current_prompt_pos) # Down history elif key_code == wx.WXK_DOWN and ( event.ControlDown() or current_line_num == self.LineCount -1 ): new_buffer = self.get_history_next() if new_buffer is not None: self.input_buffer = new_buffer # Tab-completion elif key_code == ord('\t'): current_line, current_line_num = self.CurLine if not re.match(r'^%s\s*$' % self.continuation_prompt(), current_line): self.complete_current_input() if self.AutoCompActive(): wx.CallAfter(self._popup_completion, create=True) else: event.Skip() elif key_code == wx.WXK_BACK: # If characters where erased, check if we have to # remove a line. # XXX: What about DEL? # FIXME: This logics should be in ConsoleWidget, as it is # independant of IPython current_line, _ = self.CurLine current_pos = self.GetCurrentPos() current_line_num = self.LineFromPosition(current_pos) current_col = self.GetColumn(current_pos) len_prompt = len(self.continuation_prompt()) if ( current_line.startswith(self.continuation_prompt()) and current_col == len_prompt): new_lines = [] for line_num, line in enumerate( self.input_buffer.split('\n')): if (line_num + self.current_prompt_line == current_line_num): new_lines.append(line[len_prompt:]) else: new_lines.append('\n'+line) # The first character is '\n', due to the above # code: self.input_buffer = ''.join(new_lines)[1:] self.GotoPos(current_pos - 1 - len_prompt) else: ConsoleWidget._on_key_down(self, event, skip=skip) else: ConsoleWidget._on_key_down(self, event, skip=skip) def _on_key_up(self, event, skip=True): """ Called when any key is released. """ if event.GetKeyCode() in (59, ord('.')): # Intercepting '.' event.Skip() wx.CallAfter(self._popup_completion, create=True) else: ConsoleWidget._on_key_up(self, event, skip=skip) # Make sure the continuation_prompts are always followed by a # whitespace new_lines = [] if self._input_state == 'readline': position = self.GetCurrentPos() continuation_prompt = self.continuation_prompt()[:-1] for line in self.input_buffer.split('\n'): if not line == continuation_prompt: new_lines.append(line) self.input_buffer = '\n'.join(new_lines) self.GotoPos(position) def _on_enter(self): """ Called on return key down, in readline input_state. """ last_line_num = self.LineFromPosition(self.GetLength()) current_line_num = self.LineFromPosition(self.GetCurrentPos()) new_line_pos = (last_line_num - current_line_num) if self.debug: print >>sys.__stdout__, repr(self.input_buffer) self.write('\n', refresh=False) # Under windows scintilla seems to be doing funny # stuff to the line returns here, but the getter for # input_buffer filters this out. if sys.platform == 'win32': self.input_buffer = self.input_buffer old_prompt_num = self.current_prompt_pos has_executed = PrefilterFrontEnd._on_enter(self, new_line_pos=new_line_pos) if old_prompt_num == self.current_prompt_pos: # No execution has happened self.GotoPos(self.GetLineEndPosition(current_line_num + 1)) return has_executed #-------------------------------------------------------------------------- # EditWindow API #-------------------------------------------------------------------------- def OnUpdateUI(self, event): """ Override the OnUpdateUI of the EditWindow class, to prevent syntax highlighting both for faster redraw, and for more consistent look and feel. """ if not self._input_state == 'readline': ConsoleWidget.OnUpdateUI(self, event) #-------------------------------------------------------------------------- # Private API #-------------------------------------------------------------------------- def _buffer_flush(self, event): """ Called by the timer to flush the write buffer. This is always called in the mainloop, by the wx timer. """ self._out_buffer_lock.acquire() _out_buffer = self._out_buffer self._out_buffer = [] self._out_buffer_lock.release() self.write(''.join(_out_buffer), refresh=False) def _colorize_input_buffer(self): """ Keep the input buffer lines at a bright color. """ if not self._input_state in ('readline', 'raw_input'): return end_line = self.GetCurrentLine() if not sys.platform == 'win32': end_line += 1 for i in range(self.current_prompt_line, end_line): if i in self._markers: self.MarkerDeleteHandle(self._markers[i]) self._markers[i] = self.MarkerAdd(i, _INPUT_MARKER) if __name__ == '__main__': class MainWindow(wx.Frame): def __init__(self, parent, id, title): wx.Frame.__init__(self, parent, id, title, size=(300,250)) self._sizer = wx.BoxSizer(wx.VERTICAL) self.shell = WxController(self) self._sizer.Add(self.shell, 1, wx.EXPAND) self.SetSizer(self._sizer) self.SetAutoLayout(1) self.Show(True) app = wx.PySimpleApp() frame = MainWindow(None, wx.ID_ANY, 'Ipython') frame.shell.SetFocus() frame.SetSize((680, 460)) self = frame.shell app.MainLoop()