Some live coding with python

Watching for changes

The first thing a live coding system needs is a way to trigger reloading of the source when something has changed.

In the case of this system, it will be built around files on disk. That means we want to know when one of the files has changed.

Python doesn't have any native support for doing that efficently. The best you could do would be to repeatedly check the modification data of interesting files in a polling loop with a delay. But that would be unresponsive and inefficient.

However, most platforms have native services for making this kind of file watching more efficent. After checking around a bit I found the promising looking watchdog module, which makes use of those services on each platform it supports.

So, this is how we start:

pip install watchdog

Watchdog can observe changes to any file within a tree. But we are only interested in specific files. Lets make a class to do that:

# Watcher.py

import sys,os,time,Queue

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class Watcher(FileSystemEventHandler):
    def __init__(self,path):
        self.filesToWatch = {}
        self.events = Queue.Queue()
        self.observer = Observer()
        self.observer.schedule(self, path, recursive=True)
        self.observer.start()
    def watchFile(self,fileToWatch):
        if fileToWatch not in self.filesToWatch:
            self.filesToWatch[fileToWatch] = True
    def waitForEvent(self,block=True,timeout=None):
        try:
            return self.events.get(block,timeout)
        except Queue.Empty:
            return None
    def on_modified(self,ev):
        if ev.src_path in self.filesToWatch:
            self.events.put(ev.src_path)
    def stop(self):
        self.observer.stop()
        self.observer.join()

A Watcher is created with a path to watch. Then files of interest within that tree can be added by calling watchFile()

When one of those files changes, the name is added to a queue which can be read from another thread by calling waitForEvent()

Reloading code

When a file has changed, we want to reload it. Since that might happen very often, we need to manage the environment so that the program runs in a new empty environment each time, and the results can be returned to the calling program.

One way to achieve this in Python (2.x) is with the execfile() function. It takes a filename containing code to run, and arguments for the environments which can be modified when runnning it.

Here's how we do it

# Runner.py
import os

class module(object):
    pass # dummy class to add members to

class Runner():
    def __init__(self,watcher):
        self.watcher = watcher
    def mod(self,name):
        lenv = self.run(name+".py")
        robj = module()
        robj.__dict__ = lenv
        return robj
    def run(self,filename):
        fn = os.path.abspath(filename)
        self.watcher.watchFile(fn)
        genv = {'__builtins__': __builtins__,'use':self.mod}
        lenv = {}
        execfile(fn,genv,lenv)
        return lenv

First we give the Runner class a pointer to the Watcher.

The heart of this is the run() function. It takes a filename, adds it to the list of watched files, create a new clean environment, runs the code from the filename with execfile, and returns the local environment that was created.

One final feature is the mod() function. This is injected to the environment of the code to be run with the name use(). It takes a name, appends ".py" to it, loads and runs it, then sets the properties of a dummy object with the result. This simulates the result of a normal Python module import.

Setting it up

We tie these two modules together with a little script as follows

# live.py
import sys,os,time,Queue

from Watcher import Watcher
from Runner import Runner

if __name__ == "__main__":
    fn = os.path.abspath(sys.argv[1])
    path = os.path.split(fn)[0]
    os.chdir(path)
    w = Watcher(path)
    r = Runner(w)
    run = True
    while 1:
        try:
            if run:
                env = r.run(fn)
            run = w.waitForEvent(timeout=1)
        except KeyboardInterrupt:
            break
        except:
            import traceback
            traceback.print_exc()

    w.stop()

This takes two arguments - first the root of tree which will contain all the files to be edited, then the main file to run.

It creates a Watcher and Runner, then goes into a loop. First the main file is loaded and run. For now we discard the results though.

Then it waits 1 second for a new change file event. If one comes, it reloads the main file again. Otherwise it will wait again.

The loop can be interrupted at any time with Ctrl-C. If the run code raises an exception, it will be caught by the main loop and the file can be reloaded again after a change. If the run code goes into a long-lasting or infinite loop, Ctrl-C can still be used.

Testing it

$mkdir test
$python live.py test/test.py
IOError: [Errno 2] No such file or directory: '/pylivecode/test/test.py'

The program carries on running. Since we haven't made the file test.py yet, it naturally doesn't work. From another shell we create test.py

# test.py
print "Running test.py"

As soon as it is saved, the original shell shows

Running test.py

Next we modify test.py

# test.py
import math
print math.cos(0)
mod = use("testmod")
print mod.function("test")

When we save, the first shell shows the IOError again, so now we create testmod.py

# testmod.py
def function(arg):
    return arg+" function"

When we save this, the first shell shows

1.0
test function

Great! Now we can start up our runtime environment with a main file, and then edit that file, add modules to it, and edit any of those. After every edit and save (with any normal text editor), the new code will immediately be run.

We can use any normal python import modules also, but those will not be live-editable.

Of course it's not much use yet without a use case. That comes later.

You can get the code from bitbucket: https://bitbucket.org/baldand/pylivecode