Job.py

from IPython.lib.backgroundjobs import BackgroundJobManager
from IPython.core.magic         import register_line_magic
from IPython.display            import display
from IPython                    import get_ipython
from itertools                  import islice, chain
from ipywidgets                 import IntProgress, HTML, VBox

# Iterate through a generator without peaking, yeilding a list of elements
#
def chunks(iterable, size):
    iterator = iter(iterable)
    for first in iterator:
        yield list(chain([first], islice(iterator, (len(iterable)//size) - 1)))

# Magics for interacting with jobs
#
def jobs_manager():
    jobs = BackgroundJobManager()
    
    @register_line_magic
    def job(line):
        ip = get_ipython()
        jobs.new(line, ip.user_global_ns)
        
    return jobs

# Pretty progress bars for for jupyter notebook 
#
def log_progress(sequence, every=None, size=None, name='Jobs'):
    is_iterator = False
    if size is None:
        try:
            size = len(sequence)
        except TypeError:
            is_iterator = True
    if size is not None:
        if every is None:
            if size <= 200:
                every = 1
            else:
                every = int(size / 200)     # every 0.5%
    else:
        assert every is not None, 'sequence is iterator, set every'

    if is_iterator:
        progress = IntProgress(min=0, max=1, value=1)
        progress.bar_style = 'info'
    else:
        progress = IntProgress(min=0, max=size, value=0)
    label = HTML()
    box = VBox(children=[label, progress])
    display(box)

    index = 0
    try:
        for index, record in enumerate(sequence, 1):
            if index == 1 or index % every == 0:
                if is_iterator:
                    label.value = '{name}: {index} / ?'.format(
                        name=name,
                        index=index
                    )
                else:
                    progress.value = index
                    label.value = u'{name}: {index} / {size}'.format(
                        name=name,
                        index=index,
                        size=size
                    )
            yield record
    except:
        progress.bar_style = 'danger'
        raise
    else:
        progress.bar_style = 'success'
        progress.value = index
        label.value = "{name}: {index}".format(
            name=name,
            index=str(index or '?')
        )

Threading Example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy             as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Initialize a new job manager\n",
    "from Job import *\n",
    "job = jobs_manager()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_threads = 2\n",
    "\n",
    "# An image to hold the output\n",
    "size        = 2**12\n",
    "data        = np.zeros((size, size, 3))\n",
    "\n",
    "# Diagonal gradient kernel, intentionally slow\n",
    "def test(x,y):\n",
    "    data[y, x, :] = ((y / size) + (x / size)) / 2\n",
    "\n",
    "# List of job inputs\n",
    "pixels = [(y, x) for y in range(size) for x in range(size)]\n",
    "\n",
    "# Process chunks of the job list on separate threads\n",
    "for chunk in chunks(pixels, num_threads):\n",
    "    \n",
    "    # Use interactive python to parse line from text on new thread\n",
    "    %job [test(x,y) for x,y in log_progress(chunk)]\n",
    "    \n",
    "    # Use lambda to defer execution to new thread\n",
    "    #job.new(lambda : [test(_) for _ in log_progress(chunk)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job.status()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.axis('off')\n",
    "plt.imshow(data)\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Clear the job list of dead and completed jobs\n",
    "job.flush()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This job causes an error because of invalid input to test(.,.)\n",
    "%job [test(_) for _ in log_progress(range(10, 30))]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job.status()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job.traceback(0)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}