Update copy tool

Makefile

@@ -1,3 +1,11 @@
+test:
+	-@nilmtool destroy /lees-compressor/no-leak/raw/4 || true
+	-@nilmtool destroy /lees-compressor/no-leak/raw/16 || true
+	-@nilmtool create /lees-compressor/no-leak/raw/4 float32_18 || true
+	-@nilmtool create /lees-compressor/no-leak/raw/16 float32_18 || true
+	time python nilmtools/decimate.py -s '2013-02-04 18:10:00' -e '2013-02-04 18:11:00' /lees-compressor/no-leak/raw/1 /lees-compressor/no-leak/raw/4
+	python nilmtools/decimate.py -s '2013-02-04 18:10:00' -e '2013-02-04 18:11:00' /lees-compressor/no-leak/raw/4 /lees-compressor/no-leak/raw/16
+
 all:
 	@echo "Try 'make install'"
 

nilmtools/copy.py

@@ -0,0 +1,36 @@
+#!/usr/bin/python
+
+import nilmtools.filter
+import nilmdb.client
+import numpy as np
+import sys
+
+def main():
+    f = nilmtools.filter.Filter()
+    parser = f.setup_parser("Copy a stream")
+
+    # Parse arguments
+    try:
+        args = f.parse_args()
+    except nilmtools.filter.MissingDestination as e:
+        print "Source is %s (%s)" % (e.src.path, e.src.layout)
+        print "Destination %s doesn't exist" % (e.dest.path)
+        print "You could make it with a command like:"
+        print "  nilmtool create", e.dest.path, e.src.layout
+        raise SystemExit(1)
+
+    # Copy metadata
+    meta = f.client.stream_get_metadata(f.src.path)
+    f.check_dest_metadata(meta)
+
+    # Copy all rows of data as ASCII strings
+    extractor = nilmdb.client.Client(args.url).stream_extract
+    inserter = nilmdb.client.Client(args.url).stream_insert_context
+    for i in f.intervals():
+        print "Processing", f.interval_string(i)
+        with inserter(f.dest.path, i.start, i.end) as insert_ctx:
+            for row in extractor(f.src.path, i.start, i.end):
+                insert_ctx.insert(row + "\n")
+
+if __name__ == "__main__":
+    main()

nilmtools/decimate.py

@@ -0,0 +1,79 @@
+#!/usr/bin/python
+
+import nilmtools.filter
+import nilmdb.client
+import numpy as np
+import operator
+
+def main():
+    f = nilmtools.filter.Filter()
+    parser = f.setup_parser("Decimate a stream")
+    group = parser.add_argument_group("Decimate options")
+    group.add_argument('-f', '--factor', action='store', default=4, type=int,
+                       help='Decimation factor (default: %(default)s)')
+    # Parse arguments
+    try:
+        args = f.parse_args()
+    except nilmtools.filter.MissingDestination as e:
+        # If no destination, suggest how to create it by figuring out
+        # a recommended layout.
+        src = e.src
+        dest = e.dest
+        print "Source is %s (%s)" % (src.path, src.layout)
+        print "Destination %s doesn't exist" % (dest.path)
+        if "decimate_source" in f.client.stream_get_metadata(src.path):
+            rec = src.layout
+        elif 'int32' in src.layout_type or 'float64' in src.layout_type:
+            rec = 'float64_' + str(src.layout_count * 3)
+        else:
+            rec = 'float32_' + str(src.layout_count * 3)
+        print "You could make it with a command like:"
+        print "  nilmtool create", dest.path, rec
+        raise SystemExit(1)
+
+    if not (args.factor >= 2):
+        raise Exception("factor needs to be 2 or more")
+
+    f.check_dest_metadata({ "decimate_source": args.srcpath,
+                            "decimate_factor": args.factor })
+
+    # If source is decimated, we have to decimate a bit differently
+    if "decimate_source" in f.client.stream_get_metadata(args.srcpath):
+        n = f.src.layout_count // 3
+        f.process_python(function = decimate_again, rows = args.factor,
+                         args = (n,))
+    else:
+        n = f.src.layout_count
+        f.process_python(function = decimate_first, rows = args.factor,
+                         args = (n,))
+
+def decimate_first(data, n):
+    """Decimate original data -- result has 3 times as many columns"""
+    # For this simple calculation, converting to a Numpy array
+    # and doing the math is slower than just doing it directly.
+    rows = iter(data)
+    r_sum = r_min = r_max = rows.next()
+    for row in rows:
+        r_sum = map(operator.add, r_sum, row)
+        r_min = map(min, r_min, row)
+        r_max = map(max, r_max, row)
+    r_mean = [ x / len(data) for x in r_sum ]
+    return [ [ r_mean[0] ] + r_mean[1:] + r_min[1:] + r_max[1:] ]
+
+def decimate_again(data, n):
+    """Decimate already-decimated data -- result has the same number
+    of columns"""
+    rows = iter(data)
+    r = rows.next()
+    r_sum = r[0:(n+1)]
+    r_min = r[(n+1):(2*n+1)]
+    r_max = r[(2*n+1):(3*n+1)]
+    for r in rows:
+        r_sum = map(operator.add, r_sum, r[0:(n+1)])
+        r_min = map(min, r_min, r[(n+1):(2*n+1)])
+        r_max = map(max, r_max, r[(2*n+1):(3*n+1)])
+    r_mean = [ x / len(data) for x in r_sum ]
+    return [ r_mean + r_min + r_max ]
+
+if __name__ == "__main__":
+    main()

nilmtools/filter.py

@@ -2,65 +2,244 @@
 
 import nilmdb.client
 from nilmdb.utils.printf import *
-from nilmdb.utils.time import parse_time, format_time
+from nilmdb.utils.time import (parse_time, timestamp_to_human,
+                               timestamp_to_seconds)
+from nilmdb.utils.interval import Interval
 
+import nilmtools
+
+import itertools
 import time
 import sys
 import re
 import argparse
 
+class MissingDestination(Exception):
+    def __init__(self, src, dest):
+        self.src = src
+        self.dest = dest
+        Exception.__init__(self, "destination path " + dest.path + " not found")
+
+class StreamInfo(object):
+    def __init__(self, info):
+        self.info = info
+        try:
+            self.path = info[0]
+            self.layout = info[1]
+            self.layout_type = self.layout.split('_')[0]
+            self.layout_count = int(self.layout.split('_')[1])
+            self.total_count = self.layout_count + 1
+            self.timestamp_min = info[2]
+            self.timestamp_max = info[3]
+            self.rows = info[4]
+            self.seconds = nilmdb.utils.time.timestamp_to_seconds(info[5])
+        except IndexError, TypeError:
+            pass
+
+    def __str__(self):
+        """Print stream info as a string"""
+        return sprintf("%s (%s), %.2fM rows, %.2f hours",
+                       self.path, self.layout, self.rows / 1e6,
+                       self.seconds / 3600.0)
+
 class Filter(object):
 
-    def __init__(self, description = "Filter data"):
-        self.args = None
+    def __init__(self):
+        self._parser = None
+        self._args = None
         self._client = None
-        self.parse_args(description)
+        self._using_client = False
+        self.src = None
+        self.dest = None
 
-    def parse_args(self, description):
+    @property
+    def client(self):
+        if self._using_client:
+            raise Exception("Filter client is in use; make another")
+        return self._client
+
+    def setup_parser(self, description = "Filter data"):
         parser = argparse.ArgumentParser(
-            description = description,
-            formatter_class = argparse.RawDescriptionHelpFormatter)
-        parser.add_argument("-u", "--url", action="store",
-                            default="http://localhost:12380/",
-                            help="Server URL (default: %(default)s)")
-        parser.add_argument("srcpath", action="store",
-                            help="Path of source stream, e.g. /foo/bar")
-        parser.add_argument("destpath", action="store",
-                            help="Path of destination stream, e.g. /foo/bar")
-        self.args = parser.parse_args()
+            formatter_class = argparse.RawDescriptionHelpFormatter,
+            version = nilmtools.__version__,
+            description = description)
+        group = parser.add_argument_group("General filter arguments")
+        group.add_argument("-u", "--url", action="store",
+                           default="http://localhost:12380/",
+                           help="Server URL (default: %(default)s)")
+        group.add_argument("-D", "--dry-run", action="store_true",
+                           default = False,
+                           help="Just print intervals that would be "
+                           "processed")
+        group.add_argument("-s", "--start",
+                           metavar="TIME", type=self.arg_time,
+                           help="Starting timestamp for intervals "
+                           "(free-form, inclusive)")
+        group.add_argument("-e", "--end",
+                           metavar="TIME", type=self.arg_time,
+                           help="Ending timestamp for intervals "
+                           "(free-form, noninclusive)")
+        group.add_argument("srcpath", action="store",
+                           help="Path of source stream, e.g. /foo/bar")
+        group.add_argument("destpath", action="store",
+                           help="Path of destination stream, e.g. /foo/bar")
+        self._parser = parser
+        return parser
 
+    def interval_string(self, interval):
+        return sprintf("[ %s -> %s ]",
+                       timestamp_to_human(interval.start),
+                       timestamp_to_human(interval.end))
+
+    def parse_args(self):
+        args = self._parser.parse_args()
+        self._args = args
         self._client = nilmdb.client.Client(args.url)
 
         if args.srcpath == args.destpath:
             raise Exception("source and destination path must be different")
 
         # Open and print info about the streams
-        def stream_info_string(info):
-            return sprintf("%s (%s), %.2fM rows, %.2f hours\n",
-                           info[0], info[1], info[4] / 1e6, info[5] / 3600)
-
         src = self._client.stream_list(args.srcpath, extended = True)
         if len(src) != 1:
             raise Exception("source path " + args.srcpath + " not found")
-        print "Source:", stream_info_string(src[0])
+        self.src = StreamInfo(src[0])
 
         dest = self._client.stream_list(args.destpath, extended = True)
         if len(dest) != 1:
-            raise Exception("destination path " + args.destpath + " not found")
-        print "  Dest:", stream_info_string(dest[0])
+            raise MissingDestination(self.src, StreamInfo([args.destpath]))
+        self.dest = StreamInfo(dest[0])
+
+        print "Source:", self.src
+        print "  Dest:", self.dest
+
+        if args.dry_run:
+            for interval in self.intervals():
+                print self.interval_string(interval)
+            raise SystemExit(0)
+
+        return args
 
     def intervals(self):
         """Generate all the intervals that this filter should process"""
-        for i in self._client.stream_intervals(
-            args.srcpath, diffpath = args.destpath):
-            yield i
+        self._using_client = True
+        saved_int = None
+        for (start, end) in self._client.stream_intervals(
+            self._args.srcpath, diffpath = self._args.destpath,
+            start = self._args.start, end = self._args.end):
+
+            # Join adjacent intervals
+            if saved_int is not None:
+                if saved_int.end == start:
+                    start = saved_int.start
+                else:
+                    yield saved_int
+            saved_int = Interval(start, end)
+        if saved_int is not None:
+            yield saved_int
+        self._using_client = False
+
+    # Misc helpers
+    def arg_time(self, toparse):
+        """Parse a time string argument"""
+        try:
+            return nilmdb.utils.time.parse_time(toparse)
+        except ValueError as e:
+            raise argparse.ArgumentTypeError(sprintf("%s \"%s\"",
+                                                     str(e), toparse))
+
+    def check_dest_metadata(self, data):
+        """See if the metadata jives, and complain if it doesn't.  If
+        there's no conflict, update the metadata to match 'data'."""
+        metadata = self._client.stream_get_metadata(self._args.destpath)
+        for key in data:
+            wanted = str(data[key])
+            val = metadata.get(key, wanted)
+            if val != wanted and self.dest.rows > 0:
+                m =  "Metadata in destination stream:\n"
+                m += "  %s = %s\n" % (key, val)
+                m += "doesn't match desired data:\n"
+                m += "  %s = %s\n" % (key, wanted)
+                m += "Refusing to change it.  You can change the stream's "
+                m += "metadata manually, or\n"
+                m += "remove existing data from the stream, to prevent "
+                m += "this error.\n"
+                raise Exception(m)
+        # All good -- write the metadata in case it's not already there
+        self._client.stream_update_metadata(self._args.destpath, data)
+
+    # Main processing helper
+    def process_python(self, function, rows, args = None, partial = False):
+        """Process data in chunks of 'rows' data at a time.
+
+        This provides data as nested Python lists and expects the same
+        back.
+
+        function: function to process the data
+        rows: maximum number of rows to pass to 'function' at once
+        args: tuple containing extra arguments to pass to 'function'
+        partial: if true, less than 'rows' may be passed to 'function'.
+                 if false, partial data at the end of an interval will
+                 be dropped.
+
+        'function' should be defined like:
+            function(data, *args)
+        It will be passed a list containing up to 'rows' rows of
+        data from the source stream, and any arguments passed in
+        'args'.  It should transform the data as desired, and return a
+        new list of rdata, which will be inserted into the destination
+        stream.
+        """
+        if args is None:
+            args = []
+        extractor = nilmdb.client.Client(self._args.url).stream_extract
+        inserter = nilmdb.client.Client(self._args.url).stream_insert_context
+        src = self._args.srcpath
+        dest = self._args.destpath
+
+        # Parse input data.  We use homogenous types for now, which
+        # means the timestamp type will be either float or int.
+        if "int" in self.src.layout_type:
+            parser = lambda line: [ int(x) for x in line.split() ]
+        else:
+            parser = lambda line: [ float(x) for x in line.split() ]
+
+        # Format output data.
+        formatter = lambda row: " ".join([repr(x) for x in row]) + "\n"
+
+        for interval in self.intervals():
+            print "Processing", self.interval_string(interval)
+            with inserter(dest, interval.start, interval.end) as insert_ctx:
+                src_array = []
+                for line in extractor(src, interval.start, interval.end):
+                    # Read in data
+                    src_array.append([ float(x) for x in line.split() ])
+
+                    if len(src_array) == rows:
+                        # Pass through filter function
+                        dest_array = function(src_array, *args)
+
+                        # Write result to destination
+                        out = [ formatter(row) for row in dest_array ]
+                        insert_ctx.insert("".join(out))
+
+                        # Clear source array
+                        src_array = []
+
+                # Take care of partial chunk
+                if len(src_array) and partial:
+                    dest_array = function(src_array, *args)
+                    out = [ formatter(row) for row in dest_array ]
+                    insert_ctx.insert("".join(out))
 
 def main():
     # This is just a dummy function; actual filters can use the other
     # functions to prepare stuff, and then do something with the data.
     f = Filter()
-    for interval in f.intervals():
-        print "Generic filter: need to handle interval:", interval
+    parser = f.setup_parser()
+    args = f.parse_args()
+    for i in f.intervals():
+        print "Generic filter: need to handle", f.interval_string(i)
 
 if __name__ == "__main__":
     main()

nilmtools/insert.py

@@ -2,8 +2,11 @@
 
 import nilmdb.client
 from nilmdb.utils.printf import *
-from nilmdb.utils.time import parse_time, format_time
+from nilmdb.utils.time import (parse_time, timestamp_to_human,
+                               timestamp_to_seconds, seconds_to_timestamp,
+                               rate_to_period, now as time_now)
 
+import nilmtools
 import time
 import sys
 import re
@@ -16,7 +19,10 @@ class ParseError(Exception):
         super(ParseError, self).__init__(msg)
 
 def parse_args():
-    parser = argparse.ArgumentParser(description = """\
+    parser = argparse.ArgumentParser(
+        formatter_class = argparse.RawDescriptionHelpFormatter,
+        version = nilmtools.__version__,
+        description = """\
     Insert data from ethstream, either live (using the system time as a
     reference) or prerecorded (using comments in the file as a reference).
 
@@ -24,18 +30,20 @@ def parse_args():
     Small discrepencies between the accumulated timestamps and the
     reference time are ignored; larger discrepencies cause gaps to be
     created in the stream.  Overlapping data returns an error.
-    """, formatter_class = argparse.RawDescriptionHelpFormatter)
+    """)
     parser.add_argument("-u", "--url", action="store",
                         default="http://localhost:12380/",
                         help="NilmDB server URL (default: %(default)s)")
-    parser.add_argument("-r", "--rate", action="store", default=8000, type=float,
+    parser.add_argument("-r", "--rate", action="store", default=8000,
+                        type=float,
                         help="Data rate in Hz (default: %(default)s)")
     parser.add_argument("-l", "--live", action="store_true",
                         help="Live capture; use system time to verify rate")
     parser.add_argument("path", action="store",
                         help="Path of stream, e.g. /foo/bar")
     parser.add_argument("infile", type=argparse.FileType('r'), nargs='*',
-                        default=[sys.stdin], help="Input files (default: stdin)")
+                        default=[sys.stdin],
+                        help="Input files (default: stdin)")
     args = parser.parse_args()
 
     printf("Stream path: %s\n", args.path)
@@ -55,20 +63,22 @@ def main(args = None):
     # data_ts is the timestamp that we'll use for the current line
     data_ts_base = 0
     data_ts_inc = 0
-    data_ts_step = 1.0 / args.rate
+    data_ts_rate = args.rate
 
     # clock_ts is the imprecise "real" timestamp (from the filename,
     # comments, or or system clock)
     clock_ts = None
 
     def print_clock_updated():
-        printf("Clock time updated to %s\n", format_time(clock_ts))
+        printf("Clock time updated to %s\n", timestamp_to_human(clock_ts))
         if data_ts_base != 0:
             diff = data_ts - clock_ts
             if diff >= 0:
-                printf("  (data timestamp ahead by %.6f sec)\n", diff)
+                printf("  (data timestamp ahead by %.6f sec)\n",
+                       timestamp_to_seconds(diff))
             else:
-                printf("  (data timestamp behind by %.6f sec)\n", -diff)
+                printf("  (data timestamp behind by %.6f sec)\n",
+                       timestamp_to_seconds(-diff))
 
     with client.stream_insert_context(args.path) as stream:
         for f in args.infile:
@@ -86,7 +96,7 @@ def main(args = None):
                 # Subtract 1 hour because files are created at the end
                 # of the hour.  Hopefully, we'll be able to use
                 # internal comments and this value won't matter anyway.
-                clock_ts = parse_time(filename).totimestamp() - 3600
+                clock_ts = parse_time(filename) - seconds_to_timestamp(3600)
                 print_clock_updated()
             except ValueError:
                 pass
@@ -95,7 +105,8 @@ def main(args = None):
 
             # Read each line
             for line in f:
-                data_ts = data_ts_base + data_ts_inc * data_ts_step
+                data_ts = data_ts_base + rate_to_period(data_ts_rate,
+                                                        data_ts_inc)
 
                 # If no content other than the newline, skip it
                 if len(line) <= 1:
@@ -104,7 +115,7 @@ def main(args = None):
                 # If line starts with a comment, look for a timestamp
                 if line[0] == '#':
                     try:
-                        clock_ts = parse_time(line[1:]).totimestamp()
+                        clock_ts = parse_time(line[1:])
                         print_clock_updated()
                     except ValueError:
                         pass
@@ -112,30 +123,30 @@ def main(args = None):
 
                 # If inserting live, use clock timestamp
                 if live:
-                    clock_ts = time.time()
+                    clock_ts = time_now()
 
                 # If we have a real timestamp, compare it to the data
                 # timestamp, and make sure things match up.
                 if clock_ts is not None:
-                    if (data_ts - 10) > clock_ts:
+                    if (data_ts - seconds_to_timestamp(10)) > clock_ts:
                         # Accumulated line timestamps are in the future.
                         # If we were to set data_ts=clock_ts, we'd create
                         # an overlap, so we have to just bail out here.
                         err = sprintf("Data is coming in too fast: data time "
                                       "is %s but clock time is only %s",
-                                      format_time(data_ts),
-                                      format_time(clock_ts))
+                                      timestamp_to_human(data_ts),
+                                      timestamp_to_human(clock_ts))
                         raise ParseError(filename, err)
 
-                    if (data_ts + 10) < clock_ts:
+                    if (data_ts + seconds_to_timestamp(10)) < clock_ts:
                         # Accumulated line timetamps are in the past.  We
                         # can just skip some time and leave a gap in the
                         # data.
                         if data_ts_base != 0:
                             printf("Skipping data timestamp forward from "
                                    "%s to %s to match clock time\n",
-                                   format_time(data_ts),
-                                   format_time(clock_ts))
+                                   timestamp_to_human(data_ts),
+                                   timestamp_to_human(clock_ts))
                         stream.finalize()
                         data_ts_base = data_ts = clock_ts
                         data_ts_inc = 0
@@ -160,7 +171,7 @@ def main(args = None):
                     continue
 
                 # Insert it
-                stream.insert("%.6f %s" % (data_ts, line))
+                stream.insert("%d %s" % (data_ts, line))
     print "Done"
 
 if __name__ == "__main__":

scripts/decimate-it.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+for i in "$@"; do
+    nilmtool create /lees-compressor/no-leak/raw/$((4**$i)) float32_18
+    /usr/bin/time -v nilm-decimate /lees-compressor/no-leak/raw/$((4**($i-1))) /lees-compressor/no-leak/raw/$((4**$i))
+done

setup.py

@@ -69,6 +69,7 @@ setup(name='nilmtools',
           'console_scripts': [
               'nilm-decimate = nilmtools.decimate:main',
               'nilm-insert = nilmtools.insert:main',
+              'nilm-copy = nilmtools.copy:main',
               ],
           },
       zip_safe = False,