r'^\s+raise(\s.*)?$', # raise statement (+ space + other stuff, maybe)

r'^\s+raise\([^\)]*\).*$', # wacky raise with immediate open paren

r'^\s+return(\s.*)?$', # normal return (+ space + other stuff, maybe)

r'^\s+return\([^\)]*\).*$', # wacky return with immediate open paren

r'^\s+pass\s*$', # pass (optionally followed by trailing spaces)

r'^\s+break\s*$', # break (optionally followed by trailing spaces)

r'^\s+continue\s*$', # continue (optionally followed by trailing spaces)

"""Return the number of initial spaces in a string.

ini_spaces = ini_spaces_re.match(s)

if ini_spaces:

return ini_spaces.end()

else:

type = exact_type = INCOMPLETE_STRING

def __init__(self, s, start, end, line):

self.s = s

self.start = start

self.end = end

type = exact_type = IN_MULTILINE_STATEMENT

def __init__(self, pos, line):

self.s = ''

self.start = self.end = pos

"""Iterate over tokens from a possibly-incomplete string of code.

readline = io.StringIO(s).readline

token = tokenize.TokenInfo(tokenize.NEWLINE, '', (1, 0), (1, 0), '')

try:

for token in tokenize.generate_tokens(readline):

yield token

except tokenize.TokenError as e:

# catch EOF error

lines = s.splitlines(keepends=True)

end = len(lines), len(lines[-1])

if 'multi-line string' in e.args[0]:

l, c = start = token.end

s = lines[l-1][c:] + ''.join(lines[l:])

yield IncompleteString(s, start, end, lines[-1])

elif 'multi-line statement' in e.args[0]:

yield InMultilineStatement(end, lines[-1])

else:

"""Find the number of spaces for the next line of indentation"""

tokens = list(partial_tokens(code))

if tokens[-1].type == tokenize.ENDMARKER:

tokens.pop()

if not tokens:

return 0

while (tokens[-1].type in {tokenize.DEDENT, tokenize.NEWLINE, tokenize.COMMENT}):

tokens.pop()

if tokens[-1].type == INCOMPLETE_STRING:

# Inside a multiline string

return 0

# Find the indents used before

prev_indents = [0]

def _add_indent(n):

if n != prev_indents[-1]:

prev_indents.append(n)

tokiter = iter(tokens)

for tok in tokiter:

if tok.type in {tokenize.INDENT, tokenize.DEDENT}:

_add_indent(tok.end[1])

elif (tok.type == tokenize.NL):

try:

_add_indent(next(tokiter).start[1])

except StopIteration:

break

last_indent = prev_indents.pop()

# If we've just opened a multiline statement (e.g. 'a = ['), indent more

if tokens[-1].type == IN_MULTILINE_STATEMENT:

if tokens[-2].exact_type in {tokenize.LPAR, tokenize.LSQB, tokenize.LBRACE}:

return last_indent + 4

return last_indent

if tokens[-1].exact_type == tokenize.COLON:

# Line ends with colon - indent

return last_indent + 4

if last_indent:

# Examine the last line for dedent cues - statements like return or

# raise which normally end a block of code.

last_line_starts = 0

for i, tok in enumerate(tokens):

if tok.type == tokenize.NEWLINE:

last_line_starts = i + 1

last_line_tokens = tokens[last_line_starts:]

names = [t.string for t in last_line_tokens if t.type == tokenize.NAME]

if names and names[0] in {'raise', 'return', 'pass', 'break', 'continue'}:

# Find the most recent indentation less than the current level

for indent in reversed(prev_indents):

if indent < last_indent:

return indent

"""Determine if the input source ends in a blank.

if not src: return False

ll = src.splitlines()[-1]

"""Determine if the input source ends in two blanks.

if not src: return False

# The logic here is tricky: I couldn't get a regexp to work and pass all

# the tests, so I took a different approach: split the source by lines,

# grab the last two and prepend '###\n' as a stand-in for whatever was in

# the body before the last two lines. Then, with that structure, it's

# possible to analyze with two regexps. Not the most elegant solution, but

# it works. If anyone tries to change this logic, make sure to validate

# the whole test suite first!

new_src = '\n'.join(['###\n'] + src.splitlines()[-2:])

return (bool(last_two_blanks_re.match(new_src)) or

"""Remove all comments from input source.

"""Return the default standard input encoding.

# There are strange environments for which sys.stdin.encoding is None. We

# ensure that a valid encoding is returned.

encoding = getattr(sys.stdin, 'encoding', None)

if encoding is None:

encoding = 'ascii'

r"""An object that can accumulate lines of Python source before execution.

# A cache for storing the current indentation

# The first value stores the most recently processed source input

# The second value is the number of spaces for the current indentation

# If self.source matches the first value, the second value is a valid

# current indentation. Otherwise, the cache is invalid and the indentation

# must be recalculated.

_indent_spaces_cache = None, None

# String, indicating the default input encoding. It is computed by default

# at initialization time via get_input_encoding(), but it can be reset by a

# client with specific knowledge of the encoding.

encoding = ''

# String where the current full source input is stored, properly encoded.

# Reading this attribute is the normal way of querying the currently pushed

# source code, that has been properly encoded.

source = ''

# Code object corresponding to the current source. It is automatically

# synced to the source, so it can be queried at any time to obtain the code

# object; it will be None if the source doesn't compile to valid Python.

code = None

# Private attributes

# List with lines of input accumulated so far

_buffer = None

# Command compiler

_compile = None

# Boolean indicating whether the current block is complete

_is_complete = None

# Boolean indicating whether the current block has an unrecoverable syntax error

_is_invalid = False

def __init__(self):

"""Create a new InputSplitter instance.

self._buffer = []

self._compile = codeop.CommandCompiler()

self.encoding = get_input_encoding()

def reset(self):

"""Reset the input buffer and associated state."""

self._buffer[:] = []

self.source = ''

self.code = None

self._is_complete = False

self._is_invalid = False

def source_reset(self):

"""Return the input source and perform a full reset.

out = self.source

self.reset()

return out

def check_complete(self, source):

"""Return whether a block of code is ready to execute, or should be continued

self.reset()

try:

self.push(source)

except SyntaxError:

# Transformers in IPythonInputSplitter can raise SyntaxError,

# which push() will not catch.

return 'invalid', None

else:

if self._is_invalid:

return 'invalid', None

elif self.push_accepts_more():

return 'incomplete', self.get_indent_spaces()

else:

return 'complete', None

finally:

self.reset()

def push(self, lines:str) -> bool:

"""Push one or more lines of input.

assert isinstance(lines, str)

self._store(lines)

source = self.source

# Before calling _compile(), reset the code object to None so that if an

# exception is raised in compilation, we don't mislead by having

# inconsistent code/source attributes.

self.code, self._is_complete = None, None

self._is_invalid = False

# Honor termination lines properly

if source.endswith('\\\n'):

return False

try:

with warnings.catch_warnings():

warnings.simplefilter('error', SyntaxWarning)

self.code = self._compile(source, symbol="exec")

# Invalid syntax can produce any of a number of different errors from

# inside the compiler, so we have to catch them all. Syntax errors

# immediately produce a 'ready' block, so the invalid Python can be

# sent to the kernel for evaluation with possible ipython

# special-syntax conversion.

except (SyntaxError, OverflowError, ValueError, TypeError,

MemoryError, SyntaxWarning):

self._is_complete = True

self._is_invalid = True

else:

# Compilation didn't produce any exceptions (though it may not have

# given a complete code object)

self._is_complete = self.code is not None

return self._is_complete

def push_accepts_more(self):

"""Return whether a block of interactive input can accept more input.

# With incomplete input, unconditionally accept more

# A syntax error also sets _is_complete to True - see push()

if not self._is_complete:

#print("Not complete") # debug

return True

# The user can make any (complete) input execute by leaving a blank line

last_line = self.source.splitlines()[-1]

if (not last_line) or last_line.isspace():

#print("Blank line") # debug

return False

# If there's just a single line or AST node, and we're flush left, as is

# the case after a simple statement such as 'a=1', we want to execute it

# straight away.

if self.get_indent_spaces() == 0:

if len(self.source.splitlines()) <= 1:

return False

try:

code_ast = ast.parse(u''.join(self._buffer))

except Exception:

#print("Can't parse AST") # debug

return False

else:

if len(code_ast.body) == 1 and \

not hasattr(code_ast.body[0], 'body'):

#print("Simple statement") # debug

return False

# General fallback - accept more code

return True

def get_indent_spaces(self):

sourcefor, n = self._indent_spaces_cache

if sourcefor == self.source:

return n

# self.source always has a trailing newline

n = find_next_indent(self.source[:-1])

self._indent_spaces_cache = (self.source, n)

return n

# Backwards compatibility. I think all code that used .indent_spaces was

# inside IPython, but we can leave this here until IPython 7 in case any

# other modules are using it. -TK, November 2017

indent_spaces = property(get_indent_spaces)

def _store(self, lines, buffer=None, store='source'):

"""Store one or more lines of input.

if buffer is None:

buffer = self._buffer

if lines.endswith('\n'):

buffer.append(lines)

else:

buffer.append(lines+'\n')

setattr(self, store, self._set_source(buffer))

def _set_source(self, buffer):

"""An input splitter that recognizes all of IPython's special syntax."""

# String with raw, untransformed input.

source_raw = ''

# Flag to track when a transformer has stored input that it hasn't given

# back yet.

transformer_accumulating = False

# Flag to track when assemble_python_lines has stored input that it hasn't

# given back yet.

within_python_line = False

# Private attributes

# List with lines of raw input accumulated so far.

_buffer_raw = None

def __init__(self, line_input_checker=True, physical_line_transforms=None,

logical_line_transforms=None, python_line_transforms=None):

super(IPythonInputSplitter, self).__init__()

self._buffer_raw = []

self._validate = True

if physical_line_transforms is not None:

self.physical_line_transforms = physical_line_transforms

else:

self.physical_line_transforms = [

leading_indent(),

classic_prompt(),

ipy_prompt(),

cellmagic(end_on_blank_line=line_input_checker),

]

self.assemble_logical_lines = assemble_logical_lines()

if logical_line_transforms is not None:

self.logical_line_transforms = logical_line_transforms

else:

self.logical_line_transforms = [

help_end(),

escaped_commands(),

assign_from_magic(),

assign_from_system(),

]

self.assemble_python_lines = assemble_python_lines()

if python_line_transforms is not None:

self.python_line_transforms = python_line_transforms

else:

# We don't use any of these at present

self.python_line_transforms = []

@property

def transforms(self):

"Quick access to all transformers."

return self.physical_line_transforms + \

[self.assemble_logical_lines] + self.logical_line_transforms + \

[self.assemble_python_lines] + self.python_line_transforms

@property

def transforms_in_use(self):

"""Transformers, excluding logical line transformers if we're in a

t = self.physical_line_transforms[:]

if not self.within_python_line:

t += [self.assemble_logical_lines] + self.logical_line_transforms

return t + [self.assemble_python_lines] + self.python_line_transforms

def reset(self):

"""Reset the input buffer and associated state."""

super(IPythonInputSplitter, self).reset()

self._buffer_raw[:] = []

self.source_raw = ''

self.transformer_accumulating = False

self.within_python_line = False

for t in self.transforms:

try:

t.reset()

except SyntaxError:

# Nothing that calls reset() expects to handle transformer

# errors

pass

def flush_transformers(self):

def _flush(transform, outs):

"""yield transformed lines

for out in outs:

for line in out.splitlines():

# push one line at a time

tmp = transform.push(line)

if tmp is not None:

yield tmp

# reset the transform

tmp = transform.reset()

if tmp is not None:

yield tmp

out = []

for t in self.transforms_in_use:

out = _flush(t, out)

out = list(out)

if out:

self._store('\n'.join(out))

def raw_reset(self):

"""Return raw input only and perform a full reset.

out = self.source_raw

self.reset()

return out

def source_reset(self):

try:

self.flush_transformers()

return self.source

finally:

self.reset()

def push_accepts_more(self):

if self.transformer_accumulating:

return True

else:

return super(IPythonInputSplitter, self).push_accepts_more()

def transform_cell(self, cell):

"""Process and translate a cell of input.

self.reset()

try:

self.push(cell)

self.flush_transformers()

return self.source

finally:

self.reset()

def push(self, lines:str) -> bool:

"""Push one or more lines of IPython input.

assert isinstance(lines, str)

# We must ensure all input is pure unicode

# ''.splitlines() --> [], but we need to push the empty line to transformers

lines_list = lines.splitlines()

if not lines_list:

lines_list = ['']

# Store raw source before applying any transformations to it. Note

# that this must be done *after* the reset() call that would otherwise

# flush the buffer.

self._store(lines, self._buffer_raw, 'source_raw')

transformed_lines_list = []

for line in lines_list:

transformed = self._transform_line(line)

if transformed is not None:

transformed_lines_list.append(transformed)

if transformed_lines_list:

transformed_lines = '\n'.join(transformed_lines_list)

return super(IPythonInputSplitter, self).push(transformed_lines)

else:

# Got nothing back from transformers - they must be waiting for

# more input.

return False

def _transform_line(self, line):

"""Push a line of input code through the various transformers.

def _accumulating(dbg):

#print(dbg)

self.transformer_accumulating = True

return None

for transformer in self.physical_line_transforms:

line = transformer.push(line)

if line is None:

return _accumulating(transformer)

if not self.within_python_line:

line = self.assemble_logical_lines.push(line)

if line is None:

return _accumulating('acc logical line')

for transformer in self.logical_line_transforms:

line = transformer.push(line)

if line is None:

return _accumulating(transformer)

line = self.assemble_python_lines.push(line)

if line is None:

self.within_python_line = True

return _accumulating('acc python line')

else:

self.within_python_line = False

for transformer in self.python_line_transforms:

line = transformer.push(line)

if line is None:

return _accumulating(transformer)

#print("transformers clear") #debug

self.transformer_accumulating = False