trunk/pygtk/ipython_view.py

"""
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
# http://live.gnome.org/accerciser

import gtk
import re
import sys
import os
import pango
from StringIO import StringIO

try:
    import IPython
except Exception,e:
    raise "Error importing IPython (%s)" % str(e)

ansi_colors =  {'0;30': 'Black',
                '0;31': 'Red',
                '0;32': 'Green',
                '0;33': 'Brown',
                '0;34': 'Blue',
                '0;35': 'Purple',
                '0;36': 'Cyan',
                '0;37': 'LightGray',
                '1;30': 'DarkGray',
                '1;31': 'DarkRed',
                '1;32': 'SeaGreen',
                '1;33': 'Yellow',
                '1;34': 'LightBlue',
                '1;35': 'MediumPurple',
                '1;36': 'LightCyan',
                '1;37': 'White'}

class IterableIPShell:
  def __init__(self,argv=None,user_ns=None,user_global_ns=None, 
               cin=None, cout=None,cerr=None, input_func=None):
    if input_func:
      IPython.iplib.raw_input_original = input_func
    if cin:
      IPython.Shell.Term.cin = cin
    if cout:
      IPython.Shell.Term.cout = cout
    if cerr:
      IPython.Shell.Term.cerr = cerr

    if argv is None:
      argv=[]

    # This is to get rid of the blockage that occurs during 
    # IPython.Shell.InteractiveShell.user_setup()
    IPython.iplib.raw_input = lambda x: None

    self.term = IPython.genutils.IOTerm(cin=cin, cout=cout, cerr=cerr)
    os.environ['TERM'] = 'dumb'
    excepthook = sys.excepthook 
    self.IP = IPython.Shell.make_IPython(argv,user_ns=user_ns,
                                         user_global_ns=user_global_ns,
                                         embedded=True,
                                         shell_class=IPython.Shell.InteractiveShell)
    self.IP.system = lambda cmd: self.shell(self.IP.var_expand(cmd),
                                            header='IPython system call: ',
                                            verbose=self.IP.rc.system_verbose)
    sys.excepthook = excepthook
    self.iter_more = 0
    self.history_level = 0
    self.complete_sep =  re.compile('[\s\{\}\[\]\(\)]')

  def execute(self):
    self.history_level = 0
    orig_stdout = sys.stdout
    sys.stdout = IPython.Shell.Term.cout
    try:
      line = self.IP.raw_input(None, self.iter_more)
      if self.IP.autoindent:
        self.IP.readline_startup_hook(None)
    except KeyboardInterrupt:
      self.IP.write('\nKeyboardInterrupt\n')
      self.IP.resetbuffer()
      # keep cache in sync with the prompt counter:
      self.IP.outputcache.prompt_count -= 1
        
      if self.IP.autoindent:
        self.IP.indent_current_nsp = 0
      self.iter_more = 0
    except:
      self.IP.showtraceback()
    else:
      self.iter_more = self.IP.push(line)
      if (self.IP.SyntaxTB.last_syntax_error and
          self.IP.rc.autoedit_syntax):
        self.IP.edit_syntax_error()
    if self.iter_more:
      self.prompt = str(self.IP.outputcache.prompt2).strip()
      if self.IP.autoindent:
        self.IP.readline_startup_hook(self.IP.pre_readline)
    else:
      self.prompt = str(self.IP.outputcache.prompt1).strip()
    sys.stdout = orig_stdout

  def historyBack(self):
    self.history_level -= 1
    return self._getHistory()
  
  def historyForward(self):
    self.history_level += 1
    return self._getHistory()
  
  def _getHistory(self):
    try:
      rv = self.IP.user_ns['In'][self.history_level].strip('\n')
    except IndexError:
      self.history_level = 0
      rv = ''
    return rv

  def updateNamespace(self, ns_dict):
    self.IP.user_ns.update(ns_dict)

  def complete(self, line):
    split_line = self.complete_sep.split(line)
    possibilities = self.IP.complete(split_line[-1])
    if possibilities:
      common_prefix = reduce(self._commonPrefix, possibilities)
      completed = line[:-len(split_line[-1])]+common_prefix
    else:
      completed = line
    return completed, possibilities
  
  def _commonPrefix(self, str1, str2):
    for i in range(len(str1)):
      if not str2.startswith(str1[:i+1]):
        return str1[:i]
    return str1

  def shell(self, cmd,verbose=0,debug=0,header=''):
    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):
  def __init__(self):
    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 ansi_colors:
      self.text_buffer.create_tag(code, 
                                  foreground=ansi_colors[code], 
                                  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)
    self.last_cursor_pos = 0
    
  def write(self, text, editable=False):
    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], 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):
    self.write(prompt)
    self.text_buffer.move_mark(self.line_start,self.text_buffer.get_end_iter())

  def changeLine(self, text):
    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):
    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):
    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())

  def _onKeypress(self, obj, event):
    if not event.string:
      return
    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 start_iter.compare(insert_iter) <= 0 and \
          start_iter.compare(selection_iter) <= 0:
      return
    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)             
    

class IPythonView(ConsoleView, IterableIPShell):
  def __init__(self):
    ConsoleView.__init__(self)
    self.cout = StringIO()
    IterableIPShell.__init__(self, cout=self.cout,cerr=self.cout, 
                             input_func=self.raw_input)
    self.connect('key_press_event', self.keyPress)
    self.execute()
    self.cout.truncate(0)
    self.showPrompt(self.prompt)
    self.interrupt = False

  def raw_input(self, prompt=''):
    if self.interrupt:
      self.interrupt = False
      raise KeyboardInterrupt
    return self.getCurrentLine()

  def keyPress(self, widget, event):
    if event.state & gtk.gdk.CONTROL_MASK and event.keyval == 99:
      self.interrupt = True
      self._processLine()
      return True
    elif event.keyval == gtk.keysyms.Return:
      self._processLine()
      return True
    elif event.keyval == gtk.keysyms.Up:
      self.changeLine(self.historyBack())
      return True
    elif event.keyval == gtk.keysyms.Down:
      self.changeLine(self.historyForward())
      return True
    elif event.keyval == gtk.keysyms.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.prompt)
      self.changeLine(completed or slice)
      return True

  def _processLine(self):
    self.history_pos = 0
    self.execute()
    rv = self.cout.getvalue()
    if rv: rv = rv.strip('\n')
    self.showReturned(rv)
    self.cout.truncate(0)
    
1	"""
2	Backend to the console plugin.
3
4	@author: Eitan Isaacson
5	@organization: IBM Corporation
6	@copyright: Copyright (c) 2007 IBM Corporation
7	@license: BSD
8
9	All rights reserved. This program and the accompanying materials are made
10	available under the terms of the BSD which accompanies this distribution, and
11	is available at U{http://www.opensource.org/licenses/bsd-license.php}
12	"""
13	# this file is a modified version of source code from the Accerciser project
14	# http://live.gnome.org/accerciser
15
16	import gtk
17	import re
18	import sys
19	import os
20	import pango
21	from StringIO import StringIO
22
23	try:
24	import IPython
25	except Exception,e:
26	raise "Error importing IPython (%s)" % str(e)
27
28	ansi_colors = {'0;30': 'Black',
29	'0;31': 'Red',
30	'0;32': 'Green',
31	'0;33': 'Brown',
32	'0;34': 'Blue',
33	'0;35': 'Purple',
34	'0;36': 'Cyan',
35	'0;37': 'LightGray',
36	'1;30': 'DarkGray',
37	'1;31': 'DarkRed',
38	'1;32': 'SeaGreen',
39	'1;33': 'Yellow',
40	'1;34': 'LightBlue',
41	'1;35': 'MediumPurple',
42	'1;36': 'LightCyan',
43	'1;37': 'White'}
44
45	class IterableIPShell:
46	def __init__(self,argv=None,user_ns=None,user_global_ns=None,
47	cin=None, cout=None,cerr=None, input_func=None):
48	if input_func:
49	IPython.iplib.raw_input_original = input_func
50	if cin:
51	IPython.Shell.Term.cin = cin
52	if cout:
53	IPython.Shell.Term.cout = cout
54	if cerr:
55	IPython.Shell.Term.cerr = cerr
56
57	if argv is None:
58	argv=[]
59
60	# This is to get rid of the blockage that occurs during
61	# IPython.Shell.InteractiveShell.user_setup()
62	IPython.iplib.raw_input = lambda x: None
63
64	self.term = IPython.genutils.IOTerm(cin=cin, cout=cout, cerr=cerr)
65	os.environ['TERM'] = 'dumb'
66	excepthook = sys.excepthook
67	self.IP = IPython.Shell.make_IPython(argv,user_ns=user_ns,
68	user_global_ns=user_global_ns,
69	embedded=True,
70	shell_class=IPython.Shell.InteractiveShell)
71	self.IP.system = lambda cmd: self.shell(self.IP.var_expand(cmd),
72	header='IPython system call: ',
73	verbose=self.IP.rc.system_verbose)
74	sys.excepthook = excepthook
75	self.iter_more = 0
76	self.history_level = 0
77	self.complete_sep = re.compile('[\s\{\}\[\]\(\)]')
78
79	def execute(self):
80	self.history_level = 0
81	orig_stdout = sys.stdout
82	sys.stdout = IPython.Shell.Term.cout
83	try:
84	line = self.IP.raw_input(None, self.iter_more)
85	if self.IP.autoindent:
86	self.IP.readline_startup_hook(None)
87	except KeyboardInterrupt:
88	self.IP.write('\nKeyboardInterrupt\n')
89	self.IP.resetbuffer()
90	# keep cache in sync with the prompt counter:
91	self.IP.outputcache.prompt_count -= 1
92
93	if self.IP.autoindent:
94	self.IP.indent_current_nsp = 0
95	self.iter_more = 0
96	except:
97	self.IP.showtraceback()
98	else:
99	self.iter_more = self.IP.push(line)
100	if (self.IP.SyntaxTB.last_syntax_error and
101	self.IP.rc.autoedit_syntax):
102	self.IP.edit_syntax_error()
103	if self.iter_more:
104	self.prompt = str(self.IP.outputcache.prompt2).strip()
105	if self.IP.autoindent:
106	self.IP.readline_startup_hook(self.IP.pre_readline)
107	else:
108	self.prompt = str(self.IP.outputcache.prompt1).strip()
109	sys.stdout = orig_stdout
110
111	def historyBack(self):
112	self.history_level -= 1
113	return self._getHistory()
114
115	def historyForward(self):
116	self.history_level += 1
117	return self._getHistory()
118
119	def _getHistory(self):
120	try:
121	rv = self.IP.user_ns['In'][self.history_level].strip('\n')
122	except IndexError:
123	self.history_level = 0
124	rv = ''
125	return rv
126
127	def updateNamespace(self, ns_dict):
128	self.IP.user_ns.update(ns_dict)
129
130	def complete(self, line):
131	split_line = self.complete_sep.split(line)
132	possibilities = self.IP.complete(split_line[-1])
133	if possibilities:
134	common_prefix = reduce(self._commonPrefix, possibilities)
135	completed = line[:-len(split_line[-1])]+common_prefix
136	else:
137	completed = line
138	return completed, possibilities
139
140	def _commonPrefix(self, str1, str2):
141	for i in range(len(str1)):
142	if not str2.startswith(str1[:i+1]):
143	return str1[:i]
144	return str1
145
146	def shell(self, cmd,verbose=0,debug=0,header=''):
147	stat = 0
148	if verbose or debug: print header+cmd
149	# flush stdout so we don't mangle python's buffering
150	if not debug:
151	input, output = os.popen4(cmd)
152	print output.read()
153	output.close()
154	input.close()
155
156	class ConsoleView(gtk.TextView):
157	def __init__(self):
158	gtk.TextView.__init__(self)
159	self.modify_font(pango.FontDescription('Mono'))
160	self.set_cursor_visible(True)
161	self.text_buffer = self.get_buffer()
162	self.mark = self.text_buffer.create_mark('scroll_mark',
163	self.text_buffer.get_end_iter(),
164	False)
165	for code in ansi_colors:
166	self.text_buffer.create_tag(code,
167	foreground=ansi_colors[code],
168	weight=700)
169	self.text_buffer.create_tag('0')
170	self.text_buffer.create_tag('notouch', editable=False)
171	self.color_pat = re.compile('\x01?\x1b\[(.*?)m\x02?')
172	self.line_start = \
173	self.text_buffer.create_mark('line_start',
174	self.text_buffer.get_end_iter(), True
175	)
176	self.connect('key-press-event', self._onKeypress)
177	self.last_cursor_pos = 0
178
179	def write(self, text, editable=False):
180	segments = self.color_pat.split(text)
181	segment = segments.pop(0)
182	start_mark = self.text_buffer.create_mark(None,
183	self.text_buffer.get_end_iter(),
184	True)
185	self.text_buffer.insert(self.text_buffer.get_end_iter(), segment)
186
187	if segments:
188	ansi_tags = self.color_pat.findall(text)
189	for tag in ansi_tags:
190	i = segments.index(tag)
191	self.text_buffer.insert_with_tags_by_name(self.text_buffer.get_end_iter(),
192	segments[i+1], tag)
193	segments.pop(i)
194	if not editable:
195	self.text_buffer.apply_tag_by_name('notouch',
196	self.text_buffer.get_iter_at_mark(start_mark),
197	self.text_buffer.get_end_iter())
198	self.text_buffer.delete_mark(start_mark)
199	self.scroll_mark_onscreen(self.mark)
200
201	def showPrompt(self, prompt):
202	self.write(prompt)
203	self.text_buffer.move_mark(self.line_start,self.text_buffer.get_end_iter())
204
205	def changeLine(self, text):
206	iter = self.text_buffer.get_iter_at_mark(self.line_start)
207	iter.forward_to_line_end()
208	self.text_buffer.delete(self.text_buffer.get_iter_at_mark(self.line_start), iter)
209	self.write(text, True)
210
211	def getCurrentLine(self):
212	rv = self.text_buffer.get_slice(self.text_buffer.get_iter_at_mark(self.line_start),
213	self.text_buffer.get_end_iter(), False)
214	return rv
215
216	def showReturned(self, text):
217	iter = self.text_buffer.get_iter_at_mark(self.line_start)
218	iter.forward_to_line_end()
219	self.text_buffer.apply_tag_by_name('notouch',
220	self.text_buffer.get_iter_at_mark(self.line_start),
221	iter)
222	self.write('\n'+text)
223	if text:
224	self.write('\n')
225	self.showPrompt(self.prompt)
226	self.text_buffer.move_mark(self.line_start,self.text_buffer.get_end_iter())
227	self.text_buffer.place_cursor(self.text_buffer.get_end_iter())
228
229	def _onKeypress(self, obj, event):
230	if not event.string:
231	return
232	insert_mark = self.text_buffer.get_insert()
233	insert_iter = self.text_buffer.get_iter_at_mark(insert_mark)
234	selection_mark = self.text_buffer.get_selection_bound()
235	selection_iter = self.text_buffer.get_iter_at_mark(selection_mark)
236	start_iter = self.text_buffer.get_iter_at_mark(self.line_start)
237	if start_iter.compare(insert_iter) <= 0 and \
238	start_iter.compare(selection_iter) <= 0:
239	return
240	elif start_iter.compare(insert_iter) > 0 and \
241	start_iter.compare(selection_iter) > 0:
242	self.text_buffer.place_cursor(start_iter)
243	elif insert_iter.compare(selection_iter) < 0:
244	self.text_buffer.move_mark(insert_mark, start_iter)
245	elif insert_iter.compare(selection_iter) > 0:
246	self.text_buffer.move_mark(selection_mark, start_iter)
247
248
249	class IPythonView(ConsoleView, IterableIPShell):
250	def __init__(self):
251	ConsoleView.__init__(self)
252	self.cout = StringIO()
253	IterableIPShell.__init__(self, cout=self.cout,cerr=self.cout,
254	input_func=self.raw_input)
255	self.connect('key_press_event', self.keyPress)
256	self.execute()
257	self.cout.truncate(0)
258	self.showPrompt(self.prompt)
259	self.interrupt = False
260
261	def raw_input(self, prompt=''):
262	if self.interrupt:
263	self.interrupt = False
264	raise KeyboardInterrupt
265	return self.getCurrentLine()
266
267	def keyPress(self, widget, event):
268	if event.state & gtk.gdk.CONTROL_MASK and event.keyval == 99:
269	self.interrupt = True
270	self._processLine()
271	return True
272	elif event.keyval == gtk.keysyms.Return:
273	self._processLine()
274	return True
275	elif event.keyval == gtk.keysyms.Up:
276	self.changeLine(self.historyBack())
277	return True
278	elif event.keyval == gtk.keysyms.Down:
279	self.changeLine(self.historyForward())
280	return True
281	elif event.keyval == gtk.keysyms.Tab:
282	if not self.getCurrentLine().strip():
283	return False
284	completed, possibilities = self.complete(self.getCurrentLine())
285	if len(possibilities) > 1:
286	slice = self.getCurrentLine()
287	self.write('\n')
288	for symbol in possibilities:
289	self.write(symbol+'\n')
290	self.showPrompt(self.prompt)
291	self.changeLine(completed or slice)
292	return True
293
294	def _processLine(self):
295	self.history_pos = 0
296	self.execute()
297	rv = self.cout.getvalue()
298	if rv: rv = rv.strip('\n')
299	self.showReturned(rv)
300	self.cout.truncate(0)
301