Chris Hager
Programming, Technology & More

Python Utilities by Peter Norvig

Peter Norvig, the famous American computer scientist and Director of Research at Google Inc., participated in this year’s Advent of Code (a series of small programming puzzles), and shared his experience in an interesting blog post.

The post starts with this amazing collection of Python utility functions, which may also be useful for your next project:

# Python 3.x
import re
import numpy as np
import math
import urllib.request

from collections import Counter, defaultdict, namedtuple, deque
from functools   import lru_cache
from itertools   import permutations, combinations, chain, cycle, product
from heapq       import heappop, heappush

def Input(day):
    "Open this day's input file."
    filename = 'advent2016/input{}.txt'.format(day)
    try:
        return open(filename)
    except FileNotFoundError:
        urllib.request.urlopen("http://norvig.com/ipython/" + filename)

def transpose(matrix): return zip(*matrix)

def first(iterable): return next(iter(iterable))

def firsttrue(iterable): return first(it for it in iterable if it)

def counttrue(iterable): return sum(bool(it) for it in iterable)

cat = ''.join

Ø   = frozenset() # Empty set
inf = float('inf')
BIG = 10 ** 999

def grep(pattern, lines):
    "Print lines that match pattern."
    for line in lines:
        if re.search(pattern, line):
            print(line)

def groupby(iterable, key=lambda it: it):
    "Return a dic whose keys are key(it) and whose values are all the elements of iterable with that key."
    dic = defaultdict(list)
    for it in iterable:
        dic[key(it)].append(it)
    return dic

def powerset(iterable):
    "Yield all subsets of items."
    items = list(iterable)
    for r in range(len(items)+1):
        for c in combinations(items, r):
            yield c

# 2-D points implemented using (x, y) tuples
def X(point): return point[0]
def Y(point): return point[1]

def neighbors4(point):
    "The four neighbors (without diagonals)."
    x, y = point
    return ((x+1, y), (x-1, y), (x, y+1), (x, y-1))

def neighbors8(point):
    "The eight neifhbors (with diagonals)."
    x, y = point
    return ((x+1, y), (x-1, y), (x, y+1), (x, y-1),
            (X+1, y+1), (x-1, y-1), (x+1, y-1), (x-1, y+1))

def cityblock_distance(p, q=(0, 0)):
    "City block distance between two points."
    return abs(X(p) - X(q)) + abs(Y(p) - Y(q))

def euclidean_distance(p, q=(0, 0)):
    "Euclidean (hypotenuse) distance between two points."
    return math.hypot(X(p) - X(q), Y(p) - Y(q))

def trace1(f):
    "Print a trace of the input and output of a function on one line."
    def traced_f(*args):
        result = f(*args)
        print('{} = {}'.format(_callstr(f, args), result))
        return result
    return traced_f

def trace(f):
    "Print a trace of the call and args on one line, and the return on another."
    def traced_f(*args):
        print(_callstr(f, args))
        trace.indent += 1
        try:
            result = f(*args)
        finally:
            trace.indent -= 1
        print('{} = {}'.format(_callstr(f, args), result))
        return result
    return traced_f
trace.indent = 0

def _callstr(f, args):
    "Return a string representing f(*args)."
    return '{}{}({})'.format('> ' * trace.indent, f.__name__, ', '.join(map(str, args)))

def astar_search(start, h_func, move_func):
    "Find a shortest sequence of states from start to a goal state (a state s with h_func(s) == 0)."
    frontier  = [(h_func(start), start)] # A priority queue, ordered by path length, f = g + h
    previous  = {start: None}  # start state has no previous state; other states will
    path_cost = {start: 0}     # The cost of the best path to a state.
    while frontier:
        (f, s) = heappop(frontier)
        if h_func(s) == 0:
            return Path(previous, s)
        for s2 in move_func(s):
            new_cost = path_cost[s] + 1
            if s2 not in path_cost or new_cost < path_cost[s2]:
                heappush(frontier, (new_cost + h_func(s2), s2))
                path_cost[s2] = new_cost
                previous[s2] = s
    return dict(fail=True, front=len(frontier), prev=len(previous))

def Path(previous, s):
    "Return a list of states that lead to state s, according to the previous dict."
    return ([] if (s is None) else Path(previous, previous[s]) + [s])

Read the full post for more interesting ideas and elegant solutions.


Python Thread Pool

A thread pool is a group of pre-instantiated, idle threads which stand ready to be given work. These are often preferred over instantiating new threads for each task when there is a large number of (short) tasks to be done rather than a small number of long ones.

Suppose you want do download 1000s of documents from the internet, but only have resources for downloading 50 at a time. The solution is to utilize is a thread pool, spawning a fixed number of threads to download all the URLs from a queue, 50 at a time.

In order to use thread pools, Python 3.x includes the ThreadPoolExecutor class, and both Python 2.x and 3.x have multiprocessing.dummy.ThreadPool. multiprocessing.dummy replicates the API of multiprocessing but is no more than a wrapper around the threading module.

The downside of multiprocessing.dummy.ThreadPool is that in Python 2.x, it is not possible to exit the program with eg. a KeyboardInterrupt before all tasks from the queue have been finished by the threads.

In order to achieve an interruptable thread queue in Python 2.x and 3.x (for use in PDFx), I’ve build this code, inspired by stackoverflow.com/a/7257510. It implements a thread pool which works with Python 2.x and 3.x:

import sys
IS_PY2 = sys.version_info < (3, 0)

if IS_PY2:
    from Queue import Queue
else:
    from queue import Queue

from threading import Thread


class Worker(Thread):
    """ Thread executing tasks from a given tasks queue """
    def __init__(self, tasks):
        Thread.__init__(self)
        self.tasks = tasks
        self.daemon = True
        self.start()

    def run(self):
        while True:
            func, args, kargs = self.tasks.get()
            try:
                func(*args, **kargs)
            except Exception as e:
                # An exception happened in this thread
                print(e)
            finally:
                # Mark this task as done, whether an exception happened or not
                self.tasks.task_done()


class ThreadPool:
    """ Pool of threads consuming tasks from a queue """
    def __init__(self, num_threads):
        self.tasks = Queue(num_threads)
        for _ in range(num_threads):
            Worker(self.tasks)

    def add_task(self, func, *args, **kargs):
        """ Add a task to the queue """
        self.tasks.put((func, args, kargs))

    def map(self, func, args_list):
        """ Add a list of tasks to the queue """
        for args in args_list:
            self.add_task(func, args)

    def wait_completion(self):
        """ Wait for completion of all the tasks in the queue """
        self.tasks.join()


if __name__ == "__main__":
    from random import randrange
    from time import sleep

    # Function to be executed in a thread
    def wait_delay(d):
        print("sleeping for (%d)sec" % d)
        sleep(d)

    # Generate random delays
    delays = [randrange(3, 7) for i in range(50)]

    # Instantiate a thread pool with 5 worker threads
    pool = ThreadPool(5)

    # Add the jobs in bulk to the thread pool. Alternatively you could use
    # `pool.add_task` to add single jobs. The code will block here, which
    # makes it possible to cancel the thread pool with an exception when
    # the currently running batch of workers is finished.
    pool.map(wait_delay, delays)
    pool.wait_completion()

The queue size is similar to the number of threads (see self.tasks = Queue(num_threads)), therefore adding tasks with pool.map(..) and pool.add_task(..) blocks until a new slot in the Queue is available.

When you issue a KeyboardInterrupt by pressing Ctrl+C, the current batch of workers will finish and the program quits with the exception at the pool.map(..) step.


If you have suggestions or feedback, let me know via @metachris


How to install Qt 5.6 and PyQt5 in a Python 3.4 virtual environment on Mac OS X and Linux

This is a simple guide on installing the latest Qt (currently 5.6) and PyQt5 on Mac OS X 10.11 (El Capitan) and Linux with Python 3.4, inside a virtual environment.

Installation Steps

  • Python 3
  • Xcode and command-line tools
  • Qt libraries
  • Virtual environment
  • SIP Python package
  • PyQt5 Python package

Python 3

First of all, make sure that Python 3 is available on your system. You can easily check this by opening the terminal and entering the command python3. If you need to install it, check out the Python homepage, or install it with homebrew (brew install python3) on OS X or your favorite Linux package manager.

Install Xcode and command-line tools

If you are using OS X, download Xcode and install it. Then install the command-line tools by entering the following command in the terminal: xcode-select --install. This adds a number of tools to your system, such as make, git, gcc, c++ and g++.

Install Qt Libraries

First we need to download and install the Qt libraries:

I’d recommend to install Qt into the directory /opt/qt. The installation requires about 14 GB of disk space, and includes a number of apps and utilities:

  • Qt Creator.app - a complete IDE with a graphical GUI designer and code editor (more)
  • 5.6/clang_64/bin/Designer.app - the GUI designer
  • 5.6/clang_64/bin/pixeltool.app - a tool to inspect the pixels around the mouse cursor
  • 5.6/clang_64/bin/qmlscene - execute scenes from .QML files
  • 5.6/clang_64/bin/qtdiag - Prints diagnostic output about the Qt library
  • Various others such as qmllint, qmlmin, qmlplugindump, qmlprofiler, qmlscene, qmltestrunner

Furthermore the Qt installation includes a number of examples in the Examples subdirectory.

Create a virtualenv for the PyQt5 and SIP libs

For this guide, we create a virtual environment with Python 3.4 under the home directory in ~/.venv/qtproject:

# Create the directory
$ mkdir -p ~/.venv

# Create the virtual environment
$ python3 -m venv ~/.venv/qtproject

# Activate the virtual environment
$ . ~/.venv/qtproject/bin/activate

At this point, typing the command which python3 should output something like ~/.venv/qtproject/bin/python3.

Install SIP

PyQt requires to have the SIP module installed. SIP is a tool for automatically generating Python bindings for C and C++ libraries.

You can either download the .tar.gz file, or install the latest from the source repository with mercurial (hg):

# Clone the source code
$ cd /tmp/
$ hg clone http://www.riverbankcomputing.com/hg/sip
$ cd sip

# Generate the build configuration
$ python2.7 build.py prepare  # build.py is a Python 2 script
$ python3.4 configure.py -d ~/.venv/qtproject/lib/python3.4/site-packages

# Make and install
$ make
$ sudo make install
$ sudo make clean

Install PyQt5

Finally we get to install the PyQt5 module.

Start by downloading the tar.gz file, in this case PyQt-gpl-5.5.1.tar.gz, and extracting it:

# Extract the tar.gz file
$ tar -xvf PyQt-gpl-5.5.1.tar.gz

# Change into the PyQt source directory
$ cd PyQt-gpl-5.5.1

# Generate the build configuration (make sure to reference 'qmake' from the Qt libs installation directory)
$ python3 configure.py --destdir ~/.venv/qtproject/lib/python3.4/site-packages --qmake /opt/qt/5.6/clang_64/bin/qmake

# Make and install
$ make  # this takes a really long time
$ sudo make install
$ sudo make clean

All Done!

At this point, everything is successfully installed! Now let’s check if everything works by importing PyQt5 from Python 3.4:

~/.venv/qtproject/bin/python3 -c "import PyQt5"

And just for the sake of it, let’s build a simple hello world Qt application:

from PyQt5.QtWidgets import QApplication, QWidget, QLabel


if __name__ == "__main__":
    app = QApplication()

    # Build the window widget
    w = QWidget()
    w.setGeometry(300, 300, 250, 150)  # x, y, w, h
    w.setWindowTitle("My First Qt App")

    # Add a label with tooltip
    label = QLabel("Hello World  🚀", w)
    label.setToolTip("This is a <b>QLabel</b> widget with Tooltip")
    label.resize(label.sizeHint())
    label.move(80, 50)

    # Show window and run
    w.show()
    app.exec_()

Save this program as helloqt.py and execute it with python3 helloqt.py (or, if the virtualenv is not activated, with ~/.venv/qtproject/bin/python3 helloqt.py), and be greeted with this:

🎉  Congratulations  🎉


Now have fun with some GUI programming! Here are a few useful next steps:


If you have suggestions or feedback, let me know via @metachris

Update 2016-03-25: python3 -m venv instead of virtualenv

Blog Archive
swirl