-
Notifications
You must be signed in to change notification settings - Fork 3
/
exec.py
75 lines (52 loc) · 2.38 KB
/
exec.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
# Copyright (c) 2017 Civic Knowledge. This file is licensed under the terms of the
# Revised BSD License, included in this distribution as LICENSE
"""
Functions for executing Jupyter notebooks
"""
def x_execute(nb_path, env):
"""Convert the notebook to a python script and execute it, returning the local context
as a dict"""
from nbconvert.exporters import get_exporter
preprocessors = ['metatab.jupyter.preprocessors.RemoveMagics']
exporter = get_exporter('python')(preprocessors=preprocessors)
(script, notebook) = exporter.from_filename(filename=nb_path)
exec(compile(script.replace('# coding: utf-8', ''), 'script', 'exec'), env)
return env
def execute_notebook(nb_path, pkg_dir, dataframes, write_notebook=False):
"""
Execute a notebook after adding the prolog and epilog. Can also add %mt_materialize magics to
write dataframes to files
:param nb_path: path to a notebook.
:param pkg_dir: Directory to which dataframes are materialized
:param dataframes: List of names of dataframes to materialize
:return: a Notebook object
"""
import nbformat
from metapack.jupyter.preprocessors import AddEpilog
from metapack.jupyter.exporters import ExecutePreprocessor, Config
from os.path import dirname, join, splitext, basename
from nbconvert.preprocessors.execute import CellExecutionError
with open(nb_path) as f:
nb = nbformat.read(f, as_version=4)
root, ext = splitext(basename(nb_path))
c = Config()
nb, resources = AddEpilog(config=c, pkg_dir=pkg_dir,
dataframes=dataframes
).preprocess(nb, {})
try:
ep = ExecutePreprocessor(config=c)
nb, _ = ep.preprocess(nb, {'metadata': {'path': dirname(nb_path)}})
except CellExecutionError as e:
err_nb_path = join(dirname(nb_path), root + '-errors' + ext)
with open(err_nb_path, 'wt') as f:
nbformat.write(nb, f)
raise CellExecutionError("Errors executing noteboook. See notebook at {} for details.\n{}"
.format(err_nb_path, ''))
if write_notebook:
if write_notebook is True:
exec_nb_path = join(dirname(nb_path), root + '-executed' + ext)
else:
exec_nb_path = write_notebook
with open(exec_nb_path, 'wt') as f:
nbformat.write(nb, f)
return nb