More helpful error if exemplar stream doesn't exist

nilmtools/trainola.py

@@ -6,6 +6,7 @@ import nilmtools.filter
 from nilmdb.utils.time import (timestamp_to_human,
                                timestamp_to_seconds,
                                seconds_to_timestamp)
+from nilmdb.utils import datetime_tz
 from nilmdb.utils.interval import Interval
 
 import numpy as np
@@ -15,6 +16,7 @@ from numpy.core.umath_tests import inner1d
 import nilmrun
 from collections import OrderedDict
 import sys
+import time
 import functools
 import collections
 
@@ -52,6 +54,9 @@ class Exemplar(object):
         # Get stream info
         self.client = nilmdb.client.numpyclient.NumpyClient(self.url)
         self.info = nilmtools.filter.get_stream_info(self.client, self.stream)
+        if not self.info:
+            raise DataError(sprintf("exemplar stream '%s' does not exist " +
+                                    "on server '%s'", self.stream, self.url))
 
         # Build up name => index mapping for the columns
         self.columns = build_column_mapping(exinfo['columns'], self.info)
@@ -117,6 +122,10 @@ def peak_detect(data, delta):
                 lookformax = True
     return (mins, maxs)
 
+def timestamp_to_short_human(timestamp):
+    dt = datetime_tz.datetime_tz.fromtimestamp(timestamp_to_seconds(timestamp))
+    return dt.strftime("%H:%M:%S")
+
 def trainola_matcher(data, interval, args, insert_func, final_chunk):
     """Perform cross-correlation match"""
     ( src_columns, dest_count, exemplars ) = args
@@ -183,7 +192,10 @@ def trainola_matcher(data, interval, args, insert_func, final_chunk):
     insert_func(out)
 
     # Return how many rows we processed
-    return max(valid, 0)
+    valid = max(valid, 0)
+    printf("  [%s] matched %d exemplars in %d rows\n",
+           timestamp_to_short_human(data[0][0]), np.sum(out[:,1:]), valid)
+    return valid
 
 def trainola(conf):
     print "Trainola", nilmtools.__version__
@@ -247,14 +259,20 @@ def trainola(conf):
                                   src.path, layout = src.layout, maxrows = rows)
     inserter = functools.partial(dest_client.stream_insert_numpy_context,
                                  dest.path)
+    start = time.time()
+    processed_time = 0
+    printf("Processing intervals:\n")
     for interval in intervals:
-        printf("Processing interval:\n")
-        printf("  %s\n", interval.human_string())
+        printf("%s\n", interval.human_string())
         nilmtools.filter.process_numpy_interval(
             interval, extractor, inserter, rows * 3,
             trainola_matcher, (src_columns, dest.layout_count, exemplars))
+        processed_time += (timestamp_to_seconds(interval.end) -
+                           timestamp_to_seconds(interval.start))
+    elapsed = max(time.time() - start, 1e-3)
 
-    return "done"
+    printf("Done. Processed %.2f seconds per second.\n",
+           processed_time / elapsed)
 
 def main(argv = None):
     import simplejson as json