WIP on getting stream extract to work.

git-svn-id: https://bucket.mit.edu/svn/nilm/nilmdb@10886 ddd99763-3ecb-0310-9145-efcb8ce7c51f

nilmdb/client.py

@@ -124,7 +124,7 @@ class Client(object):
             params["end"] = repr(end)
         return self.http.get_gen("stream/intervals", params)
 
-    def stream_extract(self, path, start = None, end = None, bare = False):
+    def stream_extract(self, path, start = None, end = None):
         """
         Extract data from a stream.  Returns a generator that yields
         lines of ASCII-formatted data that matches the database
@@ -138,5 +138,4 @@ class Client(object):
             params["start"] = repr(start)  # use repr to keep precision
         if end is not None:
             params["end"] = repr(end)
-        params["bare"] = bare
-        return self.http.get_gen("stream/intervals", params)
+        return self.http.get_gen("stream/extract", params)

nilmdb/cmdline/extract.py

@@ -2,6 +2,7 @@ from __future__ import absolute_import
 from nilmdb.printf import *
 import nilmdb.client
 import nilmdb.layout
+import sys
 
 def setup(self, sub):
     cmd = sub.add_parser("extract", help="Extract data",
@@ -40,11 +41,13 @@ def cmd_extract(self):
         printf("# end: %s\n", self.time_string(self.args.end))
 
     printed = False
-    for data in self.client.stream_extract(self.args.path,
-                                           self.args.start,
-                                           self.args.end,
-                                           bare = self.args.bare):
-        sys.stdout.write(data)
+    for dataline in self.client.stream_extract(self.args.path,
+                                               self.args.start,
+                                               self.args.end):
+        if self.args.bare:
+            # Strip timestamp (first element)
+            dataline = ' '.join(dataline.split(' ')[1:])
+        sys.stdout.write(dataline)
         printed = True
     if not printed:
         if self.args.annotate:

nilmdb/server.py

@@ -195,6 +195,8 @@ class Stream(NilmApp):
                 raise cherrypy.HTTPError("400 Bad Request",
                                          "end before start")
         def content(start, end):
+            # Note: disable response.stream below to get better debug info
+            # from tracebacks in this subfunction.
             while True:
                 (intervals, restart) = self.db.stream_intervals(path,start,end)
                 response = ''.join([ json.dumps(i) + "\n" for i in intervals ])
@@ -207,16 +209,37 @@ class Stream(NilmApp):
 
     # /stream/extract?path=/newton/prep&start=1234567890.0&end=1234567899.0
     @cherrypy.expose
-    @cherrypy.tools.json_out()
     def extract(self, path, start, end):
-        # If truncated is true, there is more data after the last row
-        # row given.
+        """
+        Extract data from backend database.  Streams the resulting
+        entries as ASCII text lines separated by newlines.  This may
+        make multiple requests to the nilmdb backend to avoid causing
+        it to block for too long.
+        """
         if start is not None:
             start = float(start)
         if end is not None:
             end = float(end)
-        (intervals, truncated) = self.db.stream_intervals(path, start, end)
-        return (intervals, truncated)
+
+        if start is not None and end is not None:
+            if end < start:
+                raise cherrypy.HTTPError("400 Bad Request",
+                                         "end before start")
+
+        def content(start, end):
+            while True:
+                # Note: disable response.stream below to get better debug info
+                # from tracebacks in this subfunction.
+                (data, restart) = self.db.stream_extract(path, start, end)
+                # data is a list of rows; format it as text
+                response = "timestamp foo bar baz XXX\n"
+                yield response
+                if restart == 0:
+                    break
+                start = restart
+        return content(start, end)
+    extract._cp_config = { 'response.stream': False }
+
 
 class Exiter(object):
     """App that exits the server, for testing"""

tests/test_client.py

@@ -212,3 +212,12 @@ class TestClient(object):
                             retjson=False)
         eq_(x.count('\n'), 2)
         in_("transfer-encoding: chunked", client.http._headers.lower())
+
+        # Make sure that /stream/extract properly returns a
+        # streaming, chunked response.  Pokes around in client.http
+        # internals a bit to look at the response headers.
+        x = client.http.get("stream/extract",
+                            { "path": "/newton/prep",
+                              "start": "123",
+                              "end": "123" }, retjson=False)
+        in_("transfer-encoding: chunked", client.http._headers.lower())

tests/test_cmdline.py

@@ -356,8 +356,8 @@ class TestCmdline(object):
         self.contain("Error getting stream info")
 
         # full dump
-        #self.ok("extract -a /newton/prep --start 2000-01-01 --end 2020-01-01")
-        #self.dump()
+        self.ok("extract -a /newton/prep --start 2000-01-01 --end 2020-01-01")
+        self.dump()
 
     def test_cmdline_9_truncated(self):
         # Test truncated responses by overriding the nilmdb response_size