def __init__(self):

self.reset_timer()

self.time = time.monotonic if hasattr(time, 'monotonic') else time.time

def __enter__(self):

self.start = self.time()

def __exit__(self, ty, val, tb):

self.last_command = self.time() - self.start

self.running_time += self.last_command

return False

def reset_timer(self):

self.running_time = 0.0

self.last_command = 0.0

def estimate(self):

def __init__(self, locals=None, encoding=None):

"""The syntaxerror callback can be set at any time and will be called

self.encoding = encoding or sys.getdefaultencoding()

self.syntaxerror_callback = None

# Unfortunately code.InteractiveInterpreter is a classic class, so no

# super()

code.InteractiveInterpreter.__init__(self, locals)

self.timer = RuntimeTimer()

def reset_running_time(self):

self.running_time = 0

def runsource(self, source, filename='<input>', symbol='single',

encode=True):

"""Execute Python code.

if not py3 and encode:

source = u'# coding: %s\n%s' % (self.encoding, source)

source = source.encode(self.encoding)

with self.timer:

return code.InteractiveInterpreter.runsource(self, source,

filename, symbol)

def showsyntaxerror(self, filename=None):

"""Override the regular handler, the code's copied and pasted from

if self.syntaxerror_callback is not None:

self.syntaxerror_callback()

type, value, sys.last_traceback = sys.exc_info()

sys.last_type = type

sys.last_value = value

if filename and type is SyntaxError:

# Work hard to stuff the correct filename in the exception

try:

msg, (dummy_filename, lineno, offset, line) = value.args

except:

# Not the format we expect; leave it alone

pass

else:

# Stuff in the right filename and right lineno

if not py3:

lineno -= 1

value = SyntaxError(msg, (filename, lineno, offset, line))

sys.last_value = value

list = traceback.format_exception_only(type, value)

self.writetb(list)

def showtraceback(self):

"""This needs to override the default traceback thing

try:

t, v, tb = sys.exc_info()

sys.last_type = t

sys.last_value = v

sys.last_traceback = tb

tblist = traceback.extract_tb(tb)

del tblist[:1]

# Set the right lineno (encoding header adds an extra line)

if not py3:

for i, (fname, lineno, module, something) in enumerate(tblist):

if fname == '<input>':

tblist[i] = (fname, lineno - 1, module, something)

l = traceback.format_list(tblist)

if l:

l.insert(0, "Traceback (most recent call last):\n")

l[len(l):] = traceback.format_exception_only(t, v)

finally:

tblist = tb = None

self.writetb(l)

def writetb(self, lines):

"""This outputs the traceback and should be overridden for anything

for line in lines:

"""Stores a list of matches and which one is currently selected if any.

def __init__(self):

# word being replaced in the original line of text

self.current_word = ''

# possible replacements for current_word

self.matches = None

# which word is currently replacing the current word

self.index = -1

# cursor position in the original line

self.orig_cursor_offset = None

# original line (before match replacements)

self.orig_line = None

# class describing the current type of completion

self.completer = None

def __nonzero__(self):

"""MatchesIterator is False when word hasn't been replaced yet"""

return self.index != -1

def __bool__(self):

return self.index != -1

@property

def candidate_selected(self):

"""True when word selected/replaced, False when word hasn't been

return bool(self)

def __iter__(self):

return self

def current(self):

if self.index == -1:

raise ValueError('No current match.')

return self.matches[self.index]

def next(self):

return self.__next__()

def __next__(self):

self.index = (self.index + 1) % len(self.matches)

return self.matches[self.index]

def previous(self):

if self.index <= 0:

self.index = len(self.matches)

self.index -= 1

return self.matches[self.index]

def cur_line(self):

"""Returns a cursor offset and line with the current substitution

return self.substitute(self.current())

def substitute(self, match):

"""Returns a cursor offset and line with match substituted in"""

start, end, word = self.completer.locate(self.orig_cursor_offset,

self.orig_line)

return (start + len(match),

self.orig_line[:start] + match + self.orig_line[end:])

def is_cseq(self):

return bool(

os.path.commonprefix(self.matches)[len(self.current_word):])

def substitute_cseq(self):

"""Returns a new line by substituting a common sequence in, and update

cseq = os.path.commonprefix(self.matches)

new_cursor_offset, new_line = self.substitute(cseq)

if len(self.matches) == 1:

self.clear()

else:

self.update(new_cursor_offset, new_line, self.matches,

self.completer)

if len(self.matches) == 1:

self.clear()

return new_cursor_offset, new_line

def update(self, cursor_offset, current_line, matches, completer):

"""Called to reset the match index and update the word being replaced

if matches is None:

raise ValueError("Matches may not be None.")

self.orig_cursor_offset = cursor_offset

self.orig_line = current_line

self.matches = matches

self.completer = completer

self.index = -1

self.start, self.end, self.current_word = self.completer.locate(

self.orig_cursor_offset, self.orig_line)

def clear(self):

self.matches = []

self.cursor_offset = -1

self.current_line = ''

self.current_word = ''

self.start = None

self.end = None

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

self.config = config

if statusbar:

self.statusbar = statusbar

def confirm(self, s):

raise NotImplementedError

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

raise NotImplementedError

def file_prompt(self, s):

"""Implements the necessary guff for a Python-repl-alike interface

def __init__(self, interp, config):

"""Initialise the repl.

self.config = config

self.cut_buffer = ''

self.buffer = []

self.interp = interp

self.interp.syntaxerror_callback = self.clear_current_line

self.match = False

self.rl_history = History(duplicates=config.hist_duplicates,

hist_size=config.hist_length)

self.s_hist = []

self.history = []

self.evaluating = False

self.matches_iter = MatchesIterator()

self.argspec = None

self.current_func = None

self.highlighted_paren = None

self._C = {}

self.prev_block_finished = 0

self.interact = Interaction(self.config)

# previous pastebin content to prevent duplicate pastes, filled on call

# to repl.pastebin

self.prev_pastebin_content = ''

self.prev_pastebin_url = ''

self.prev_removal_url = ''

# Necessary to fix mercurial.ui.ui expecting sys.stderr to have this

# attribute

self.closed = False

self.clipboard = get_clipboard()

pythonhist = os.path.expanduser(self.config.hist_file)

if os.path.exists(pythonhist):

try:

self.rl_history.load(pythonhist,

getpreferredencoding() or "ascii")

except EnvironmentError:

pass

@property

def ps1(self):

try:

if not py3:

return sys.ps1.decode(getpreferredencoding())

else:

return sys.ps1

except AttributeError:

return u'>>> '

@property

def ps2(self):

try:

if not py3:

return sys.ps2.decode(getpreferredencoding())

else:

return sys.ps2

except AttributeError:

return u'... '

def startup(self):

"""

filename = os.environ.get('PYTHONSTARTUP')

if filename:

encoding = inspection.get_encoding_file(filename)

with io.open(filename, 'rt', encoding=encoding) as f:

source = f.read()

if not py3:

# Python 2.6 and early 2.7.X need bytes.

source = source.encode(encoding)

self.interp.runsource(source, filename, 'exec', encode=False)

def current_string(self, concatenate=False):

"""If the line ends in a string get it, otherwise return ''"""

tokens = self.tokenize(self.current_line)

string_tokens = list(takewhile(token_is_any_of([Token.String,

Token.Text]),

reversed(tokens)))

if not string_tokens:

return ''

opening = string_tokens.pop()[1]

string = list()

for (token, value) in reversed(string_tokens):

if token is Token.Text:

continue

elif opening is None:

opening = value

elif token is Token.String.Doc:

string.append(value[3:-3])

opening = None

elif value == opening:

opening = None

if not concatenate:

string = list()

else:

string.append(value)

if opening is None:

return ''

return ''.join(string)

def get_object(self, name):

attributes = name.split('.')

obj = eval(attributes.pop(0), self.interp.locals)

while attributes:

with inspection.AttrCleaner(obj):

obj = getattr(obj, attributes.pop(0))

return obj

def get_args(self):

"""Check if an unclosed parenthesis exists, then attempt to get the

self.current_func = None

if not self.config.arg_spec:

return False

# Get the name of the current function and where we are in

# the arguments

stack = [['', 0, '']]

try:

for (token, value) in PythonLexer().get_tokens(

self.current_line):

if token is Token.Punctuation:

if value in '([{':

stack.append(['', 0, value])

elif value in ')]}':

stack.pop()

elif value == ',':

try:

stack[-1][1] += 1

except TypeError:

stack[-1][1] = ''

stack[-1][0] = ''

elif value == ':' and stack[-1][2] == 'lambda':

stack.pop()

else:

stack[-1][0] = ''

elif (token is Token.Name or token in Token.Name.subtypes or

token is Token.Operator and value == '.'):

stack[-1][0] += value

elif token is Token.Operator and value == '=':

stack[-1][1] = stack[-1][0]

stack[-1][0] = ''

elif token is Token.Keyword and value == 'lambda':

stack.append(['', 0, value])

else:

stack[-1][0] = ''

while stack[-1][2] in '[{':

stack.pop()

_, arg_number, _ = stack.pop()

func, _, _ = stack.pop()

except IndexError:

return False

if not func:

return False

try:

f = self.get_object(func)

except Exception:

# another case of needing to catch every kind of error

# since user code is run in the case of descriptors

# XXX: Make sure you raise here if you're debugging the completion

# stuff !

return False

if inspect.isclass(f):

try:

if f.__init__ is not object.__init__:

f = f.__init__

except AttributeError:

return None

self.current_func = f

self.argspec = inspection.getargspec(func, f)

if self.argspec:

self.argspec.append(arg_number)

return True

return False

def get_source_of_current_name(self):

"""Return the unicode source code of the object which is bound to the

obj = self.current_func

try:

if obj is None:

line = self.current_line

if not line.strip():

raise SourceNotFound(_("Nothing to get source of"))

if inspection.is_eval_safe_name(line):

obj = self.get_object(line)

return inspection.get_source_unicode(obj)

except (AttributeError, NameError) as e:

msg = _("Cannot get source: %s") % (str(e), )

except IOError as e:

msg = str(e)

except TypeError as e:

if "built-in" in str(e):

msg = _("Cannot access source of %r") % (obj, )

else:

msg = _("No source code found for %s") % (self.current_line, )

raise SourceNotFound(msg)

def set_docstring(self):

self.docstring = None

if not self.get_args():

self.argspec = None

elif self.current_func is not None:

try:

self.docstring = pydoc.getdoc(self.current_func)

except IndexError:

self.docstring = None

else:

# pydoc.getdoc() returns an empty string if no

# docstring was found

if not self.docstring:

self.docstring = None

# What complete() does:

# Should we show the completion box? (are there matches, or is there a

# docstring to show?)

# Some completions should always be shown, other only if tab=True

# set the current docstring to the "current function's" docstring

# Populate the matches_iter object with new matches from the current state

# if none, clear the matches iterator

# If exactly one match that is equal to current line, clear matches

# If example one match and tab=True, then choose that and clear matches

def complete(self, tab=False):

"""Construct a full list of possible completions and

self.set_docstring()

matches, completer = autocomplete.get_completer_bpython(

cursor_offset=self.cursor_offset,

line=self.current_line,

locals_=self.interp.locals,

argspec=self.argspec,

current_block='\n'.join(self.buffer + [self.current_line]),

complete_magic_methods=self.config.complete_magic_methods,

history=self.history)

# TODO implement completer.shown_before_tab == False (filenames

# shouldn't fill screen)

if len(matches) == 0:

self.matches_iter.clear()

return bool(self.argspec)

self.matches_iter.update(self.cursor_offset,

self.current_line, matches, completer)

if len(matches) == 1:

if tab:

# if this complete is being run for a tab key press, substitute

# common sequence

self._cursor_offset, self._current_line = \

self.matches_iter.substitute_cseq()

return Repl.complete(self) # again for

elif self.matches_iter.current_word == matches[0]:

self.matches_iter.clear()

return False

return completer.shown_before_tab

else:

assert len(matches) > 1

return tab or completer.shown_before_tab

def format_docstring(self, docstring, width, height):

"""Take a string and try to format it into a sane list of strings to be

lines = docstring.split('\n')

out = []

i = 0

for line in lines:

i += 1

if not line.strip():

out.append('\n')

for block in textwrap.wrap(line, width):

out.append(' ' + block + '\n')

if i >= height:

return out

i += 1

# Drop the last newline

out[-1] = out[-1].rstrip()

return out

def next_indentation(self):

"""Return the indentation of the next line based on the current

if self.buffer:

indentation = next_indentation(self.buffer[-1],

self.config.tab_length)

if indentation and self.config.dedent_after > 0:

line_is_empty = lambda line: not line.strip()

empty_lines = takewhile(line_is_empty, reversed(self.buffer))

if sum(1 for _ in empty_lines) >= self.config.dedent_after:

indentation -= 1

else:

indentation = 0

return indentation

def formatforfile(self, s):

"""Format the stdout buffer to something suitable for writing to disk,

def process():

for line in s.split('\n'):

if line.startswith(self.ps1):

yield line[len(self.ps1):]

elif line.startswith(self.ps2):

yield line[len(self.ps2):]

elif line.rstrip():

yield "# OUT: %s" % (line,)

return "\n".join(process())

def write2file(self):

"""Prompt for a filename and write the current contents of the stdout

try:

fn = self.interact.file_prompt(_('Save to file (Esc to cancel): '))

if not fn:

self.interact.notify(_('Save cancelled.'))

return

except ValueError:

self.interact.notify(_('Save cancelled.'))

return

if fn.startswith('~'):

fn = os.path.expanduser(fn)

if not fn.endswith('.py') and self.config.save_append_py:

fn = fn + '.py'

mode = 'w'

if os.path.exists(fn):

mode = self.interact.file_prompt(_('%s already exists. Do you '

'want to (c)ancel, '

' (o)verwrite or '

'(a)ppend? ') % (fn, ))

if mode in ('o', 'overwrite', _('overwrite')):

mode = 'w'

elif mode in ('a', 'append', _('append')):

mode = 'a'

else:

self.interact.notify(_('Save cancelled.'))

return

s = self.formatforfile(self.getstdout())

try:

with open(fn, mode) as f:

f.write(s)

except IOError as e:

self.interact.notify(_("Error writing file '%s': %s") % (fn,

str(e)))

else:

self.interact.notify(_('Saved to %s.') % (fn, ))

def copy2clipboard(self):

"""Copy current content to clipboard."""

if self.clipboard is None:

self.interact.notify(_('No clipboard available.'))

return

content = self.formatforfile(self.getstdout())

try:

self.clipboard.copy(content)

except CopyFailed:

self.interact.notify(_('Could not copy to clipboard.'))

else:

self.interact.notify(_('Copied content to clipboard.'))

def pastebin(self, s=None):

"""Upload to a pastebin and display the URL in the status bar."""

if s is None:

s = self.getstdout()

if (self.config.pastebin_confirm and

not self.interact.confirm(_("Pastebin buffer? (y/N) "))):

self.interact.notify(_("Pastebin aborted."))

return

return self.do_pastebin(s)

def do_pastebin(self, s):

"""Actually perform the upload."""

if s == self.prev_pastebin_content:

self.interact.notify(_('Duplicate pastebin. Previous URL: %s. '

'Removal URL: %s') %

(self.prev_pastebin_url,

self.prev_removal_url), 10)

return self.prev_pastebin_url

if self.config.pastebin_helper:

return self.do_pastebin_helper(s)

else:

return self.do_pastebin_json(s)

def do_pastebin_json(self, s):

"""Upload to pastebin via json interface."""

url = urljoin(self.config.pastebin_url, '/json/new')

payload = {

'code': s,

'lexer': 'pycon',

'expiry': self.config.pastebin_expiry

}

self.interact.notify(_('Posting data to pastebin...'))

try:

response = requests.post(url, data=payload, verify=True)

response.raise_for_status()

except requests.exceptions.RequestException as exc:

self.interact.notify(_('Upload failed: %s') % (str(exc), ))

return

self.prev_pastebin_content = s

data = response.json()

paste_url_template = Template(self.config.pastebin_show_url)

paste_id = urlquote(data['paste_id'])

paste_url = paste_url_template.safe_substitute(paste_id=paste_id)

removal_url_template = Template(self.config.pastebin_removal_url)

removal_id = urlquote(data['removal_id'])

removal_url = removal_url_template.safe_substitute(

removal_id=removal_id)

self.prev_pastebin_url = paste_url

self.prev_removal_url = removal_url

self.interact.notify(_('Pastebin URL: %s - Removal URL: %s') %

(paste_url, removal_url), 10)

return paste_url

def do_pastebin_helper(self, s):

"""Call out to helper program for pastebin upload."""

self.interact.notify(_('Posting data to pastebin...'))

try:

helper = subprocess.Popen('',

executable=self.config.pastebin_helper,

stdin=subprocess.PIPE,

stdout=subprocess.PIPE)

helper.stdin.write(s.encode(getpreferredencoding()))

output = helper.communicate()[0].decode(getpreferredencoding())

paste_url = output.split()[0]

except OSError as e:

if e.errno == errno.ENOENT:

self.interact.notify(_('Upload failed: '

'Helper program not found.'))

else:

self.interact.notify(_('Upload failed: '

'Helper program could not be run.'))

return

if helper.returncode != 0:

self.interact.notify(_('Upload failed: '

'Helper program returned non-zero exit '

'status %d.' % (helper.returncode, )))

return

if not paste_url:

self.interact.notify(_('Upload failed: '

'No output from helper program.'))

return

else:

parsed_url = urlparse(paste_url)

if (not parsed_url.scheme

or any(unicodedata.category(c) == 'Cc'

for c in paste_url)):

self.interact.notify(_("Upload failed: "

"Failed to recognize the helper "

"program's output as an URL."))

return

self.prev_pastebin_content = s

self.interact.notify(_('Pastebin URL: %s') % (paste_url, ), 10)

return paste_url

def push(self, s, insert_into_history=True):

"""Push a line of code onto the buffer so it can process it all

s = s.rstrip('\n')

self.buffer.append(s)

if insert_into_history:

self.insert_into_history(s)

more = self.interp.runsource('\n'.join(self.buffer))

if not more:

self.buffer = []

return more

def insert_into_history(self, s):

try:

self.rl_history.append_reload_and_write(s, self.config.hist_file,

getpreferredencoding())

except RuntimeError as e:

self.interact.notify(str(e))

def prompt_undo(self):

"""Returns how many lines to undo, 0 means don't undo"""

if (self.config.single_undo_time < 0 or

self.interp.timer.estimate() < self.config.single_undo_time):

return 1

est = self.interp.timer.estimate()

n = self.interact.file_prompt(

_("Undo how many lines? (Undo will take up to ~%.1f seconds) [1]")

% (est,))

try:

if n == '':

n = '1'

n = int(n)

except ValueError:

self.interact.notify(_('Undo canceled'), .1)

return 0

else:

if n == 0:

self.interact.notify(_('Undo canceled'), .1)

return 0

else:

message = ngettext('Undoing %d line... (est. %.1f seconds)',

'Undoing %d lines... (est. %.1f seconds)',

n)

self.interact.notify(message % (n, est), .1)

return n

def undo(self, n=1):

"""Go back in the undo history n steps and call reevaluate()

if not self.history:

return None

self.interp.timer.reset_timer()

if len(self.history) < n:

n = len(self.history)

entries = list(self.rl_history.entries)

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

self.reevaluate()

self.rl_history.entries = entries

def flush(self):

"""Olivier Grisel brought it to my attention that the logging

def close(self):

"""See the flush() method docstring."""

def tokenize(self, s, newline=False):

"""Tokenizes a line of code, returning pygments tokens

highlighted_paren = None

source = '\n'.join(self.buffer + [s])

cursor = len(source) - self.cpos

if self.cpos:

cursor += 1

stack = list()

all_tokens = list(PythonLexer().get_tokens(source))

# Unfortunately, Pygments adds a trailing newline and strings with

# no size, so strip them

while not all_tokens[-1][1]:

all_tokens.pop()

all_tokens[-1] = (all_tokens[-1][0], all_tokens[-1][1].rstrip('\n'))

line = pos = 0

parens = dict(zip('{([', '})]'))

line_tokens = list()

saved_tokens = list()

search_for_paren = True

for (token, value) in split_lines(all_tokens):

pos += len(value)

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

line += 1

# Remove trailing newline

line_tokens = list()

saved_tokens = list()

continue

line_tokens.append((token, value))

saved_tokens.append((token, value))

if not search_for_paren:

continue

under_cursor = (pos == cursor)

if token is Token.Punctuation:

if value in parens:

if under_cursor:

line_tokens[-1] = (Parenthesis.UnderCursor, value)

# Push marker on the stack

stack.append((Parenthesis, value))

else:

stack.append((line, len(line_tokens) - 1,

line_tokens, value))

elif value in itervalues(parens):

saved_stack = list(stack)

try:

while True:

opening = stack.pop()

if parens[opening[-1]] == value:

break

except IndexError:

# SyntaxError.. more closed parentheses than

# opened or a wrong closing paren

opening = None

if not saved_stack:

search_for_paren = False

else: