Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
working_dir = os.path.join(*([temp_dir] + notebook.split('/')))
if no_cache == '1':
logging.info("Cleaning and setting up temp directory '{}'".format(working_dir))
shutil.rmtree(temp_dir, ignore_errors=True)
errors = []
notebook = None
if not os.path.isdir(working_dir):
os.makedirs(working_dir)
try:
notebook = nbformat.read(notebook_path + '.ipynb', as_version=IPYTHON_VERSION)
if kernel is not None:
eprocessor = ExecutePreprocessor(timeout=TIME_OUT, kernel_name=kernel)
else:
eprocessor = ExecutePreprocessor(timeout=TIME_OUT)
success = False
# There is a low (< 1%) chance that starting a notebook executor will fail due to the kernel
# taking to long to start, or a port collision, etc.
for i in range(ATTEMPTS):
try:
nb, _ = eprocessor.preprocess(notebook, {'metadata': {'path': working_dir}})
success = True
except RuntimeError as rte:
# We check if the exception has to do with the Jupyter kernel failing to start. If
# not, we rethrow to prevent the notebook from erring ATTEMPTS times. It is not
# ideal to inspect the exception message, but necessary for retry logic, as Jupyter
# client throws the generic RuntimeError that can be confused with other Runtime
# errors.
if str(rte) != KERNEL_ERROR_MSG:
raise rte
def run_notebook(self, notebook_filename):
"""Copy to a sandbox"""
nb_dir, nb_name = os.path.split(notebook_filename)
sandboxed_nb = os.path.join(self.sandboxdir, nb_name)
shutil.copy2(notebook_filename, sandboxed_nb)
with open(notebook_filename) as f:
nb = nbformat.read(f, as_version=4)
ep = ExecutePreprocessor(timeout=600, kernel_name='python3')
ep.extra_arguments = ['--Application.log_level=0']
print("Executing notebook %s in %s" % (notebook_filename, self.sandboxdir))
ep.preprocess(nb, {'metadata': {'path': self.sandboxdir}})
def test_timeseries_controls(self):
nb_path = Path(_NB_FOLDER).joinpath(_NB_NAME)
abs_path = Path(_NB_FOLDER).absolute()
with open(nb_path) as f:
nb = nbformat.read(f, as_version=4)
ep = ExecutePreprocessor(timeout=600, kernel_name="python3")
try:
ep.preprocess(nb, {"metadata": {"path": abs_path}})
except CellExecutionError:
nb_err = str(nb_path).replace(".ipynb", "-err.ipynb")
msg = f"Error executing the notebook '{nb_path}'.\n"
msg += f"See notebook '{nb_err}' for the traceback."
print(msg)
with open(nb_err, mode="w", encoding="utf-8") as f:
nbformat.write(nb, f)
raise
def _preproc():
pythonkernel = 'python' + str(sys.version_info[0])
return ExecutePreprocessor(timeout=300, kernel_name=pythonkernel,
interrupt_on_timeout=True)
passing = True
print("\n--------------- Testing {0} ---------------".format(nbname))
print(" {0}".format(nbpath))
if nbname in py2Ignore and sys.version_info[0] == 2:
print(" Skipping {}".format(nbname))
return
ep = ClearOutputPreprocessor()
with open(nbpath) as f:
nb = nbformat.read(f, as_version=4)
ep.preprocess(nb, {})
ex = ExecutePreprocessor(
timeout=600,
kernel_name='python{}'.format(sys.version_info[0]),
allow_errors=True
)
out = ex.preprocess(nb, {})
for cell in out[0]['cells']:
if 'outputs' in cell.keys():
for output in cell['outputs']:
if output['output_type'] == 'error':
passing = False
err_msg = []
for o in output['traceback']:
err_msg += ["{}".format(o)]
def execute_notebook(path):
notebook = nbformat.read(path, as_version=4)
notebook.cells.extend([unittest_cell(), doctest_cell()])
processor = ExecutePreprocessor(timeout=-1, kernel_name='python3')
processor.preprocess(notebook, metadata(path))
return parse_test_result(notebook.cells)
cell_code = nbformat.v4.new_code_cell(source=cell_code.strip())
nb['cells'].append(cell_code)
# substitute global variables and disable OpenGL/Mayavi GUI
cell_separator = '\n##{}\n'.format(uuid.uuid4().hex)
src = cell_separator.join(get_code_cells(nb))
parameters = dict(x.split('=', 1) for x in new_values)
src = substitute_variable_values(src, strings_as_is=True, keep_original=False,
**parameters)
src_no_gui = mock_es_visualization(src)
# update notebook with new code
set_code_cells(nb, src_no_gui.split(cell_separator))
# execute notebook
ep = ExecutePreprocessor(timeout=20 * 60, kernel_name='python3')
ep.preprocess(nb, {'metadata': {'path': notebook_dirname}})
# restore notebook with code before the GUI removal step
set_code_cells(nb, src.split(cell_separator))
# write edited notebook
with open(notebook_filepath_edited, 'w', encoding='utf-8') as f:
nbformat.write(nb, f)
def _notebook_run(path, SCOPETYPE='OPENADC', PLATFORM='CWLITEARM', **kwargs):
"""Execute a notebook via nbconvert and collect output.
:returns (parsed nb object, execution errors)
"""
html_path = Path("html/" + path + "-{}-{}".format(SCOPETYPE,PLATFORM) + ".html")
real_path = Path(path)
with open(real_path) as nbfile:
nb = nbformat.read(nbfile, as_version=4)
orig_parameters = extract_parameters(nb)
params = parameter_values(orig_parameters, SCOPETYPE=SCOPETYPE, PLATFORM=PLATFORM, **kwargs)
new_nb = replace_definitions(nb, params, execute=False)
ep = ExecutePreprocessor(timeout=None, kernel_name='python3', allow_errors=True)
ep.preprocess(new_nb, {'metadata': {'path': './'}})
errors = [[i+1,output] for i,cell in enumerate(new_nb.cells) if "outputs" in cell
for output in cell["outputs"]\
if output.output_type == "error"]
with open(html_path, "w", encoding='utf-8') as html_file:
html_exporter = HTMLExporter()
body, res = html_exporter.from_notebook_node(new_nb)
body = ansi2html(body)
html_file.write(body)
from __future__ import unicode_literals, print_function
from nbconvert.preprocessors import ExecutePreprocessor
from nbconvert.preprocessors.execute import CellExecutionError
from traitlets import Bool, Instance
class PapermillExecutePreprocessor(ExecutePreprocessor):
"""Module containing a preprocessor that executes the code cells
and updates outputs"""
log_output = Bool(False).tag(config=True)
stdout_file = Instance(object, default_value=None).tag(config=True)
stderr_file = Instance(object, default_value=None).tag(config=True)
def preprocess(self, nb_man, resources, km=None):
"""
Wraps the parent class process call slightly
"""
with self.setup_preprocessor(nb_man.nb, resources, km=km):
if self.log_output:
self.log.info("Executing notebook with kernel: {}".format(self.kernel_name))
nb, resources = self.papermill_process(nb_man, resources)
info_msg = self._wait_for_reply(self.kc.kernel_info())
execute = nbsphinx_metadata.get('execute', self._execute)
if execute not in ('always', 'never', 'auto'):
raise ValueError('invalid execute option: {!r}'.format(execute))
auto_execute = (
execute == 'auto' and
# At least one code cell actually containing source code:
any(c.source for c in nb.cells if c.cell_type == 'code') and
# No outputs, not even a prompt number:
not any(c.get('outputs') or c.get('execution_count')
for c in nb.cells if c.cell_type == 'code')
)
if auto_execute or execute == 'always':
allow_errors = nbsphinx_metadata.get(
'allow_errors', self._allow_errors)
timeout = nbsphinx_metadata.get('timeout', self._timeout)
pp = nbconvert.preprocessors.ExecutePreprocessor(
kernel_name=self._kernel_name,
extra_arguments=self._execute_arguments,
allow_errors=allow_errors, timeout=timeout)
nb, resources = pp.preprocess(nb, resources)
# Call into RSTExporter
rststr, resources = super(Exporter, self).from_notebook_node(
nb, resources, **kw)
orphan = nbsphinx_metadata.get('orphan', False)
if orphan is True:
resources['nbsphinx_orphan'] = True
elif orphan is not False:
raise ValueError('invalid orphan option: {!r}'.format(orphan))
if 'application/vnd.jupyter.widget-state+json' in nb.metadata.get(