Improve boolean parameter parsing

Multiprocessing and Apache's mod_wsgi don't play nicely.  Switch to
manually managing processes via subprocess.Popen etc instead.  When
running arbitrary code, we write it to an external file, and running
functions directly is no longer supported.

README.txt

@@ -9,7 +9,7 @@ Prerequisites:
   # Base dependencies
   sudo apt-get install python-numpy python-scipy
 
-  nilmdb (1.8.0+)
+  nilmdb (1.8.2+)
 
 Install:
 

docs/wsgi.md

@@ -21,13 +21,13 @@ arbitrary commands.
         SSLEngine On
 
         WSGIScriptAlias /nilmrun /home/nilm/nilmrun.wsgi
-        WSGIApplicationGroup nilmrun-appgroup
-        WSGIProcessGroup nilmrun-procgroup
         WSGIDaemonProcess nilmrun-procgroup threads=32 user=nilm group=nilm
-
-        # Access control example:
         <Location /nilmrun>
+            WSGIProcessGroup nilmrun-procgroup
+            WSGIApplicationGroup nilmrun-appgroup
             SSLRequireSSL
+
+            # Access control example:
             Order deny,allow
             Deny from all
             Allow from 1.2.3.4

nilmrun/filters/__init__.py

@@ -1 +0,0 @@
-# Filters

nilmrun/filters/trainola.py

@@ -1,260 +0,0 @@
-#!/usr/bin/python
-
-from nilmdb.utils.printf import *
-import nilmdb.client
-import nilmtools.filter
-from nilmdb.utils.time import (timestamp_to_human,
-                               timestamp_to_seconds,
-                               seconds_to_timestamp)
-from nilmdb.utils.interval import Interval
-
-import numpy as np
-import scipy
-import scipy.signal
-from numpy.core.umath_tests import inner1d
-import nilmrun
-from collections import OrderedDict
-
-class DataError(ValueError):
-    pass
-
-class Data(object):
-    def __init__(self, name, url, stream, start, end, columns):
-        """Initialize, get stream info, check columns"""
-        self.name = name
-        self.url = url
-        self.stream = stream
-        self.start = start
-        self.end = end
-
-        # Get stream info
-        self.client = nilmdb.client.numpyclient.NumpyClient(url)
-        self.info = nilmtools.filter.get_stream_info(self.client, stream)
-
-        # Build up name => index mapping for the columns
-        self.columns = OrderedDict()
-        for c in columns:
-            if (c['name'] in self.columns.keys() or
-                c['index'] in self.columns.values()):
-                raise DataError("duplicated columns")
-            if (c['index'] < 0 or c['index'] >= self.info.layout_count):
-                raise DataError("bad column number")
-            self.columns[c['name']] = c['index']
-        if not len(self.columns):
-            raise DataError("no columns")
-
-        # Count points
-        self.count = self.client.stream_count(self.stream, self.start, self.end)
-
-    def __str__(self):
-        return sprintf("%-20s: %s%s, %s rows",
-                       self.name, self.stream, str(self.columns.keys()),
-                       self.count)
-
-    def fetch(self, min_rows = 10, max_rows = 100000):
-        """Fetch all the data into self.data.  This is intended for
-        exemplars, and can only handle a relatively small number of
-        rows"""
-        # Verify count
-        if self.count == 0:
-            raise DataError("No data in this exemplar!")
-        if self.count < min_rows:
-            raise DataError("Too few data points: " + str(self.count))
-        if self.count > max_rows:
-            raise DataError("Too many data points: " + str(self.count))
-
-        # Extract the data
-        datagen = self.client.stream_extract_numpy(self.stream,
-                                                   self.start, self.end,
-                                                   self.info.layout,
-                                                   maxrows = self.count)
-        self.data = list(datagen)[0]
-
-        # Discard timestamp
-        self.data = self.data[:,1:]
-
-        # Subtract the mean from each column
-        self.data = self.data - self.data.mean(axis=0)
-
-        # Get scale factors for each column by computing dot product
-        # of each column with itself.
-        self.scale = inner1d(self.data.T, self.data.T)
-
-        # Ensure a minimum (nonzero) scale and convert to list
-        self.scale = np.maximum(self.scale, [1e-9]).tolist()
-
-def process(main, function, args = None, rows = 200000):
-    """Process through the data; similar to nilmtools.Filter.process_numpy"""
-    if args is None:
-        args = []
-
-    extractor = main.client.stream_extract_numpy
-    old_array = np.array([])
-    for new_array in extractor(main.stream, main.start, main.end,
-                               layout = main.info.layout, maxrows = rows):
-        # If we still had old data left, combine it
-        if old_array.shape[0] != 0:
-            array = np.vstack((old_array, new_array))
-        else:
-            array = new_array
-
-        # Process it
-        processed = function(array, args)
-
-        # Save the unprocessed parts
-        if processed >= 0:
-            old_array = array[processed:]
-        else:
-            raise Exception(sprintf("%s return value %s must be >= 0",
-                                    str(function), str(processed)))
-
-        # Warn if there's too much data remaining
-        if old_array.shape[0] > 3 * rows:
-            printf("warning: %d unprocessed rows in buffer\n",
-                   old_array.shape[0])
-
-    # Handle leftover data
-    if old_array.shape[0] != 0:
-        processed = function(array, args)
-
-def peak_detect(data, delta):
-    """Simple min/max peak detection algorithm, taken from my code
-    in the disagg.m from the 10-8-5 paper"""
-    mins = [];
-    maxs = [];
-    cur_min = (None, np.inf)
-    cur_max = (None, -np.inf)
-    lookformax = False
-    for (n, p) in enumerate(data):
-        if p > cur_max[1]:
-            cur_max = (n, p)
-        if p < cur_min[1]:
-            cur_min = (n, p)
-        if lookformax:
-            if p < (cur_max[1] - delta):
-                maxs.append(cur_max)
-                cur_min = (n, p)
-                lookformax = False
-        else:
-            if p > (cur_min[1] + delta):
-                mins.append(cur_min)
-                cur_max = (n, p)
-                lookformax = True
-    return (mins, maxs)
-
-def match(data, args):
-    """Perform cross-correlation match"""
-    ( columns, exemplars ) = args
-    nrows = data.shape[0]
-
-    # We want at least 10% more points than the widest exemplar.
-    widest = max([ x.count for x in exemplars ])
-    if (widest * 1.1) > nrows:
-        return 0
-
-    # This is how many points we'll consider valid in the
-    # cross-correlation.
-    valid = nrows + 1 - widest
-    matches = []
-
-    # Try matching against each of the exemplars
-    for e_num, e in enumerate(exemplars):
-        corrs = []
-
-        # Compute cross-correlation for each column
-        for c in e.columns:
-            a = data[:,columns[c] + 1]
-            b = e.data[:,e.columns[c]]
-            corr = scipy.signal.fftconvolve(a, np.flipud(b), 'valid')[0:valid]
-
-            # Scale by the norm of the exemplar
-            corr = corr / e.scale[columns[c]]
-            corrs.append(corr)
-
-        # Find the peaks using the column with the largest amplitude
-        biggest = e.scale.index(max(e.scale))
-        peaks_minmax = peak_detect(corrs[biggest], 0.1)
-        peaks = [ p[0] for p in peaks_minmax[1] ]
-
-        # Now look at every peak
-        for p in peaks:
-            # Correlation for each column must be close enough to 1.
-            for (corr, scale) in zip(corrs, e.scale):
-                # The accepted distance from 1 is based on the relative
-                # amplitude of the column.  Use a linear mapping:
-                # scale 1.0 -> distance 0.1
-                # scale 0.0 -> distance 1.0
-                distance = 1 - 0.9 * (scale / e.scale[biggest])
-                if abs(corr[p] - 1) > distance:
-                    # No match
-                    break
-            else:
-                # Successful match
-                matches.append((p, e_num))
-
-    # Print matches
-    for (point, e_num) in sorted(matches):
-        # Ignore matches that showed up at the very tail of the window,
-        # and shorten the window accordingly.  This is an attempt to avoid
-        # problems at chunk boundaries.
-        if point > (valid - 50):
-            valid -= 50
-            break
-        print "matched", data[point,0], "exemplar", exemplars[e_num].name
-
-    #from matplotlib import pyplot as p
-    #p.plot(data[:,1:3])
-    #p.show()
-
-    return max(valid, 0)
-
-def trainola(conf):
-    # Load main stream data
-    print "Loading stream data"
-    main = Data(None, conf['url'], conf['stream'],
-                conf['start'], conf['end'], conf['columns'])
-
-    # Pull in the exemplar data
-    exemplars = []
-    for n, e in enumerate(conf['exemplars']):
-        print sprintf("Loading exemplar %d: %s", n, e['name'])
-        ex = Data(e['name'], e['url'], e['stream'],
-                  e['start'], e['end'], e['columns'])
-        ex.fetch()
-        exemplars.append(ex)
-
-    # Verify that the exemplar columns are all represented in the main data
-    for n, ex in enumerate(exemplars):
-        for col in ex.columns:
-            if col not in main.columns:
-                raise DataError(sprintf("Exemplar %d column %s is not "
-                                        "available in main data", n, col))
-
-    # Process the main data
-    process(main, match, (main.columns, exemplars))
-
-    return "done"
-
-filterfunc = trainola
-
-def main(argv = None):
-    import simplejson as json
-    import argparse
-    import sys
-
-    parser = argparse.ArgumentParser(
-        formatter_class = argparse.RawDescriptionHelpFormatter,
-        version = nilmrun.__version__,
-        description = """Run Trainola using parameters passed in as
-        JSON-formatted data.""")
-    parser.add_argument("file", metavar="FILE", nargs="?",
-                        type=argparse.FileType('r'), default=sys.stdin)
-    args = parser.parse_args(argv)
-
-    conf = json.loads(args.file.read())
-    result = trainola(conf)
-    print json.dumps(result, sort_keys = True, indent = 2 * ' ')
-
-if __name__ == "__main__":
-    main()
-

nilmrun/processmanager.py

@@ -3,7 +3,7 @@
 from nilmdb.utils.printf import *
 
 import threading
-import multiprocessing
+import subprocess
 import cStringIO
 import sys
 import os
@@ -11,8 +11,12 @@ import signal
 import time
 import uuid
 import psutil
-import imp
+import tempfile
-import traceback
+import atexit
+import shutil
+
+class ProcessError(Exception):
+    pass
 
 class LogReceiver(object):
     """Spawn a thread that listens to a pipe for log messages,
@@ -38,9 +42,8 @@ class LogReceiver(object):
         self.log = cStringIO.StringIO()
 
 class Process(object):
-    """Spawn and manage a process that calls a Python function"""
+    """Spawn and manage a subprocess, and capture its output."""
-    def __init__(self, name, function, parameters):
+    def __init__(self, name, argv, tempfile = None):
-        self.parameters = parameters
         self.start_time = None
         self.name = name
 
@@ -48,65 +51,39 @@ class Process(object):
         (rpipe, wpipe) = os.pipe()
         self._log = LogReceiver(rpipe)
 
-        # Start the function in a new process
+        # Stdin is null
-        self._process = multiprocessing.Process(
+        nullfd = os.open(os.devnull, os.O_RDONLY)
-            target = self._trampoline, name = name,
-            args = (function, rpipe, wpipe, parameters))
-        self._process.daemon = True
-        self._process.start()
 
-        # Close the writer end of the pipe, get process info
+        # Spawn the new process
-        os.close(wpipe)
+        try:
+            self._process = subprocess.Popen(args = argv, stdin = nullfd,
+                                             stdout = wpipe, stderr = wpipe,
+                                             close_fds = True, cwd = "/tmp")
+        except (OSError, TypeError) as e:
+            raise ProcessError(str(e))
+        finally:
+            # Close the FDs we don't need
+            os.close(wpipe)
+            os.close(nullfd)
+
+        # Get process info
         self.start_time = time.time()
         self.pid = str(uuid.uuid1(self._process.pid or 0))
 
-    def _trampoline(self, func, rpipe, wpipe, param): # pragma: no cover
+    def _join(self, timeout = 1.0):
-        # No coverage report for this, because it's executed in a subprocess
+        start = time.time()
-        """Trampoline function to set up stdio and call the real function."""
+        while True:
-        # Close the reader end of the pipe
+            if self._process.poll() is not None:
-        os.close(rpipe)
+                return True
-
+            if (time.time() - start) >= timeout:
-        # Like os.close() but ignores errors
+                return False
-        def tryclose(fd):
+            time.sleep(0.1)
-            try:
-                os.close(fd)
-            except OSError:
-                pass
-
-        # Remap stdio to go to the pipe.  We do this at the OS level,
-        # replacing FDs, so that future spawned processes do the right thing.
-
-        # stdin
-        sys.stdin.close()
-        tryclose(0)
-        fd = os.open(os.devnull, os.O_RDONLY) # 0
-        sys.stdin = os.fdopen(fd, 'r', 0)
-
-        # stdout
-        sys.stdout.close()
-        tryclose(1)
-        fd = os.dup(wpipe) # 1
-        sys.stdout = os.fdopen(fd, 'w', 0)
-
-        # stdout
-        sys.stderr.close()
-        tryclose(2)
-        fd = os.dup(wpipe) # 2
-        sys.stderr = os.fdopen(fd, 'w', 0)
-
-        # Don't need this extra fd
-        os.close(wpipe)
-
-        # Ready to go -- call the function, exit when it's done
-        func(param)
-        sys.exit(0)
 
     def terminate(self, timeout = 1.0):
         """Terminate a process, and all of its children that are in the same
         process group."""
         # First give it some time to die on its own
-        self._process.join(timeout)
+        if self._join(timeout):
-        if not self.alive:
             return True
 
         def getpgid(pid):
@@ -126,8 +103,7 @@ class Process(object):
                 os.kill(proc.pid, signal.SIGTERM)
 
         # Wait for it to die again
-        self._process.join(timeout)
+        if self._join(timeout):
-        if not self.alive:
             return True
 
         # One more try with SIGKILL
@@ -136,8 +112,7 @@ class Process(object):
                 os.kill(proc.pid, signal.SIGKILL)
 
         # See if it worked
-        self._process.join(timeout)
+        return self._join(timeout)
-        return not self.alive
 
     def clear_log(self):
         self._log.clear()
@@ -148,59 +123,19 @@ class Process(object):
 
     @property
     def alive(self):
-        return self._process.is_alive()
+        return self._process.poll() is None
 
     @property
     def exitcode(self):
-        return self._process.exitcode
+        return self._process.returncode
-
-def _exec_user_code(codeargs): # pragma: no cover (runs in subprocess)
-    """Execute 'code' as if it were placed into a file and executed"""
-    (code, args) = codeargs
-    # This is split off into a separate function because the Python3
-    # syntax of "exec" triggers a SyntaxError in Python2, if it's within
-    # a nested function.
-    imp.acquire_lock()
-    try:
-        module = imp.new_module("__main__")
-    finally:
-        imp.release_lock()
-    module.__file__ = "<user-code>"
-    sys.argv = [''] + args
-    # Wrap the compile and exec in a try/except so we can format the
-    # exception more nicely.
-    try:
-        codeobj = compile(code, '<user-code>', 'exec',
-                          flags = 0, dont_inherit = 1)
-        exec(codeobj, module.__dict__, {})
-    except:
-        # Pull out the exception
-        info = sys.exc_info()
-        tblist = traceback.extract_tb(info[2])
-
-        # First entry is probably this code; get rid of it
-        if len(tblist) and tblist[0][2] == '_exec_user_code':
-            tblist = tblist[1:]
-
-        # Add the user's source code to every line that's missing it
-        lines = code.splitlines()
-        for (n, (name, line, func, text)) in enumerate(tblist):
-            if name == '<user-code>' and text is None and line <= len(lines):
-                tblist[n] = (name, line, func, lines[line-1].strip())
-
-        # Print it to stderr in the usual format
-        out = ['Traceback (most recent call last):\n']
-        out.extend(traceback.format_list(tblist))
-        out.extend(traceback.format_exception_only(info[0], info[1]))
-        sys.stderr.write("".join(out))
-        sys.stderr.flush()
-        sys.exit(1)
-    sys.exit(0)
 
 class ProcessManager(object):
     """Track and manage a collection of Process objects"""
     def __init__(self):
         self.processes = {}
+        self.tmpfiles = {}
+        self.tmpdir = tempfile.mkdtemp(prefix = "nilmrun-usercode-")
+        atexit.register(shutil.rmtree, self.tmpdir)
 
     def __iter__(self):
         return iter(self.processes.keys())
@@ -208,35 +143,35 @@ class ProcessManager(object):
     def __getitem__(self, key):
         return self.processes[key]
 
-    def run_function(self, procname, function, parameters):
-        """Run a Python function that already exists"""
-        new = Process(procname, function, parameters)
-        self.processes[new.pid] = new
-        return new.pid
-
     def run_code(self, procname, code, args):
         """Evaluate 'code' as if it were placed into a Python file and
-        executed.  The arguments will be accessible in the code as
+        executed.  The arguments, which must be strings, will be
-        sys.argv[1:]."""
+        accessible in the code as sys.argv[1:]."""
-        return self.run_function(procname, _exec_user_code, (code, args))
+        # The easiest way to do this, by far, is to just write the
+        # code to a file.
+        (fd, path) = tempfile.mkstemp(prefix = "nilmrun-usercode-",
+                                      suffix = ".py", dir=self.tmpdir)
+        with os.fdopen(fd, 'w') as f:
+            f.write(code)
+        argv = [ sys.executable, "-B", "-s", "-u", path ] + args
+        pid = self.run_command(procname, argv)
+        self.tmpfiles[pid] = path
+        return pid
 
     def run_command(self, procname, argv):
         """Execute a command line program"""
-        def spwan_user_command(argv): # pragma: no cover (runs in subprocess)
+        new = Process(procname, argv)
-            try:
+        self.processes[new.pid] = new
-                maxfd = os.sysconf("SC_OPEN_MAX")
+        return new.pid
-            except Exception:
-                maxfd = 256
-            os.closerange(3, maxfd)
-            try:
-                os.chdir("/tmp")
-            except OSError:
-                pass
-            os.execvp(argv[0], argv)
-        return self.run_function(procname, spwan_user_command, argv)
 
     def terminate(self, pid):
         return self.processes[pid].terminate()
 
     def remove(self, pid):
+        if pid in self.tmpfiles:
+            try:
+                os.unlink(self.tmpfiles[pid])
+            except OSError: # pragma: no cover
+                pass
+            del self.tmpfiles[pid]
         del self.processes[pid]

nilmrun/server.py

@@ -23,10 +23,10 @@ from nilmdb.server.serverutil import (
         json_error_page,
         cherrypy_start,
         cherrypy_stop,
+        bool_param,
         )
 import nilmrun
-import nilmrun.filters.trainola
+import nilmrun.testfilter
-import nilmrun.filters.dummy
 
 # Add CORS_allow tool
 cherrypy.tools.CORS_allow = cherrypy.Tool('on_start_resource', CORS_allow)
@@ -69,7 +69,6 @@ class AppProcess(object):
             "exitcode": self.manager[pid].exitcode,
             "name": self.manager[pid].name,
             "start_time": self.manager[pid].start_time,
-            "parameters": self.manager[pid].parameters,
             "log": self.manager[pid].log,
             }
 
@@ -79,6 +78,7 @@ class AppProcess(object):
     def status(self, pid, clear = False):
         """Return status about a process.  If clear = True, also clear
         the log."""
+        clear = bool_param(clear)
         if pid not in self.manager:
             raise cherrypy.HTTPError("404 Not Found", "No such PID")
         status = self.process_status(pid)
@@ -117,44 +117,34 @@ class AppRun(object):
     @cherrypy.expose
     @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
+    @exception_to_httperror(nilmrun.processmanager.ProcessError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
     def command(self, argv):
         """Execute an arbitrary program on the server.  argv is a
         list of the program and its arguments: 'argv[0]' is the program
         and 'argv[1:]' are arguments"""
+        if not isinstance(argv, list):
+            raise cherrypy.HTTPError("400 Bad Request",
+                                     "argv must be a list of strings")
         return self.manager.run_command("command", argv)
 
+    # /run/code
     @cherrypy.expose
     @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
+    @exception_to_httperror(nilmrun.processmanager.ProcessError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
     def code(self, code, args):
         """Execute arbitrary Python code.  'code' is a formatted string.
         It will be run as if it were written into a Python file and
-        executed, with the arguments in 'args' passed on the command line
+        executed.  'args' is a list of strings, and they are passed
-        (i.e., they end up in sys.argv[1:])"""
+        on the command line as additional arguments (i.e., they end up
+        in sys.argv[1:])"""
+        if not isinstance(args, list):
+            raise cherrypy.HTTPError("400 Bad Request",
+                                     "args must be a list of strings")
         return self.manager.run_code("usercode", code, args)
 
-    # /run/trainola
-    @cherrypy.expose
-    @cherrypy.tools.json_in()
-    @cherrypy.tools.json_out()
-    @exception_to_httperror(KeyError, ValueError)
-    @cherrypy.tools.CORS_allow(methods = ["POST"])
-    def trainola(self, data):
-        return self.manager.run_function(
-            "trainola", nilmrun.filters.trainola.filterfunc, data)
-
-    # /run/dummy
-    @cherrypy.expose
-    @cherrypy.tools.json_in()
-    @cherrypy.tools.json_out()
-    @exception_to_httperror(KeyError, ValueError)
-    @cherrypy.tools.CORS_allow(methods = ["POST"])
-    def dummy(self, data):
-        return self.manager.run_function(
-            "dummy", nilmrun.filters.dummy.filterfunc, data)
-
 class Server(object):
     def __init__(self, host = '127.0.0.1', port = 8080,
                  embedded = True,         # hide diagnostics and output, etc

nilmrun/testfilter.py

@@ -6,7 +6,7 @@ import signal
 import sys
 
 # This is just for testing the process management.
-def filterfunc(n):
+def test(n):
     n = int(n)
     if n < 0: # raise an exception
         raise Exception("test exception")

setup.py

@@ -61,7 +61,7 @@ setup(name='nilmrun',
       long_description = "NILM Database Filter Runner",
       license = "Proprietary",
       author_email = 'jim@jtan.com',
-      install_requires = [ 'nilmdb >= 1.8.0',
+      install_requires = [ 'nilmdb >= 1.8.2',
                            'nilmtools >= 1.2.2',
                            'numpy',
                            'scipy',

tests/test.order

@@ -1,3 +1,3 @@
-test_client.py
+test_nilmrun.py
 
 test_*.py

tests/test_nilmrun.py

@@ -30,6 +30,7 @@ import textwrap
 from testutil.helpers import *
 
 testurl = "http://localhost:32181/"
+#testurl = "http://bucket.mit.edu/nilmrun/"
 
 def setup_module():
     global test_server
@@ -49,10 +50,10 @@ class TestClient(object):
 
     def wait_kill(self, client, pid, timeout = 1):
         time.sleep(timeout)
-        status = client.get("/process/status", { "pid": pid })
+        status = client.get("process/status", { "pid": pid })
         if not status["alive"]:
             raise AssertionError("died before we could kill it")
-        status = client.post("/process/remove", { "pid": pid })
+        status = client.post("process/remove", { "pid": pid })
         if status["alive"]:
             raise AssertionError("didn't get killed")
         return status
@@ -61,180 +62,46 @@ class TestClient(object):
         start = time.time()
         status = None
         while (time.time() - start) < timeout:
-            status = client.get("/process/status", { "pid": pid })
+            status = client.get("process/status", { "pid": pid })
             if status["alive"] == False:
                 break
         else:
             raise AssertionError("process " + str(pid) + " didn't die in " +
                                  str(timeout) + " seconds: " + repr(status))
         if remove:
-            status = client.post("/process/remove", { "pid": pid })
+            status = client.post("process/remove", { "pid": pid })
         return status
 
     def test_client_01_basic(self):
         client = HTTPClient(baseurl = testurl)
-        version = client.get("/version")
+        version = client.get("version")
         eq_(distutils.version.LooseVersion(version),
             distutils.version.LooseVersion(nilmrun.__version__))
 
-        in_("This is NilmRun", client.get("/"))
+        in_("This is NilmRun", client.get(""))
 
         with assert_raises(ClientError):
-            client.get("/favicon.ico")
+            client.get("favicon.ico")
 
     def test_client_02_manager(self):
         client = HTTPClient(baseurl = testurl)
 
-        eq_(client.get("/process/list"), [])
+        eq_(client.get("process/list"), [])
 
         with assert_raises(ClientError) as e:
-            client.get("/process/status", { "pid": 12345 })
+            client.get("process/status", { "pid": 12345 })
         in_("No such PID", str(e.exception))
         with assert_raises(ClientError):
-            client.get("/process/remove", { "pid": 12345 })
+            client.get("process/remove", { "pid": 12345 })
         in_("No such PID", str(e.exception))
 
-    def test_client_03_process_basic(self):
+    def test_client_03_run_command(self):
         client = HTTPClient(baseurl = testurl, post_json = True)
-
+        eq_(client.get("process/list"), [])
-        # start dummy filter
-        pid = client.post("/run/dummy", { "data": 30 })
-        eq_(client.get("/process/list"), [pid])
-        time.sleep(1)
-
-        # Verify that status looks OK
-        status = client.get("/process/status", { "pid": pid, "clear": True })
-        for x in [ "pid", "alive", "exitcode", "name",
-                   "start_time", "parameters", "log" ]:
-            in_(x, status)
-        in_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
-        eq_(status["alive"], True)
-        eq_(status["exitcode"], None)
-
-        # Check that the log got cleared
-        status = client.get("/process/status", { "pid": pid })
-        nin_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
-
-        # See that it ended properly
-        status = self.wait_end(client, pid, remove = False)
-        in_("dummy 27\ndummy 28\ndummy 29\n", status["log"])
-        eq_(status["exitcode"], 0)
-
-        # Remove it
-        killstatus = client.post("/process/remove", { "pid": pid })
-        eq_(status, killstatus)
-        eq_(client.get("/process/list"), [])
-        with assert_raises(ClientError) as e:
-            client.post("/process/remove", { "pid": pid })
-        in_("No such PID", str(e.exception))
-
-    def test_client_04_process_terminate(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-
-        # Trigger exception in filter
-        pid = client.post("/run/dummy", { "data": -1 })
-        time.sleep(0.5)
-        status = client.get("/process/status", { "pid": pid })
-        eq_(status["alive"], False)
-        eq_(status["exitcode"], 1)
-        in_("Exception: test exception", status["log"])
-        client.post("/process/remove", { "pid": pid })
-
-        # Kill a running filter by removing it early
-        newpid = client.post("/run/dummy", { "data": 50 })
-        ne_(newpid, pid)
-        time.sleep(0.5)
-        start = time.time()
-        status = client.post("/process/remove", { "pid": newpid })
-        elapsed = time.time() - start
-        # Should have died in slightly over 1 second
-        assert(0.5 < elapsed < 2)
-        eq_(status["alive"], False)
-        ne_(status["exitcode"], 0)
-
-        # No more
-        eq_(client.get("/process/list"), [])
-
-        # Try to remove a running filter that ignored SIGTERM
-        pid = client.post("/run/dummy", { "data": 0 })
-        start = time.time()
-        status = client.post("/process/remove", { "pid": pid })
-        elapsed = time.time() - start
-        # Should have died in slightly over 2 seconds
-        assert(1.5 < elapsed < 3)
-        eq_(status["alive"], False)
-        ne_(status["exitcode"], 0)
-
-    def test_client_05_trainola_simple(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-        pid = client.post("/run/trainola", { "data": {} })
-        status = self.wait_end(client, pid, remove = False)
-        ne_(status["exitcode"], 0)
-        status = client.post("/process/remove", { "pid": pid })
-
-    @unittest.skip("needs a running nilmdb")
-    def test_client_06_trainola(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-
-        data = { "url": "http://bucket.mit.edu/nilmdb",
-                 "stream": "/sharon/prep-a",
-                 "start": 1366111383280463,
-                 "end": 1366126163457797,
-                 "columns": [ { "name": "P1", "index": 0 },
-                              { "name": "Q1", "index": 1 },
-                              { "name": "P3", "index": 2 } ],
-                 "exemplars": [
-                     { "name": "Boiler Pump ON",
-                       "url": "http://bucket.mit.edu/nilmdb",
-                       "stream": "/sharon/prep-a",
-                       "start": 1366260494269078,
-                       "end": 1366260608185031,
-                       "columns": [ { "name": "P1", "index": 0 },
-                                    { "name": "Q1", "index": 1 }
-                                    ]
-                       },
-                     { "name": "Boiler Pump OFF",
-                       "url": "http://bucket.mit.edu/nilmdb",
-                       "stream": "/sharon/prep-a",
-                       "start": 1366260864215764,
-                       "end": 1366260870882998,
-                       "columns": [ { "name": "P1", "index": 0 },
-                                    { "name": "Q1", "index": 1 }
-                                    ]
-                       }
-                     ]
-                 }
-
-        # start trainola
-        pid = client.post("/run/trainola", { "data": data })
-
-        # wait for it to finish
-        for i in range(60):
-            time.sleep(1)
-            if i == 2:
-                status = client.get("/process/status", { "pid": pid,
-                                                         "clear": True })
-                in_("Loading stream data", status['log'])
-            elif i == 3:
-                status = client.get("/process/status", { "pid": pid })
-                nin_("Loading stream data", status['log'])
-            else:
-                status = client.get("/process/status", { "pid": pid })
-            if status["alive"] == False:
-                break
-        else:
-            client.post("/process/remove", {"pid": pid })
-            raise AssertionError("took too long")
-        if i < 3:
-            raise AssertionError("too fast?")
-
-    def test_client_07_run_command(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-        eq_(client.get("/process/list"), [])
 
         def do(argv, kill):
-            pid = client.post("/run/command", { "argv": argv } )
+            pid = client.post("run/command", { "argv": argv } )
-            eq_(client.get("/process/list"), [pid])
+            eq_(client.get("process/list"), [pid])
             if kill:
                 return self.wait_kill(client, pid)
             return self.wait_end(client, pid)
@@ -250,26 +117,153 @@ class TestClient(object):
         eq_("3\n", status["log"])
 
         # Missing command
-        status = do(["/no-such-command-blah-blah"], False)
+        with assert_raises(ClientError) as e:
-        ne_(status["exitcode"], 0)
+            do(["/no-such-command-blah-blah"], False)
+        in_("No such file or directory", str(e.exception))
 
         # Kill a slow command
         status = do(["sleep", "60"], True)
         ne_(status["exitcode"], 0)
 
-    def test_client_08_run_code(self):
+    def _run_testfilter(self, client, args):
+        code = textwrap.dedent("""
+        import nilmrun.testfilter
+        import simplejson as json
+        import sys
+        nilmrun.testfilter.test(json.loads(sys.argv[1]))
+        """)
+        jsonargs = json.dumps(args)
+        return client.post("run/code", { "code": code, "args": [ jsonargs ] })
+
+    def test_client_04_process_basic(self):
         client = HTTPClient(baseurl = testurl, post_json = True)
-        eq_(client.get("/process/list"), [])
+
+        # start dummy filter
+        pid = self._run_testfilter(client, 30)
+        eq_(client.get("process/list"), [pid])
+        time.sleep(1)
+
+        # Verify that status looks OK
+        status = client.get("process/status", { "pid": pid, "clear": True })
+        for x in [ "pid", "alive", "exitcode", "name", "start_time", "log" ]:
+            in_(x, status)
+        in_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
+        eq_(status["alive"], True)
+        eq_(status["exitcode"], None)
+
+        # Check that the log got cleared
+        status = client.get("process/status", { "pid": pid })
+        nin_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
+
+        # See that it ended properly
+        status = self.wait_end(client, pid, remove = False)
+        in_("dummy 27\ndummy 28\ndummy 29\n", status["log"])
+        eq_(status["exitcode"], 0)
+
+        # Remove it
+        killstatus = client.post("process/remove", { "pid": pid })
+        eq_(status, killstatus)
+        eq_(client.get("process/list"), [])
+        with assert_raises(ClientError) as e:
+            client.post("process/remove", { "pid": pid })
+        in_("No such PID", str(e.exception))
+
+    def test_client_05_process_terminate(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        # Trigger exception in filter
+        pid = self._run_testfilter(client, -1)
+        time.sleep(0.5)
+        status = client.get("process/status", { "pid": pid })
+        eq_(status["alive"], False)
+        eq_(status["exitcode"], 1)
+        in_("Exception: test exception", status["log"])
+        client.post("process/remove", { "pid": pid })
+
+        # Kill a running filter by removing it early
+        newpid = self._run_testfilter(client, 50)
+        ne_(newpid, pid)
+        time.sleep(0.5)
+        start = time.time()
+        status = client.post("process/remove", { "pid": newpid })
+        elapsed = time.time() - start
+        # Should have died in slightly over 1 second
+        assert(0.5 < elapsed < 2)
+        eq_(status["alive"], False)
+        ne_(status["exitcode"], 0)
+
+        # No more
+        eq_(client.get("process/list"), [])
+
+        # Try to remove a running filter that ignored SIGTERM
+        pid = self._run_testfilter(client, 0)
+        start = time.time()
+        status = client.post("process/remove", { "pid": pid })
+        elapsed = time.time() - start
+        # Should have died in slightly over 2 seconds
+        assert(1.5 < elapsed < 3)
+        eq_(status["alive"], False)
+        ne_(status["exitcode"], 0)
+
+    @unittest.skip("needs a running nilmdb; trainola moved to nilmtools")
+    def test_client_06_trainola(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        data = { "url": "http://bucket.mit.edu/nilmdb",
+                 "dest_stream": "/sharon/prep-a-matches",
+                 "stream": "/sharon/prep-a",
+                 "start": 1366111383280463,
+                 "end": 1366126163457797,
+                 "columns": [ { "name": "P1", "index": 0 },
+                              { "name": "Q1", "index": 1 },
+                              { "name": "P3", "index": 2 } ],
+                 "exemplars": [
+                     { "name": "Boiler Pump ON",
+                       "url": "http://bucket.mit.edu/nilmdb",
+                       "stream": "/sharon/prep-a",
+                       "start": 1366260494269078,
+                       "end": 1366260608185031,
+                       "dest_column": 0,
+                       "columns": [ { "name": "P1", "index": 0 },
+                                    { "name": "Q1", "index": 1 }
+                                    ]
+                       },
+                     { "name": "Boiler Pump OFF",
+                       "url": "http://bucket.mit.edu/nilmdb",
+                       "stream": "/sharon/prep-a",
+                       "start": 1366260864215764,
+                       "end": 1366260870882998,
+                       "dest_column": 1,
+                       "columns": [ { "name": "P1", "index": 0 },
+                                    { "name": "Q1", "index": 1 }
+                                    ]
+                       }
+                     ]
+                 }
+        pid = client.post("run/code", { "code": "import nilmtools.trainola\n" +
+                                         "nilmtools.trainola.main()",
+                                         "args": [ data ] })
+        while True:
+            status = client.get("process/status", { "pid": pid, "clear": 1 })
+            sys.stdout.write(status["log"])
+            sys.stdout.flush()
+            if status["alive"] == False:
+                break
+        status = client.post("process/remove", { "pid": pid })
+        os._exit(int(status["exitcode"]))
+
+    def test_client_07_run_code(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        eq_(client.get("process/list"), [])
 
         def do(code, args, kill):
-            pid = client.post("/run/code", { "code": code, "args": args } )
+            pid = client.post("run/code", { "code": code, "args": args } )
-            eq_(client.get("/process/list"), [pid])
+            eq_(client.get("process/list"), [pid])
             if kill:
                 return self.wait_kill(client, pid)
             return self.wait_end(client, pid)
 
         # basic code snippet
-        code=textwrap.dedent("""
+        code = textwrap.dedent("""
         print 'hello'
         def foo(arg):
             print 'world'
@@ -279,7 +273,7 @@ class TestClient(object):
         eq_(status["exitcode"], 0)
 
         # compile error
-        code=textwrap.dedent("""
+        code = textwrap.dedent("""
         def foo(arg:
             print 'hello'
         """)
@@ -288,31 +282,36 @@ class TestClient(object):
         eq_(status["exitcode"], 1)
 
         # traceback in user code should be formatted nicely
-        code=textwrap.dedent("""
+        code = textwrap.dedent("""
         def foo(arg):
             raise Exception(arg)
         foo(123)
         """)
         status = do(code, [], False)
+        cleaned_log = re.sub('File "[^"]*",', 'File "",', status["log"])
         eq_('Traceback (most recent call last):\n' +
-        '  File "<user-code>", line 4, in <module>\n' +
+        '  File "", line 4, in <module>\n' +
         '    foo(123)\n' +
-        '  File "<user-code>", line 3, in foo\n' +
+        '  File "", line 3, in foo\n' +
         '    raise Exception(arg)\n' +
-        'Exception: 123\n', status["log"])
+        'Exception: 123\n', cleaned_log)
         eq_(status["exitcode"], 1)
 
         # argument handling (strings come in as unicode)
-        code=textwrap.dedent("""
+        code = textwrap.dedent("""
         import sys
         print sys.argv[1].encode('ascii'), sys.argv[2]
+        sys.exit(0)  # also test raising SystemExit
         """)
-        status = do(code, ["hello", 123], False)
+        with assert_raises(ClientError) as e:
+	    do(code, ["hello", 123], False)
+        in_("400 Bad Request", str(e.exception))
+        status = do(code, ["hello", "123"], False)
         eq_(status["log"], "hello 123\n")
         eq_(status["exitcode"], 0)
 
         # try killing a long-running process
-        code=textwrap.dedent("""
+        code = textwrap.dedent("""
         import time
         print 'hello'
         time.sleep(60)
@@ -321,3 +320,14 @@ class TestClient(object):
         status = do(code, [], True)
         eq_(status["log"], "hello\n")
         ne_(status["exitcode"], 0)
+
+    def test_client_08_bad_types(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        with assert_raises(ClientError) as e:
+            client.post("run/code", { "code": "asdf", "args": "qwer" })
+        in_("must be a list", str(e.exception))
+
+        with assert_raises(ClientError) as e:
+            client.post("run/command", { "argv": "asdf" })
+        in_("must be a list", str(e.exception))