"""

block = []

lines = part.split('\n')

N = len(lines)

i = 0

decorator = None

while 1:

if i==N:

# nothing left to parse -- the last line

break

line = lines[i]

i += 1

line_stripped = line.strip()

if line_stripped.startswith('#'):

block.append((COMMENT, line))

continue

if any(

line_stripped.startswith("@" + pseudo_decorator)

for pseudo_decorator in PSEUDO_DECORATORS

):

if decorator:

raise RuntimeError(

"Applying multiple pseudo-decorators on one line is not supported"

)

else:

decorator = line_stripped

continue

# does this look like an input line?

matchin = rgxin.match(line)

if matchin:

lineno, inputline = int(matchin.group(1)), matchin.group(2)

# the ....: continuation string

continuation = ' %s:'%''.join(['.']*(len(str(lineno))+2))

Nc = len(continuation)

# input lines can continue on for more than one line, if

# we have a '\' line continuation char or a function call

# echo line 'print'. The input line can only be

# terminated by the end of the block or an output line, so

# we parse out the rest of the input line if it is

# multiline as well as any echo text

rest = []

while i<N:

# look ahead; if the next line is blank, or a comment, or

# an output line, we're done

nextline = lines[i]

matchout = rgxout.match(nextline)

# print("nextline=%s, continuation=%s, starts=%s"%(nextline, continuation, nextline.startswith(continuation)))

if matchout or nextline.startswith('#'):

break

elif nextline.startswith(continuation):

# The default ipython_rgx* treat the space following the colon as optional.

# However, If the space is there we must consume it or code

# employing the cython_magic extension will fail to execute.

#

# This works with the default ipython_rgx* patterns,

# If you modify them, YMMV.

nextline = nextline[Nc:]

if nextline and nextline[0] == ' ':

nextline = nextline[1:]

inputline += '\n' + nextline

else:

rest.append(nextline)

i+= 1

block.append((INPUT, (decorator, inputline, '\n'.join(rest))))

continue

# if it looks like an output line grab all the text to the end

# of the block

matchout = rgxout.match(line)

if matchout:

lineno, output = int(matchout.group(1)), matchout.group(2)

if i<N-1:

output = '\n'.join([output] + lines[i:])

block.append((OUTPUT, output))

break

"""An embedded IPython instance to run inside Sphinx"""

def __init__(self, exec_lines=None):

self.cout = StringIO()

if exec_lines is None:

exec_lines = []

# Create config object for IPython

config = Config()

config.HistoryManager.hist_file = ':memory:'

config.InteractiveShell.autocall = False

config.InteractiveShell.autoindent = False

config.InteractiveShell.colors = "nocolor"

# create a profile so instance history isn't saved

tmp_profile_dir = tempfile.mkdtemp(prefix='profile_')

profname = 'auto_profile_sphinx_build'

pdir = os.path.join(tmp_profile_dir,profname)

profile = ProfileDir.create_profile_dir(pdir)

# Create and initialize global ipython, but don't start its mainloop.

# This will persist across different EmbeddedSphinxShell instances.

IP = InteractiveShell.instance(config=config, profile_dir=profile)

atexit.register(self.cleanup)

# Store a few parts of IPython we'll need.

self.IP = IP

self.user_ns = self.IP.user_ns

self.user_global_ns = self.IP.user_global_ns

self.input = ''

self.output = ''

self.tmp_profile_dir = tmp_profile_dir

self.is_verbatim = False

self.is_doctest = False

self.is_suppress = False

# Optionally, provide more detailed information to shell.

# this is assigned by the SetUp method of IPythonDirective

# to point at itself.

#

# So, you can access handy things at self.directive.state

self.directive = None

# on the first call to the savefig decorator, we'll import

# pyplot as plt so we can make a call to the plt.gcf().savefig

self._pyplot_imported = False

# Prepopulate the namespace.

for line in exec_lines:

self.process_input_line(line, store_history=False)

def cleanup(self):

shutil.rmtree(self.tmp_profile_dir, ignore_errors=True)

def clear_cout(self):

self.cout.seek(0)

self.cout.truncate(0)

def process_input_line(self, line, store_history):

return self.process_input_lines([line], store_history=store_history)

def process_input_lines(self, lines, store_history=True):

"""process the input, capturing stdout"""

stdout = sys.stdout

source_raw = '\n'.join(lines)

try:

sys.stdout = self.cout

self.IP.run_cell(source_raw, store_history=store_history)

finally:

sys.stdout = stdout

def process_image(self, decorator):

"""

savefig_dir = self.savefig_dir

source_dir = self.source_dir

saveargs = decorator.split(' ')

filename = saveargs[1]

# insert relative path to image file in source

# as absolute path for Sphinx

# sphinx expects a posix path, even on Windows

path = pathlib.Path(savefig_dir, filename)

outfile = '/' + path.relative_to(source_dir).as_posix()

imagerows = ['.. image:: %s' % outfile]

for kwarg in saveargs[2:]:

arg, val = kwarg.split('=')

arg = arg.strip()

val = val.strip()

imagerows.append(' :%s: %s'%(arg, val))

image_file = os.path.basename(outfile) # only return file name

image_directive = '\n'.join(imagerows)

return image_file, image_directive

# Callbacks for each type of token

def process_input(self, data, input_prompt, lineno):

"""

decorator, input, rest = data

image_file = None

image_directive = None

is_verbatim = decorator=='@verbatim' or self.is_verbatim

is_doctest = (decorator is not None and \

decorator.startswith('@doctest')) or self.is_doctest

is_suppress = decorator=='@suppress' or self.is_suppress

is_okexcept = decorator=='@okexcept' or self.is_okexcept

is_okwarning = decorator=='@okwarning' or self.is_okwarning

is_savefig = decorator is not None and \

decorator.startswith('@savefig')

input_lines = input.split('\n')

if len(input_lines) > 1:

if input_lines[-1] != "":

input_lines.append('') # make sure there's a blank line

# so splitter buffer gets reset

continuation = ' %s:'%''.join(['.']*(len(str(lineno))+2))

if is_savefig:

image_file, image_directive = self.process_image(decorator)

ret = []

is_semicolon = False

# Hold the execution count, if requested to do so.

if is_suppress and self.hold_count:

store_history = False

else:

store_history = True

# Note: catch_warnings is not thread safe

with warnings.catch_warnings(record=True) as ws:

if input_lines[0].endswith(';'):

is_semicolon = True

#for i, line in enumerate(input_lines):

# process the first input line

if is_verbatim:

self.process_input_lines([''])

self.IP.execution_count += 1 # increment it anyway

else:

# only submit the line in non-verbatim mode

self.process_input_lines(input_lines, store_history=store_history)

if not is_suppress:

for i, line in enumerate(input_lines):

if i == 0:

formatted_line = '%s %s'%(input_prompt, line)

else:

formatted_line = '%s %s'%(continuation, line)

ret.append(formatted_line)

if not is_suppress and len(rest.strip()) and is_verbatim:

# The "rest" is the standard output of the input. This needs to be

# added when in verbatim mode. If there is no "rest", then we don't

# add it, as the new line will be added by the processed output.

ret.append(rest)

# Fetch the processed output. (This is not the submitted output.)

self.cout.seek(0)

processed_output = self.cout.read()

if not is_suppress and not is_semicolon:

#

# In IPythonDirective.run, the elements of `ret` are eventually

# combined such that '' entries correspond to newlines. So if

# `processed_output` is equal to '', then the adding it to `ret`

# ensures that there is a blank line between consecutive inputs

# that have no outputs, as in:

#

# In [1]: x = 4

#

# In [2]: x = 5

#

# When there is processed output, it has a '\n' at the tail end. So

# adding the output to `ret` will provide the necessary spacing

# between consecutive input/output blocks, as in:

#

# In [1]: x

# Out[1]: 5

#

# In [2]: x

# Out[2]: 5

#

# When there is stdout from the input, it also has a '\n' at the

# tail end, and so this ensures proper spacing as well. E.g.:

#

# In [1]: print(x)

# 5

#

# In [2]: x = 5

#

# When in verbatim mode, `processed_output` is empty (because

# nothing was passed to IP. Sometimes the submitted code block has

# an Out[] portion and sometimes it does not. When it does not, we

# need to ensure proper spacing, so we have to add '' to `ret`.

# However, if there is an Out[] in the submitted code, then we do

# not want to add a newline as `process_output` has stuff to add.

# The difficulty is that `process_input` doesn't know if

# `process_output` will be called---so it doesn't know if there is

# Out[] in the code block. The requires that we include a hack in

# `process_block`. See the comments there.

#

ret.append(processed_output)

elif is_semicolon:

# Make sure there is a newline after the semicolon.

ret.append('')

# context information

filename = "Unknown"

lineno = 0

if self.directive.state:

filename = self.directive.state.document.current_source

lineno = self.directive.state.document.current_line

# Use sphinx logger for warnings

logger = logging.getLogger(__name__)

# output any exceptions raised during execution to stdout

# unless :okexcept: has been specified.

if not is_okexcept and (

("Traceback" in processed_output) or ("SyntaxError" in processed_output)

):

s = "\n>>>" + ("-" * 73) + "\n"

s += "Exception in %s at block ending on line %s\n" % (filename, lineno)

s += "Specify :okexcept: as an option in the ipython:: block to suppress this message\n"

s += processed_output + "\n"

s += "<<<" + ("-" * 73)

logger.warning(s)

if self.warning_is_error:

raise RuntimeError(

"Unexpected exception in `{}` line {}".format(filename, lineno)

)

# output any warning raised during execution to stdout

# unless :okwarning: has been specified.

if not is_okwarning:

for w in ws:

s = "\n>>>" + ("-" * 73) + "\n"

s += "Warning in %s at block ending on line %s\n" % (filename, lineno)

s += "Specify :okwarning: as an option in the ipython:: block to suppress this message\n"

s += ("-" * 76) + "\n"

s += warnings.formatwarning(

w.message, w.category, w.filename, w.lineno, w.line

)

s += "<<<" + ("-" * 73)

logger.warning(s)

if self.warning_is_error:

raise RuntimeError(

"Unexpected warning in `{}` line {}".format(filename, lineno)

)

self.clear_cout()

return (ret, input_lines, processed_output,

is_doctest, decorator, image_file, image_directive)

def process_output(self, data, output_prompt, input_lines, output,

is_doctest, decorator, image_file):

"""

# Recall: `data` is the submitted output, and `output` is the processed

# output from `input_lines`.

TAB = ' ' * 4

if is_doctest and output is not None:

found = output # This is the processed output

found = found.strip()

submitted = data.strip()

if self.directive is None:

source = 'Unavailable'

content = 'Unavailable'

else:

source = self.directive.state.document.current_source

content = self.directive.content

# Add tabs and join into a single string.

content = '\n'.join([TAB + line for line in content])

# Make sure the output contains the output prompt.

ind = found.find(output_prompt)

if ind < 0:

e = ('output does not contain output prompt\n\n'

'Document source: {0}\n\n'

'Raw content: \n{1}\n\n'

'Input line(s):\n{TAB}{2}\n\n'

'Output line(s):\n{TAB}{3}\n\n')

e = e.format(source, content, '\n'.join(input_lines),

repr(found), TAB=TAB)

raise RuntimeError(e)

found = found[len(output_prompt):].strip()

# Handle the actual doctest comparison.

if decorator.strip() == '@doctest':

# Standard doctest

if found != submitted:

e = ('doctest failure\n\n'

'Document source: {0}\n\n'

'Raw content: \n{1}\n\n'

'On input line(s):\n{TAB}{2}\n\n'

'we found output:\n{TAB}{3}\n\n'

'instead of the expected:\n{TAB}{4}\n\n')

e = e.format(source, content, '\n'.join(input_lines),

repr(found), repr(submitted), TAB=TAB)

raise RuntimeError(e)

else:

self.custom_doctest(decorator, input_lines, found, submitted)

# When in verbatim mode, this holds additional submitted output

# to be written in the final Sphinx output.

# https://github.com/ipython/ipython/issues/5776

out_data = []

is_verbatim = decorator=='@verbatim' or self.is_verbatim

if is_verbatim and data.strip():

# Note that `ret` in `process_block` has '' as its last element if

# the code block was in verbatim mode. So if there is no submitted

# output, then we will have proper spacing only if we do not add

# an additional '' to `out_data`. This is why we condition on

# `and data.strip()`.

# The submitted output has no output prompt. If we want the

# prompt and the code to appear, we need to join them now

# instead of adding them separately---as this would create an

# undesired newline. How we do this ultimately depends on the

# format of the output regex. I'll do what works for the default

# prompt for now, and we might have to adjust if it doesn't work

# in other cases. Finally, the submitted output does not have

# a trailing newline, so we must add it manually.

out_data.append("{0} {1}\n".format(output_prompt, data))

return out_data

def process_comment(self, data):

"""Process data fPblock for COMMENT token."""

if not self.is_suppress:

return [data]

def save_image(self, image_file):

"""

self.ensure_pyplot()

command = 'plt.gcf().savefig("%s")'%image_file

# print('SAVEFIG', command) # dbg

self.process_input_line('bookmark ipy_thisdir', store_history=False)

self.process_input_line('cd -b ipy_savedir', store_history=False)

self.process_input_line(command, store_history=False)

self.process_input_line('cd -b ipy_thisdir', store_history=False)

self.process_input_line('bookmark -d ipy_thisdir', store_history=False)

self.clear_cout()

def process_block(self, block):

"""

ret = []

output = None

input_lines = None

lineno = self.IP.execution_count

input_prompt = self.promptin % lineno

output_prompt = self.promptout % lineno

image_file = None

image_directive = None

found_input = False

for token, data in block:

if token == COMMENT:

out_data = self.process_comment(data)

elif token == INPUT:

found_input = True

(out_data, input_lines, output, is_doctest,

decorator, image_file, image_directive) = \

self.process_input(data, input_prompt, lineno)

elif token == OUTPUT:

if not found_input:

TAB = ' ' * 4

linenumber = 0

source = 'Unavailable'

content = 'Unavailable'

if self.directive:

linenumber = self.directive.state.document.current_line

source = self.directive.state.document.current_source

content = self.directive.content

# Add tabs and join into a single string.

content = '\n'.join([TAB + line for line in content])

e = ('\n\nInvalid block: Block contains an output prompt '

'without an input prompt.\n\n'

'Document source: {0}\n\n'

'Content begins at line {1}: \n\n{2}\n\n'

'Problematic block within content: \n\n{TAB}{3}\n\n')

e = e.format(source, linenumber, content, block, TAB=TAB)

# Write, rather than include in exception, since Sphinx

# will truncate tracebacks.

sys.stdout.write(e)

raise RuntimeError('An invalid block was detected.')

out_data = \

self.process_output(data, output_prompt, input_lines,

output, is_doctest, decorator,

image_file)

if out_data:

# Then there was user submitted output in verbatim mode.

# We need to remove the last element of `ret` that was

# added in `process_input`, as it is '' and would introduce

# an undesirable newline.

assert(ret[-1] == '')

del ret[-1]

if out_data:

ret.extend(out_data)

# save the image files

if image_file is not None:

self.save_image(image_file)

return ret, image_directive

def ensure_pyplot(self):

"""

# We are here if the @figure pseudo decorator was used. Thus, it's

# possible that we could be here even if python_mplbackend were set to

# `None`. That's also strange and perhaps worthy of raising an

# exception, but for now, we just set the backend to 'agg'.

if not self._pyplot_imported:

if 'matplotlib.backends' not in sys.modules:

# Then ipython_matplotlib was set to None but there was a

# call to the @figure decorator (and ipython_execlines did

# not set a backend).

#raise Exception("No backend was set, but @figure was used!")

import matplotlib

matplotlib.use('agg')

# Always import pyplot into embedded shell.

self.process_input_line('import matplotlib.pyplot as plt',

store_history=False)

self._pyplot_imported = True

def process_pure_python(self, content):

"""

output = []

savefig = False # keep up with this to clear figure

multiline = False # to handle line continuation

multiline_start = None

fmtin = self.promptin

ct = 0

for lineno, line in enumerate(content):

line_stripped = line.strip()

if not len(line):

output.append(line)

continue

# handle pseudo-decorators, whilst ensuring real python decorators are treated as input

if any(

line_stripped.startswith("@" + pseudo_decorator)

for pseudo_decorator in PSEUDO_DECORATORS

):

output.extend([line])

if 'savefig' in line:

savefig = True # and need to clear figure

continue

# handle comments

if line_stripped.startswith('#'):

output.extend([line])

continue

# deal with lines checking for multiline

continuation = u' %s:'% ''.join(['.']*(len(str(ct))+2))

if not multiline:

modified = u"%s %s" % (fmtin % ct, line_stripped)

output.append(modified)

ct += 1

try:

ast.parse(line_stripped)

output.append(u'')

except Exception: # on a multiline

multiline = True

multiline_start = lineno

else: # still on a multiline

modified = u'%s %s' % (continuation, line)

output.append(modified)

# if the next line is indented, it should be part of multiline

if len(content) > lineno + 1:

nextline = content[lineno + 1]

if len(nextline) - len(nextline.lstrip()) > 3:

continue

try:

mod = ast.parse(

'\n'.join(content[multiline_start:lineno+1]))

if isinstance(mod.body[0], ast.FunctionDef):

# check to see if we have the whole function

for element in mod.body[0].body:

if isinstance(element, ast.Return):

multiline = False

else:

output.append(u'')

multiline = False

except Exception:

pass

if savefig: # clear figure if plotted

self.ensure_pyplot()

self.process_input_line('plt.clf()', store_history=False)

self.clear_cout()

savefig = False

return output

def custom_doctest(self, decorator, input_lines, found, submitted):

"""

from .custom_doctests import doctests

args = decorator.split()

doctest_type = args[1]

if doctest_type in doctests:

doctests[doctest_type](self, args, input_lines, found, submitted)

else:

e = "Invalid option to @doctest: {0}".format(doctest_type)

has_content: bool = True

required_arguments: int = 0

optional_arguments: int = 4 # python, suppress, verbatim, doctest

final_argumuent_whitespace: bool = True

option_spec: Dict[str, Any] = {

"python": directives.unchanged,

"suppress": directives.flag,

"verbatim": directives.flag,

"doctest": directives.flag,

"okexcept": directives.flag,

"okwarning": directives.flag,

}

shell = None

seen_docs: Set = set()

def get_config_options(self):

# contains sphinx configuration variables

config = self.state.document.settings.env.config

# get config variables to set figure output directory

savefig_dir = config.ipython_savefig_dir

source_dir = self.state.document.settings.env.srcdir

savefig_dir = os.path.join(source_dir, savefig_dir)

# get regex and prompt stuff

rgxin = config.ipython_rgxin

rgxout = config.ipython_rgxout

warning_is_error= config.ipython_warning_is_error

promptin = config.ipython_promptin

promptout = config.ipython_promptout

mplbackend = config.ipython_mplbackend

exec_lines = config.ipython_execlines

hold_count = config.ipython_holdcount

return (savefig_dir, source_dir, rgxin, rgxout,

promptin, promptout, mplbackend, exec_lines, hold_count, warning_is_error)

def setup(self):

# Get configuration values.

(savefig_dir, source_dir, rgxin, rgxout, promptin, promptout,

mplbackend, exec_lines, hold_count, warning_is_error) = self.get_config_options()

try:

os.makedirs(savefig_dir)

except OSError as e:

if e.errno != errno.EEXIST:

raise

if self.shell is None:

# We will be here many times. However, when the

# EmbeddedSphinxShell is created, its interactive shell member

# is the same for each instance.

if mplbackend and 'matplotlib.backends' not in sys.modules and use_matplotlib:

import matplotlib

matplotlib.use(mplbackend)

# Must be called after (potentially) importing matplotlib and

# setting its backend since exec_lines might import pylab.

self.shell = EmbeddedSphinxShell(exec_lines)

# Store IPython directive to enable better error messages

self.shell.directive = self

# reset the execution count if we haven't processed this doc

#NOTE: this may be borked if there are multiple seen_doc tmp files

#check time stamp?

if self.state.document.current_source not in self.seen_docs:

self.shell.IP.history_manager.reset()

self.shell.IP.execution_count = 1

self.seen_docs.add(self.state.document.current_source)

# and attach to shell so we don't have to pass them around

self.shell.rgxin = rgxin

self.shell.rgxout = rgxout

self.shell.promptin = promptin

self.shell.promptout = promptout

self.shell.savefig_dir = savefig_dir

self.shell.source_dir = source_dir

self.shell.hold_count = hold_count

self.shell.warning_is_error = warning_is_error

# setup bookmark for saving figures directory

self.shell.process_input_line(

'bookmark ipy_savedir "%s"' % savefig_dir, store_history=False

)

self.shell.clear_cout()

return rgxin, rgxout, promptin, promptout

def teardown(self):

# delete last bookmark

self.shell.process_input_line('bookmark -d ipy_savedir',

store_history=False)