Execute notebooks

This notebook executes all phconvert notebooks to automate testing.

In [ ]:
import os
from pathlib import Path

import nbformat
from nbconvert.preprocessors import ExecutePreprocessor
from IPython.display import display, FileLink


def run_notebook(in_filepath, run_path=None, out_dir=None, out_suffix=''):
    """Runs the notebook `notebook_name` (file name with no extension).

    This function executes notebook with name `notebook_name` (no extension)
    and saves the fully executed notebook in a new file appending "-out"
    to the original file name.

    It also displays links to the original and executed notebooks.
    """
    if not in_filepath.is_file():
        raise IOError('File "%s" not found.' % in_filepath)
    in_filepath = in_filepath.resolve()
    
    if run_path is None:
        run_path = str(in_filepath.parent)
    
    if out_dir is None:
        out_dir = in_filepath.parent
    else:
        out_dir = Path(out_dir).resolve()
       
    out_filepath = Path(out_dir, "%s%s.ipynb" % (in_filepath.stem, out_suffix))
    
    nb = nbformat.read(in_filepath.open(), as_version=4)
    ep = ExecutePreprocessor(timeout = 3600)

    try:
        out = ep.preprocess(nb, {'metadata': {'path': run_path}})
    except Exception:
        msg = 'Error executing the notebook "%s".\n\n' % in_filepath
        msg += 'See notebook "%s" for the traceback.' % out_filepath
        print(msg)
        raise
    finally:
        nbformat.write(nb, out_filepath.open(mode='wt'))
In [ ]:
## Monkey patch needed until nbconvert > 4.0.0 is released
## https://github.com/jupyter/nbconvert/releases

from nbformat.v4 import output_from_msg
from nbconvert.preprocessors.execute import CellExecutionError
from textwrap import dedent

def preprocess_cell(self, cell, resources, cell_index):
    """
    Apply a transformation on each code cell. See base.py for details.
    """
    if cell.cell_type != 'code':
        return cell, resources

    outputs = self.run_cell(cell)
    cell.outputs = outputs

    if not self.allow_errors:
        for out in outputs:
            if out.output_type == 'error':
                pattern = """\
                An error occurred while executing the following cell:
                ------------------
                {cell.source}
                ------------------

                {out.ename}: {out.evalue}
                """
                msg = dedent(pattern).format(out=out, cell=cell)
                raise CellExecutionError(msg)            
    return cell, resources

def run_cell(self, cell):
    msg_id = self.kc.execute(cell.source)
    self.log.debug("Executing cell:\n%s", cell.source)
    # wait for finish, with timeout
    while True:
        try:
            msg = self.kc.shell_channel.get_msg(timeout=self.timeout)
        except Empty:
            self.log.error("""Timeout waiting for execute reply (%is).
            If your cell should take longer than this, you can increase the timeout with:

                c.ExecutePreprocessor.timeout = SECONDS

            in jupyter_nbconvert_config.py
            """ % self.timeout)
            if self.interrupt_on_timeout:
                self.log.error("Interrupting kernel")
                self.km.interrupt_kernel()
                break
            else:
                try:
                    exception = TimeoutError
                except NameError:
                    exception = RuntimeError
                raise exception("Cell execution timed out, see log"
                                " for details.")

        if msg['parent_header'].get('msg_id') == msg_id:
            break
        else:
            # not our reply
            continue

    outs = []

    while True:
        try:
            msg = self.kc.iopub_channel.get_msg(timeout=self.timeout)
        except Empty:
            self.log.warn("Timeout waiting for IOPub output")
            break
        if msg['parent_header'].get('msg_id') != msg_id:
            # not an output from our execution
            continue

        msg_type = msg['msg_type']
        self.log.debug("output: %s", msg_type)
        content = msg['content']

        # set the prompt number for the input and the output
        if 'execution_count' in content:
            cell['execution_count'] = content['execution_count']

        if msg_type == 'status':
            if content['execution_state'] == 'idle':
                break
            else:
                continue
        elif msg_type == 'execute_input':
            continue
        elif msg_type == 'clear_output':
            outs = []
            continue
        elif msg_type.startswith('comm'):
            continue

        try:
            out = output_from_msg(msg)
        except ValueError:
            self.log.error("unhandled iopub msg: " + msg_type)
        else:
            outs.append(out)

    return outs

ExecutePreprocessor.preprocess_cell = preprocess_cell
ExecutePreprocessor.run_cell = run_cell
In [ ]:
%pwd
In [ ]:
pathlist = list(f for f in Path('../notebooks').glob('*.ipynb') if not f.name.startswith('_'))
pathlist
In [ ]:
for nbpath in pathlist:
    print(nbpath.stem)
    run_notebook(nbpath, out_dir=Path(nbpath.parent, 'out'))