"""Given a stream of tokens and a screen width plus an optional

lines = 1

pos = cursor

for (token, value) in tokens:

if token is Token.Text and value == '\n':

lines += 1

else:

pos += len(value)

lines += pos // width

pos %= width

@functools.wraps(func)

def newfunc(self, *args, **kwargs):

dest = self.get_dest()

if self is dest:

return func(self, *args, **kwargs)

else:

return getattr(self.get_dest(), newfunc.__name__)(*args, **kwargs)

"""Provide a fake file object which calls functions on the interface

def __init__(self, interface, get_dest):

self.encoding = getpreferredencoding()

self.interface = interface

self.get_dest = get_dest

@forward_if_not_current

def write(self, s):

self.interface.write(s)

@forward_if_not_current

def writelines(self, l):

for s in l:

self.write(s)

def isatty(self):

# some third party (amongst them mercurial) depend on this

return True

def flush(self):

"""Provide a fake stdin type for things like raw_input() etc."""

def __init__(self, interface):

"""Take the curses Repl on init and assume it provides a get_key method

self.encoding = getpreferredencoding()

self.interface = interface

self.buffer = list()

def __iter__(self):

return iter(self.readlines())

def flush(self):

"""Flush the internal buffer. This is a no-op. Flushing stdin

def write(self, value):

# XXX IPython expects sys.stdin.write to exist, there will no doubt be

# others, so here's a hack to keep them happy

raise IOError(errno.EBADF, "sys.stdin is read-only")

def isatty(self):

return True

def readline(self, size=-1):

"""I can't think of any reason why anything other than readline would

if not size:

return ''

elif self.buffer:

buffer = self.buffer.pop(0)

else:

buffer = ''

curses.raw(True)

try:

while not buffer.endswith(('\n', '\r')):

key = self.interface.get_key()

if key in [curses.erasechar(), 'KEY_BACKSPACE']:

y, x = self.interface.scr.getyx()

if buffer:

self.interface.scr.delch(y, x - 1)

buffer = buffer[:-1]

continue

elif key == chr(4) and not buffer:

# C-d

return ''

elif (key not in ('\n', '\r') and

(len(key) > 1 or unicodedata.category(key) == 'Cc')):

continue

sys.stdout.write(key)

# Include the \n in the buffer - raw_input() seems to deal with trailing

# linebreaks and will break if it gets an empty string.

buffer += key

finally:

curses.raw(False)

if size > 0:

rest = buffer[size:]

if rest:

self.buffer.append(rest)

buffer = buffer[:size]

if py3:

return buffer

else:

return buffer.encode(getpreferredencoding())

def read(self, size=None):

if size == 0:

return ''

data = list()

while size is None or size > 0:

line = self.readline(size or -1)

if not line:

break

if size is not None:

size -= len(line)

data.append(line)

return ''.join(data)

def readlines(self, size=-1):

"""Init all the colours in curses and bang them into a dictionary"""

# blacK, Red, Green, Yellow, Blue, Magenta, Cyan, White, Default:

c = {

'k': 0,

'r': 1,

'g': 2,

'y': 3,

'b': 4,

'm': 5,

'c': 6,

'w': 7,

'd': -1,

}

if platform.system() == 'Windows':

c = dict(list(c.items()) +

[

('K', 8),

('R', 9),

('G', 10),

('Y', 11),

('B', 12),

('M', 13),

('C', 14),

('W', 15),

]

)

for i in range(63):

if i > 7:

j = i // 8

else:

j = c[config.color_scheme['background']]

curses.init_pair(i + 1, i % 8, j)

def __init__(self, config, statusbar=None):

super(CLIInteraction, self).__init__(config, statusbar)

def confirm(self, q):

"""Ask for yes or no and return boolean"""

try:

reply = self.statusbar.prompt(q)

except ValueError:

return False

return reply.lower() in (_('y'), _('yes'))

def notify(self, s, n=10, wait_for_keypress=False):

return self.statusbar.message(s, n)

def file_prompt(self, s):

def __init__(self, scr, interp, statusbar, config, idle=None):

super(CLIRepl, self).__init__(interp, config)

self.interp.writetb = self.writetb

self.scr = scr

self.stdout_hist = '' # native str (bytes in Py2, unicode in Py3)

self.list_win = newwin(get_colpair(config, 'background'), 1, 1, 1, 1)

self.cpos = 0

self.do_exit = False

self.exit_value = ()

self.f_string = ''

self.idle = idle

self.in_hist = False

self.paste_mode = False

self.last_key_press = time.time()

self.s = ''

self.statusbar = statusbar

self.formatter = BPythonFormatter(config.color_scheme)

self.interact = CLIInteraction(self.config, statusbar=self.statusbar)

if config.cli_suggestion_width <= 0 or config.cli_suggestion_width > 1:

config.cli_suggestion_width = 0.8

def _get_cursor_offset(self):

return len(self.s) - self.cpos

def _set_cursor_offset(self, offset):

self.cpos = len(self.s) - offset

cursor_offset = property(_get_cursor_offset, _set_cursor_offset, None,

"The cursor offset from the beginning of the line")

def addstr(self, s):

"""Add a string to the current input line and figure out

self.rl_history.reset()

if not self.cpos:

self.s += s

else:

l = len(self.s)

self.s = self.s[:l - self.cpos] + s + self.s[l - self.cpos:]

self.complete()

def atbol(self):

"""Return True or False accordingly if the cursor is at the beginning

return not self.s.lstrip()

def bs(self, delete_tabs=True):

"""Process a backspace"""

self.rl_history.reset()

y, x = self.scr.getyx()

if not self.s:

return

if x == self.ix and y == self.iy:

return

n = 1

self.clear_wrapped_lines()

if not self.cpos:

# I know the nested if blocks look nasty. :(

if self.atbol() and delete_tabs:

n = len(self.s) % self.config.tab_length

if not n:

n = self.config.tab_length

self.s = self.s[:-n]

else:

self.s = self.s[:-self.cpos - 1] + self.s[-self.cpos:]

self.print_line(self.s, clr=True)

return n

def bs_word(self):

self.rl_history.reset()

pos = len(self.s) - self.cpos - 1

deleted = []

# First we delete any space to the left of the cursor.

while pos >= 0 and self.s[pos] == ' ':

deleted.append(self.s[pos])

pos -= self.bs()

# Then we delete a full word.

while pos >= 0 and self.s[pos] != ' ':

deleted.append(self.s[pos])

pos -= self.bs()

return ''.join(reversed(deleted))

def check(self):

"""Check if paste mode should still be active and, if not, deactivate

if (self.paste_mode

and time.time() - self.last_key_press > self.config.paste_time):

self.paste_mode = False

self.print_line(self.s)

def clear_current_line(self):

"""Called when a SyntaxError occurred in the interpreter. It is

repl.Repl.clear_current_line(self)

self.s = ''

def clear_wrapped_lines(self):

"""Clear the wrapped lines of the current input."""

# curses does not handle this on its own. Sad.

height, width = self.scr.getmaxyx()

max_y = min(self.iy + (self.ix + len(self.s)) // width + 1, height)

for y in range(self.iy + 1, max_y):

self.scr.move(y, 0)

self.scr.clrtoeol()

def complete(self, tab=False):

"""Get Autocomplete list and window.

if self.paste_mode:

self.scr.touchwin() # TODO necessary?

return

list_win_visible = repl.Repl.complete(self, tab)

if list_win_visible:

try:

self.show_list(self.matches_iter.matches, self.arg_pos,

topline=self.funcprops,

formatter=self.matches_iter.completer.format)

except curses.error:

# XXX: This is a massive hack, it will go away when I get

# cusswords into a good enough state that we can start

# using it.

self.list_win.border()

self.list_win.refresh()

list_win_visible = False

if not list_win_visible:

self.scr.redrawwin()

self.scr.refresh()

def clrtobol(self):

"""Clear from cursor to beginning of line; usual C-u behaviour"""

self.clear_wrapped_lines()

if not self.cpos:

self.s = ''

else:

self.s = self.s[-self.cpos:]

self.print_line(self.s, clr=True)

self.scr.redrawwin()

self.scr.refresh()

def _get_current_line(self):

return self.s

def _set_current_line(self, line):

self.s = line

current_line = property(_get_current_line, _set_current_line, None,

"The characters of the current line")

def cut_to_buffer(self):

"""Clear from cursor to end of line, placing into cut buffer"""

self.cut_buffer = self.s[-self.cpos:]

self.s = self.s[:-self.cpos]

self.cpos = 0

self.print_line(self.s, clr=True)

self.scr.redrawwin()

self.scr.refresh()

def delete(self):

"""Process a del"""

if not self.s:

return

if self.mvc(-1):

self.bs(False)

def echo(self, s, redraw=True):

"""Parse and echo a formatted string with appropriate attributes. It

if not py3 and isinstance(s, unicode):

s = s.encode(getpreferredencoding())

a = get_colpair(self.config, 'output')

if '\x01' in s:

rx = re.search('\x01([A-Za-z])([A-Za-z]?)', s)

if rx:

fg = rx.groups()[0]

bg = rx.groups()[1]

col_num = self._C[fg.lower()]

if bg and bg != 'I':

col_num *= self._C[bg.lower()]

a = curses.color_pair(int(col_num) + 1)

if bg == 'I':

a = a | curses.A_REVERSE

s = re.sub('\x01[A-Za-z][A-Za-z]?', '', s)

if fg.isupper():

a = a | curses.A_BOLD

s = s.replace('\x03', '')

s = s.replace('\x01', '')

# Replace NUL bytes, as addstr raises an exception otherwise

s = s.replace('\0', '')

# Replace \r\n bytes, as addstr remove the current line otherwise

s = s.replace('\r\n', '\n')

self.scr.addstr(s, a)

if redraw and not self.evaluating:

self.scr.refresh()

def end(self, refresh=True):

self.cpos = 0

h, w = gethw()

y, x = divmod(len(self.s) + self.ix, w)

y += self.iy

self.scr.move(y, x)

if refresh:

self.scr.refresh()

return True

def hbegin(self):

"""Replace the active line with first line in history and

self.cpos = 0

self.clear_wrapped_lines()

self.rl_history.enter(self.s)

self.s = self.rl_history.first()

self.print_line(self.s, clr=True)

def hend(self):

"""Same as hbegin() but, well, forward"""

self.cpos = 0

self.clear_wrapped_lines()

self.rl_history.enter(self.s)

self.s = self.rl_history.last()

self.print_line(self.s, clr=True)

def back(self):

"""Replace the active line with previous line in history and

self.cpos = 0

self.clear_wrapped_lines()

self.rl_history.enter(self.s)

self.s = self.rl_history.back()

self.print_line(self.s, clr=True)

def fwd(self):

"""Same as back() but, well, forward"""

self.cpos = 0

self.clear_wrapped_lines()

self.rl_history.enter(self.s)

self.s = self.rl_history.forward()

self.print_line(self.s, clr=True)

def search(self):

"""Search with the partial matches from the history object."""

self.cpo = 0

self.clear_wrapped_lines()

self.rl_history.enter(self.s)

self.s = self.rl_history.back(start=False, search=True)

self.print_line(self.s, clr=True)

def get_key(self):

key = ''

while True:

try:

key += self.scr.getkey()

if py3:

# Seems like we get a in the locale's encoding

# encoded string in Python 3 as well, but of

# type str instead of bytes, hence convert it to

# bytes first and decode then

key = key.encode('latin-1').decode(getpreferredencoding())

else:

key = key.decode(getpreferredencoding())

self.scr.nodelay(False)

except UnicodeDecodeError:

# Yes, that actually kind of sucks, but I don't see another way to get

# input right

self.scr.nodelay(True)

except curses.error:

# I'm quite annoyed with the ambiguity of this exception handler. I previously

# caught "curses.error, x" and accessed x.message and checked that it was "no

# input", which seemed a crappy way of doing it. But then I ran it on a

# different computer and the exception seems to have entirely different

# attributes. So let's hope getkey() doesn't raise any other crazy curses

# exceptions. :)

self.scr.nodelay(False)

# XXX What to do here? Raise an exception?

if key:

return key

else:

if key != '\x00':

t = time.time()

self.paste_mode = (

t - self.last_key_press <= self.config.paste_time

)

self.last_key_press = t

return key

else:

key = ''

finally:

if self.idle:

self.idle(self)

def get_line(self):

"""Get a line of text and return it

self.s = ''

self.rl_history.reset()

self.iy, self.ix = self.scr.getyx()

if not self.paste_mode:

for _ in range(self.next_indentation()):

self.p_key('\t')

self.cpos = 0

while True:

key = self.get_key()

if self.p_key(key) is None:

if self.config.cli_trim_prompts and self.s.startswith(">>> "):

self.s = self.s[4:]

return self.s

def home(self, refresh=True):

self.scr.move(self.iy, self.ix)

self.cpos = len(self.s)

if refresh:

self.scr.refresh()

return True

def lf(self):

"""Process a linefeed character; it only needs to check the

if self.cpos:

for _ in range(self.cpos):

self.mvc(-1)

# Reprint the line (as there was maybe a highlighted paren in it)

self.print_line(self.s, newline=True)

self.echo("\n")

def mkargspec(self, topline, in_arg, down):

"""This figures out what to do with the argspec and puts it nicely into

r = 3

fn = topline.func

args = topline.argspec.args

kwargs = topline.argspec.defaults

_args = topline.argspec.varargs

_kwargs = topline.argspec.varkwargs

is_bound_method = topline.is_bound_method

if py3:

kwonly = topline.argspec.kwonly

kwonly_defaults = topline.argspec.kwonly_defaults or dict()

max_w = int(self.scr.getmaxyx()[1] * 0.6)

self.list_win.erase()

self.list_win.resize(3, max_w)

h, w = self.list_win.getmaxyx()

self.list_win.addstr('\n ')

self.list_win.addstr(fn,

get_colpair(self.config, 'name') | curses.A_BOLD)

self.list_win.addstr(': (', get_colpair(self.config, 'name'))

maxh = self.scr.getmaxyx()[0]

if is_bound_method and isinstance(in_arg, int):

in_arg += 1

punctuation_colpair = get_colpair(self.config, 'punctuation')

for k, i in enumerate(args):

y, x = self.list_win.getyx()

ln = len(str(i))

kw = None

if kwargs and k + 1 > len(args) - len(kwargs):

kw = repr(kwargs[k - (len(args) - len(kwargs))])

ln += len(kw) + 1

if ln + x >= w:

ty = self.list_win.getbegyx()[0]

if not down and ty > 0:

h += 1

self.list_win.mvwin(ty - 1, 1)

self.list_win.resize(h, w)

elif down and h + r < maxh - ty:

h += 1

self.list_win.resize(h, w)

else:

break

r += 1

self.list_win.addstr('\n\t')

if str(i) == 'self' and k == 0:

color = get_colpair(self.config, 'name')

else:

color = get_colpair(self.config, 'token')

if k == in_arg or i == in_arg:

color |= curses.A_BOLD

if not py3:

# See issue #138: We need to format tuple unpacking correctly

# We use the undocumented function inspection.strseq() for

# that. Fortunately, that madness is gone in Python 3.

self.list_win.addstr(inspect.strseq(i, str), color)

else:

self.list_win.addstr(str(i), color)

if kw is not None:

self.list_win.addstr('=', punctuation_colpair)

self.list_win.addstr(kw, get_colpair(self.config, 'token'))

if k != len(args) - 1:

self.list_win.addstr(', ', punctuation_colpair)

if _args:

if args:

self.list_win.addstr(', ', punctuation_colpair)

self.list_win.addstr('*%s' % (_args, ),

get_colpair(self.config, 'token'))

if py3 and kwonly:

if not _args:

if args:

self.list_win.addstr(', ', punctuation_colpair)

self.list_win.addstr('*', punctuation_colpair)

marker = object()

for arg in kwonly:

self.list_win.addstr(', ', punctuation_colpair)

color = get_colpair(self.config, 'token')

if arg == in_arg:

color |= curses.A_BOLD

self.list_win.addstr(arg, color)

default = kwonly_defaults.get(arg, marker)

if default is not marker:

self.list_win.addstr('=', punctuation_colpair)

self.list_win.addstr(repr(default),

get_colpair(self.config, 'token'))

if _kwargs:

if args or _args or (py3 and kwonly):

self.list_win.addstr(', ', punctuation_colpair)

self.list_win.addstr('**%s' % (_kwargs, ),

get_colpair(self.config, 'token'))

self.list_win.addstr(')', punctuation_colpair)

return r

def mvc(self, i, refresh=True):

"""This method moves the cursor relatively from the current

y, x = self.scr.getyx()

if self.cpos == 0 and i < 0:

return False

if x == self.ix and y == self.iy and i >= 1:

return False

h, w = gethw()

if x - i < 0:

y -= 1

x = w

if x - i >= w:

y += 1

x = 0 + i

self.cpos += i

self.scr.move(y, x - i)

if refresh:

self.scr.refresh()

return True

def p_key(self, key):

"""Process a keypress"""

if key is None:

return ''

config = self.config

if platform.system() == 'Windows':

C_BACK = chr(127)

BACKSP = chr(8)

else:

C_BACK = chr(8)

BACKSP = chr(127)

if key == C_BACK: # C-Backspace (on my computer anyway!)

self.clrtobol()

key = '\n'

# Don't return; let it get handled

if key == chr(27): # Escape Key

return ''

if key in (BACKSP, 'KEY_BACKSPACE'):

self.bs()

self.complete()

return ''

elif key in key_dispatch[config.delete_key] and not self.s:

# Delete on empty line exits

self.do_exit = True

return None

elif key in ('KEY_DC', ) + key_dispatch[config.delete_key]:

self.delete()

self.complete()

# Redraw (as there might have been highlighted parens)

self.print_line(self.s)

return ''

elif key in key_dispatch[config.undo_key]: # C-r

n = self.prompt_undo()

if n > 0:

self.undo(n=n)

return ''

elif key in key_dispatch[config.search_key]:

self.search()

return ''

elif key in ('KEY_UP', ) + key_dispatch[config.up_one_line_key]:

# Cursor Up/C-p

self.back()

return ''

elif key in ('KEY_DOWN', ) + key_dispatch[config.down_one_line_key]:

# Cursor Down/C-n

self.fwd()

return ''

elif key in ("KEY_LEFT", ' ^B', chr(2)): # Cursor Left or ^B

self.mvc(1)

# Redraw (as there might have been highlighted parens)

self.print_line(self.s)

elif key in ("KEY_RIGHT", '^F', chr(6)): # Cursor Right or ^F

self.mvc(-1)

# Redraw (as there might have been highlighted parens)

self.print_line(self.s)

elif key in ("KEY_HOME", '^A', chr(1)): # home or ^A

self.home()

# Redraw (as there might have been highlighted parens)

self.print_line(self.s)

elif key in ("KEY_END", '^E', chr(5)): # end or ^E

self.end()

# Redraw (as there might have been highlighted parens)

self.print_line(self.s)

elif key in ("KEY_NPAGE", '\T'): # page_down or \T

self.hend()

self.print_line(self.s)

elif key in ("KEY_PPAGE", '\S'): # page_up or \S

self.hbegin()

self.print_line(self.s)

elif key in key_dispatch[config.cut_to_buffer_key]: # cut to buffer

self.cut_to_buffer()

return ''

elif key in key_dispatch[config.yank_from_buffer_key]:

# yank from buffer

self.yank_from_buffer()

return ''

elif key in key_dispatch[config.clear_word_key]:

self.cut_buffer = self.bs_word()

self.complete()

return ''

elif key in key_dispatch[config.clear_line_key]:

self.clrtobol()

return ''

elif key in key_dispatch[config.clear_screen_key]:

self.s_hist = [self.s_hist[-1]]

self.highlighted_paren = None

self.redraw()

return ''

elif key in key_dispatch[config.exit_key]:

if not self.s:

self.do_exit = True

return None

else:

return ''

elif key in key_dispatch[config.save_key]:

self.write2file()

return ''

elif key in key_dispatch[config.pastebin_key]:

self.pastebin()

return ''

elif key in key_dispatch[config.copy_clipboard_key]:

self.copy2clipboard()

return ''

elif key in key_dispatch[config.last_output_key]:

page(self.stdout_hist[self.prev_block_finished:-4])

return ''

elif key in key_dispatch[config.show_source_key]:

try:

source = self.get_source_of_current_name()

except repl.SourceNotFound as e:

self.statusbar.message('%s' % (e, ))

else:

if config.highlight_show_source:

source = format(PythonLexer().get_tokens(source),

TerminalFormatter())

page(source)

return ''

elif key in ('\n', '\r', 'PADENTER'):

self.lf()

return None

elif key == '\t':

return self.tab()

elif key == 'KEY_BTAB':

return self.tab(back=True)

elif key in key_dispatch[config.suspend_key]:

if platform.system() != 'Windows':

self.suspend()

return ''