Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
/* indent single paragraph */
div.admonition {
text-indent: 20px;
}
/* don't indent multiple paragraphs */
div.admonition > p {
text-indent: 0;
}
/* remove excessive padding */
div.admonition.inline-title p.admonition-title {
padding-left: .2em;
}
"""
class Exporter(nbconvert.RSTExporter):
"""Convert Jupyter notebooks to reStructuredText.
Uses nbconvert to convert Jupyter notebooks to a reStructuredText
string with custom reST directives for input and output cells.
Notebooks without output cells are automatically executed before
conversion.
"""
def __init__(self, execute='auto', kernel_name='', execute_arguments=[],
allow_errors=False, timeout=30, codecell_lexer='none'):
"""Initialize the Exporter."""
# NB: The following stateful Jinja filters are a hack until
# template-based processing is dropped
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 apply_preprocessors(preprocessors, nbname):
notebooks_path = os.path.join(os.path.split(__file__)[0], 'notebooks')
with open(os.path.join(notebooks_path, nbname)) as f:
nb = nbformat.read(f, nbformat.NO_CONVERT)
exporter = nbconvert.PythonExporter()
for preprocessor in preprocessors:
exporter.register_preprocessor(preprocessor)
source, meta = exporter.from_notebook_node(nb)
return source
def _tst_notebook(self, notebook_name):
notebook_filename = os.path.abspath(os.path.join(
self.this_file_directory, '..', '..', 'examples', notebook_name))
with open(notebook_filename) as f:
nb = nbformat.read(f, as_version=4)
python_nb, metadata = export(PythonExporter, nb)
# Remove magic lines manually
python_nb = '\n'.join([
line for line in python_nb.split('\n')
if 'get_ipython().run_line_magic(' not in line
])
exec(python_nb)
print(
"\n---------------------"
" Testing {0}.ipynb "
"---------------------".format(nbname)
)
if (
(nbname in self.ignore) or
(nbname in self.py2_ignore and sys.version_info[0] == 2)
):
print(" Skipping {}".format(nbname))
return
run_path = os.path.sep.join(nbpath.split(os.path.sep)[:-1])
os.chdir(run_path)
ep = ClearOutputPreprocessor(
resources={'metadata': {'path': run_path}}
)
with open(nbpath) as f:
nb = nbformat.read(f, as_version=4)
ep.preprocess(nb, {})
ex = ExecutePreprocessor(
timeout=timeout,
kernel_name='python{}'.format(sys.version_info[0]),
allow_errors=True,
resources={'metadata': {'path': run_path}}
)
out = ex.preprocess(nb, {})
def run_test(project, zone, cluster, new_values): # pylint: disable=too-many-locals
# TODO(jeremy@lewi.us): Need to configure the notebook and test to build
# using GCB.
dirname = os.path.dirname(__file__)
if not dirname:
logging.info("__file__ doesn't apper to be absolute path.")
dirname = os.getcwd()
notebook_path = os.path.join(dirname, "TF on GKE.ipynb")
logging.info("Reading notebook %s", notebook_path)
if not os.path.exists(notebook_path):
raise ValueError("%s does not exist" % notebook_path)
with open(notebook_path) as hf:
node = nbformat.read(hf, nbformat.NO_CONVERT)
exporter = nbconvert.PythonExporter()
raw, _ = nbconvert.export(exporter, node)
credentials = GoogleCredentials.get_application_default()
gke = discovery.build("container", "v1", credentials=credentials)
lines = raw.splitlines()
modified = replace_vars(lines, new_values)
modified = strip_appendix(modified)
modified = strip_unexecutable(modified)
with tempfile.NamedTemporaryFile(suffix="notebook.py", prefix="tmpGke",
mode="w", delete=False) as hf:
code_path = hf.name
hf.write("\n".join(modified))
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)]