Compare commits

...

31 Commits

Author SHA1 Message Date
309f38d0ed Merge branch '32bit' 2013-05-08 17:20:31 -04:00
9a27b6ef6a Make rocket code suitable for 32-bit architectures 2013-05-08 16:35:32 -04:00
99532cf9e0 Fix coverage 2013-05-07 23:00:44 -04:00
dfdd0e5c74 Fix line parsing in http client 2013-05-07 22:56:00 -04:00
9a2699adfc Attempt at fixing up more Unicode issues with metadata. 2013-05-07 13:44:03 -04:00
9bbb95b18b Add unicode decode/encode helpers 2013-05-07 12:56:59 -04:00
6bbed322c5 Fix unicode in completion 2013-05-07 12:49:12 -04:00
2317894355 Tweak cache sizes to account for large numbers of decimated tables 2013-05-06 11:54:57 -04:00
539c92226c Add more disk space info 2013-05-06 11:36:28 -04:00
77c766d85d Bump MAX_LAYOUT_COUNT to 1024 2013-05-02 15:27:31 -04:00
49d04db1d6 Allow start==end in stream_insert_context, if no data was provided. 2013-04-11 13:25:37 -04:00
ea838d05ae Warn against reused context managers, and fix broken tests 2013-04-11 13:25:00 -04:00
f2a48bdb2a Test binary extract; fix bugs 2013-04-11 13:24:11 -04:00
6d14e0b8aa Allow binary extract 2013-04-11 11:30:41 -04:00
b31b9327b9 Add tool to fix oversize files (the bug fixed by b98ff13) 2013-04-11 11:02:53 -04:00
b98ff1331a Fix bug where too much data was getting written to each file.
We were still calculating the maximum number of rows correctly,
so the extra data was really extra and would get re-written to the
beginning of the subsequent file.

The only case in which this would lead to database issues is if the
very last file was lengthened incorrectly, and the "nrows" calculation
would therefore be wrong when the database was reopened.  Still, even
in that case, it should just leave a small gap in the data, not cause
any errors.
2013-04-10 23:22:03 -04:00
00e6ba1124 Avoid ENOENT in nilmdb.utils.diskusage.du
ENOENT might show up if we're actively deleting files in the nilmdb
thread while trying to read available space from e.g. the server
thread.
2013-04-10 22:25:22 -04:00
01029230c9 Tweaks to sorting 2013-04-10 19:59:38 -04:00
ecc4e5ef9d Improve test coverage 2013-04-10 19:08:05 -04:00
23f31c472b Split sort_streams_nicely into separate file 2013-04-10 19:07:58 -04:00
a1e2746360 Fix bug in nilmdb.stream_remove with max_removals 2013-04-10 18:37:21 -04:00
1c40d59a52 server: use a generator in /stream/remove
Instead of returning a single number at the end of N nilmdb calls, we
now use a generator that returns one line of text every time there's a
new count of rows removed.  This ensures that the connection will stay
alive for very long removals.
2013-04-10 18:11:58 -04:00
bfb09a189f Fix coverage 2013-04-10 16:33:08 -04:00
416a499866 Support wildcards for destroy 2013-04-10 16:23:07 -04:00
637d193807 Fix unicode processing of command line arguments 2013-04-10 16:22:51 -04:00
b7fa5745ce nilmtool list: allow multiple paths to be supplied 2013-04-10 15:34:33 -04:00
0104c8edd9 nilmtool remove: allow wildcards and multiple paths 2013-04-10 15:27:46 -04:00
cf3b8e787d Add test for wrong number of fields in numpy insert 2013-04-10 15:06:50 -04:00
83d022016c nilmtool list: add new --layout option to show layouts 2013-04-10 14:58:44 -04:00
43b740ecaa nilmtool list: remove old -p parameter 2013-04-10 14:48:23 -04:00
4ce059b920 Give a slightly more clear error on bad array sizes 2013-04-09 19:56:58 -04:00
23 changed files with 501 additions and 224 deletions

View File

@@ -421,3 +421,20 @@ and has all of the same functions. It adds three new functions:
It is significantly faster! It is about 20 times faster to decimate a It is significantly faster! It is about 20 times faster to decimate a
stream with `nilm-decimate` when the filter code is using the new stream with `nilm-decimate` when the filter code is using the new
binary/numpy interface. binary/numpy interface.
WSGI interface & chunked requests
---------------------------------
mod_wsgi requires "WSGIChunkedRequest On" to handle
"Transfer-encoding: Chunked" requests. However, `/stream/insert`
doesn't handle this correctly right now, because:
- The `cherrpy.request.body.read()` call needs to be fixed for chunked requests
- We don't want to just buffer endlessly in the server, and it will
require some thought on how to handle data in chunks (what to do about
interval endpoints).
It is probably better to just keep the endpoint management on the client
side, so leave "WSGIChunkedRequest off" for now.

View File

@@ -0,0 +1,50 @@
#!/usr/bin/python
import os
import sys
import cPickle as pickle
import argparse
import fcntl
import re
from nilmdb.client.numpyclient import layout_to_dtype
parser = argparse.ArgumentParser(
description = """
Fix database corruption where binary writes caused too much data to be
written to the file. Truncates files to the correct length. This was
fixed by b98ff1331a515ad47fd3203615e835b529b039f9.
""")
parser.add_argument("path", action="store", help='Database root path')
parser.add_argument("-y", "--yes", action="store_true", help='Fix them')
args = parser.parse_args()
lock = os.path.join(args.path, "data.lock")
with open(lock, "w") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
fix = {}
for (path, dirs, files) in os.walk(args.path):
if "_format" in files:
with open(os.path.join(path, "_format")) as format:
fmt = pickle.load(format)
rowsize = layout_to_dtype(fmt["layout"]).itemsize
maxsize = rowsize * fmt["rows_per_file"]
fix[path] = maxsize
if maxsize < 128000000: # sanity check
raise Exception("bad maxsize " + str(maxsize))
for fixpath in fix:
for (path, dirs, files) in os.walk(fixpath):
for fn in files:
if not re.match("^[0-9a-f]{4,}$", fn):
continue
fn = os.path.join(path, fn)
size = os.path.getsize(fn)
maxsize = fix[fixpath]
if size > maxsize:
diff = size - maxsize
print diff, "too big:", fn
if args.yes:
with open(fn, "a+") as dbfile:
dbfile.truncate(maxsize)

View File

@@ -6,7 +6,6 @@ import nilmdb.utils
import nilmdb.client.httpclient import nilmdb.client.httpclient
from nilmdb.client.errors import ClientError from nilmdb.client.errors import ClientError
import re
import time import time
import simplejson as json import simplejson as json
import contextlib import contextlib
@@ -66,12 +65,8 @@ class Client(object):
params["layout"] = layout params["layout"] = layout
if extended: if extended:
params["extended"] = 1 params["extended"] = 1
def sort_streams_nicely(x): streams = self.http.get("stream/list", params)
"""Human-friendly sort (/stream/2 before /stream/10)""" return nilmdb.utils.sort.sort_human(streams, key = lambda s: s[0])
num = lambda t: int(t) if t.isdigit() else t
key = lambda k: [ num(c) for c in re.split('([0-9]+)', k[0]) ]
return sorted(x, key = key)
return sort_streams_nicely(self.http.get("stream/list", params))
def stream_get_metadata(self, path, keys = None): def stream_get_metadata(self, path, keys = None):
params = { "path": path } params = { "path": path }
@@ -122,7 +117,10 @@ class Client(object):
params["start"] = timestamp_to_string(start) params["start"] = timestamp_to_string(start)
if end is not None: if end is not None:
params["end"] = timestamp_to_string(end) params["end"] = timestamp_to_string(end)
return self.http.post("stream/remove", params) total = 0
for count in self.http.post_gen("stream/remove", params):
total += int(count)
return total
@contextlib.contextmanager @contextlib.contextmanager
def stream_insert_context(self, path, start = None, end = None): def stream_insert_context(self, path, start = None, end = None):
@@ -146,6 +144,7 @@ class Client(object):
ctx = StreamInserter(self, path, start, end) ctx = StreamInserter(self, path, start, end)
yield ctx yield ctx
ctx.finalize() ctx.finalize()
ctx.destroy()
def stream_insert(self, path, data, start = None, end = None): def stream_insert(self, path, data, start = None, end = None):
"""Insert rows of data into a stream. data should be a string """Insert rows of data into a stream. data should be a string
@@ -295,6 +294,15 @@ class StreamInserter(object):
self._block_data = [] self._block_data = []
self._block_len = 0 self._block_len = 0
self.destroyed = False
def destroy(self):
"""Ensure this object can't be used again without raising
an error"""
def error(*args, **kwargs):
raise Exception("don't reuse this context object")
self._send_block = self.insert = self.finalize = self.send = error
def insert(self, data): def insert(self, data):
"""Insert a chunk of ASCII formatted data in string form. The """Insert a chunk of ASCII formatted data in string form. The
overall data must consist of lines terminated by '\\n'.""" overall data must consist of lines terminated by '\\n'."""
@@ -441,7 +449,7 @@ class StreamInserter(object):
self._interval_start = end_ts self._interval_start = end_ts
# Double check endpoints # Double check endpoints
if start_ts is None or end_ts is None: if (start_ts is None or end_ts is None) or (start_ts == end_ts):
# If the block has no non-comment lines, it's OK # If the block has no non-comment lines, it's OK
try: try:
self._get_first_noncomment(block) self._get_first_noncomment(block)

View File

@@ -123,19 +123,50 @@ class HTTPClient(object):
""" """
(response, isjson) = self._do_req(method, url, query, body, (response, isjson) = self._do_req(method, url, query, body,
stream = True, headers = headers) stream = True, headers = headers)
# Like the iter_lines function in Requests, but only splits on
# the specified line ending.
def lines(source, ending):
pending = None
for chunk in source:
if pending is not None:
chunk = pending + chunk
tmp = chunk.split(ending)
lines = tmp[:-1]
if chunk.endswith(ending):
pending = None
else:
pending = tmp[-1]
for line in lines:
yield line
if pending is not None: # pragma: no cover (missing newline)
yield pending
# Yield the chunks or lines as requested
if binary: if binary:
for chunk in response.iter_content(chunk_size = 65536): for chunk in response.iter_content(chunk_size = 65536):
yield chunk yield chunk
elif isjson: elif isjson:
for line in response.iter_lines(): for line in lines(response.iter_content(chunk_size = 1),
ending = '\r\n'):
yield json.loads(line) yield json.loads(line)
else: else:
for line in response.iter_lines(): for line in lines(response.iter_content(chunk_size = 65536),
ending = '\n'):
yield line yield line
def get_gen(self, url, params = None, binary = False): def get_gen(self, url, params = None, binary = False):
"""Simple GET (parameters in URL) returning a generator""" """Simple GET (parameters in URL) returning a generator"""
return self._req_gen("GET", url, params, binary = binary) return self._req_gen("GET", url, params, binary = binary)
def post_gen(self, url, params = None):
"""Simple POST (parameters in body) returning a generator"""
if self.post_json:
return self._req_gen("POST", url, None,
json.dumps(params),
{ 'Content-type': 'application/json' })
else:
return self._req_gen("POST", url, None, params)
# Not much use for a POST or PUT generator, since they don't # Not much use for a POST or PUT generator, since they don't
# return much data. # return much data.

View File

@@ -98,6 +98,7 @@ class NumpyClient(nilmdb.client.client.Client):
ctx = StreamInserterNumpy(self, path, start, end, dtype) ctx = StreamInserterNumpy(self, path, start, end, dtype)
yield ctx yield ctx
ctx.finalize() ctx.finalize()
ctx.destroy()
def stream_insert_numpy(self, path, data, start = None, end = None, def stream_insert_numpy(self, path, data, start = None, end = None,
layout = None): layout = None):
@@ -133,16 +134,8 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
contiguous interval and may be None. 'dtype' is the Numpy contiguous interval and may be None. 'dtype' is the Numpy
dtype for this stream. dtype for this stream.
""" """
self.last_response = None super(StreamInserterNumpy, self).__init__(client, path, start, end)
self._dtype = dtype self._dtype = dtype
self._client = client
self._path = path
# Start and end for the overall contiguous interval we're
# filling
self._interval_start = start
self._interval_end = end
# Max rows to send at once # Max rows to send at once
self._max_rows = self._max_data // self._dtype.itemsize self._max_rows = self._max_data // self._dtype.itemsize
@@ -162,9 +155,12 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
elif array.ndim == 2: elif array.ndim == 2:
# Convert to structured array # Convert to structured array
sarray = numpy.zeros(array.shape[0], dtype=self._dtype) sarray = numpy.zeros(array.shape[0], dtype=self._dtype)
sarray['timestamp'] = array[:,0] try:
# Need the squeeze in case sarray['data'] is 1 dimensional sarray['timestamp'] = array[:,0]
sarray['data'] = numpy.squeeze(array[:,1:]) # Need the squeeze in case sarray['data'] is 1 dimensional
sarray['data'] = numpy.squeeze(array[:,1:])
except (IndexError, ValueError):
raise ValueError("wrong number of fields for this data type")
array = sarray array = sarray
else: else:
raise ValueError("wrong number of dimensions in array") raise ValueError("wrong number of dimensions in array")
@@ -247,9 +243,12 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
# Next block continues where this one ended # Next block continues where this one ended
self._interval_start = end_ts self._interval_start = end_ts
# If we have no endpoints, it's because we had no data to send. # If we have no endpoints, or equal endpoints, it's OK as long
if start_ts is None or end_ts is None: # as there's no data to send
return if (start_ts is None or end_ts is None) or (start_ts == end_ts):
if len(array) == 0:
return
raise ClientError("have data to send, but invalid start/end times")
# Send it # Send it
data = array.tostring() data = array.tostring()

View File

@@ -72,15 +72,27 @@ class Complete(object): # pragma: no cover
path = parsed_args.path path = parsed_args.path
if not path: if not path:
return [] return []
return ( self.escape(k + '=' + v) results = []
for (k,v) in client.stream_get_metadata(path).iteritems() # prefix comes in as UTF-8, but results need to be Unicode,
if k.startswith(prefix) ) # weird. Still doesn't work in all cases, but that's bugs in
# argcomplete.
prefix = nilmdb.utils.unicode.decode(prefix)
for (k,v) in client.stream_get_metadata(path).iteritems():
kv = self.escape(k + '=' + v)
if kv.startswith(prefix):
results.append(kv)
return results
class Cmdline(object): class Cmdline(object):
def __init__(self, argv = None): def __init__(self, argv = None):
self.argv = argv or sys.argv[1:] self.argv = argv or sys.argv[1:]
try:
# Assume command line arguments are encoded with stdin's encoding,
# and reverse it. Won't be needed in Python 3, but for now..
self.argv = [ x.decode(sys.stdin.encoding) for x in self.argv ]
except Exception: # pragma: no cover
pass
self.client = None self.client = None
self.def_url = os.environ.get("NILMDB_URL", "http://localhost/nilmdb/") self.def_url = os.environ.get("NILMDB_URL", "http://localhost/nilmdb/")
self.subcmd = {} self.subcmd = {}

View File

@@ -1,5 +1,6 @@
from nilmdb.utils.printf import * from nilmdb.utils.printf import *
import nilmdb.client import nilmdb.client
import fnmatch
from argparse import ArgumentDefaultsHelpFormatter as def_form from argparse import ArgumentDefaultsHelpFormatter as def_form
@@ -10,25 +11,39 @@ def setup(self, sub):
Destroy the stream at the specified path. Destroy the stream at the specified path.
The stream must be empty. All metadata The stream must be empty. All metadata
related to the stream is permanently deleted. related to the stream is permanently deleted.
Wildcards and multiple paths are supported.
""") """)
cmd.set_defaults(handler = cmd_destroy) cmd.set_defaults(handler = cmd_destroy)
group = cmd.add_argument_group("Options") group = cmd.add_argument_group("Options")
group.add_argument("-R", "--remove", action="store_true", group.add_argument("-R", "--remove", action="store_true",
help="Remove all data before destroying stream") help="Remove all data before destroying stream")
group.add_argument("-q", "--quiet", action="store_true",
help="Don't display names when destroying "
"multiple paths")
group = cmd.add_argument_group("Required arguments") group = cmd.add_argument_group("Required arguments")
group.add_argument("path", group.add_argument("path", nargs='+',
help="Path of the stream to delete, e.g. /foo/bar", help="Path of the stream to delete, e.g. /foo/bar/*",
).completer = self.complete.path ).completer = self.complete.path
return cmd return cmd
def cmd_destroy(self): def cmd_destroy(self):
"""Destroy stream""" """Destroy stream"""
if self.args.remove: streams = [ s[0] for s in self.client.stream_list() ]
paths = []
for path in self.args.path:
new = fnmatch.filter(streams, path)
if not new:
self.die("error: no stream matched path: %s", path)
paths.extend(new)
for path in paths:
if not self.args.quiet and len(paths) > 1:
printf("Destroying %s\n", path)
try: try:
count = self.client.stream_remove(self.args.path) if self.args.remove:
count = self.client.stream_remove(path)
self.client.stream_destroy(path)
except nilmdb.client.ClientError as e: except nilmdb.client.ClientError as e:
self.die("error removing data: %s", str(e)) self.die("error destroying stream: %s", str(e))
try:
self.client.stream_destroy(self.args.path)
except nilmdb.client.ClientError as e:
self.die("error destroying stream: %s", str(e))

View File

@@ -1,6 +1,7 @@
from __future__ import print_function from __future__ import print_function
from nilmdb.utils.printf import * from nilmdb.utils.printf import *
import nilmdb.client import nilmdb.client
import sys
def setup(self, sub): def setup(self, sub):
cmd = sub.add_parser("extract", help="Extract data", cmd = sub.add_parser("extract", help="Extract data",
@@ -24,6 +25,8 @@ def setup(self, sub):
).completer = self.complete.time ).completer = self.complete.time
group = cmd.add_argument_group("Output format") group = cmd.add_argument_group("Output format")
group.add_argument("-B", "--binary", action="store_true",
help="Raw binary output")
group.add_argument("-b", "--bare", action="store_true", group.add_argument("-b", "--bare", action="store_true",
help="Exclude timestamps from output lines") help="Exclude timestamps from output lines")
group.add_argument("-a", "--annotate", action="store_true", group.add_argument("-a", "--annotate", action="store_true",
@@ -42,6 +45,11 @@ def cmd_extract_verify(self):
if self.args.start > self.args.end: if self.args.start > self.args.end:
self.parser.error("start is after end") self.parser.error("start is after end")
if self.args.binary:
if (self.args.bare or self.args.annotate or self.args.markup or
self.args.timestamp_raw or self.args.count):
self.parser.error("--binary cannot be combined with other options")
def cmd_extract(self): def cmd_extract(self):
streams = self.client.stream_list(self.args.path) streams = self.client.stream_list(self.args.path)
if len(streams) != 1: if len(streams) != 1:
@@ -60,16 +68,23 @@ def cmd_extract(self):
printf("# end: %s\n", time_string(self.args.end)) printf("# end: %s\n", time_string(self.args.end))
printed = False printed = False
if self.args.binary:
printer = sys.stdout.write
else:
printer = print
bare = self.args.bare
count = self.args.count
for dataline in self.client.stream_extract(self.args.path, for dataline in self.client.stream_extract(self.args.path,
self.args.start, self.args.start,
self.args.end, self.args.end,
self.args.count, self.args.count,
self.args.markup): self.args.markup,
if self.args.bare and not self.args.count: self.args.binary):
if bare and not count:
# Strip timestamp (first element). Doesn't make sense # Strip timestamp (first element). Doesn't make sense
# if we are only returning a count. # if we are only returning a count.
dataline = ' '.join(dataline.split(' ')[1:]) dataline = ' '.join(dataline.split(' ')[1:])
print(dataline) printer(dataline)
printed = True printed = True
if not printed: if not printed:
if self.args.annotate: if self.args.annotate:

View File

@@ -21,5 +21,8 @@ def cmd_info(self):
printf("Server URL: %s\n", self.client.geturl()) printf("Server URL: %s\n", self.client.geturl())
dbinfo = self.client.dbinfo() dbinfo = self.client.dbinfo()
printf("Server database path: %s\n", dbinfo["path"]) printf("Server database path: %s\n", dbinfo["path"])
printf("Server database size: %s\n", human_size(dbinfo["size"])) for (desc, field) in [("used by NilmDB", "size"),
printf("Server database free space: %s\n", human_size(dbinfo["free"])) ("used by other", "other"),
("reserved", "reserved"),
("free", "free")]:
printf("Server disk space %s: %s\n", desc, human_size(dbinfo[field]))

View File

@@ -10,22 +10,16 @@ def setup(self, sub):
formatter_class = def_form, formatter_class = def_form,
description=""" description="""
List streams available in the database, List streams available in the database,
optionally filtering by layout or path. Wildcards optionally filtering by path. Wildcards
are accepted. are accepted; non-matching paths or wildcards
are ignored.
""") """)
cmd.set_defaults(verify = cmd_list_verify, cmd.set_defaults(verify = cmd_list_verify,
handler = cmd_list) handler = cmd_list)
group = cmd.add_argument_group("Stream filtering") group = cmd.add_argument_group("Stream filtering")
group.add_argument("-p", "--path", metavar="PATH", default="*", group.add_argument("path", metavar="PATH", default=["*"], nargs='*',
help="Match only this path (-p can be omitted)",
).completer = self.complete.path ).completer = self.complete.path
group.add_argument("path_positional", default="*",
nargs="?", help=argparse.SUPPRESS,
).completer = self.complete.path
group.add_argument("-l", "--layout", default="*",
help="Match only this stream layout",
).completer = self.complete.layout
group = cmd.add_argument_group("Interval info") group = cmd.add_argument_group("Interval info")
group.add_argument("-E", "--ext", action="store_true", group.add_argument("-E", "--ext", action="store_true",
@@ -49,20 +43,12 @@ def setup(self, sub):
group = cmd.add_argument_group("Misc options") group = cmd.add_argument_group("Misc options")
group.add_argument("-T", "--timestamp-raw", action="store_true", group.add_argument("-T", "--timestamp-raw", action="store_true",
help="Show raw timestamps when printing times") help="Show raw timestamps when printing times")
group.add_argument("-l", "--layout", action="store_true",
help="Show layout type next to path name")
return cmd return cmd
def cmd_list_verify(self): def cmd_list_verify(self):
# A hidden "path_positional" argument lets the user leave off the
# "-p" when specifying the path. Handle it here.
got_opt = self.args.path != "*"
got_pos = self.args.path_positional != "*"
if got_pos:
if got_opt:
self.parser.error("too many paths specified")
else:
self.args.path = self.args.path_positional
if self.args.start is not None and self.args.end is not None: if self.args.start is not None and self.args.end is not None:
if self.args.start >= self.args.end: if self.args.start >= self.args.end:
self.parser.error("start must precede end") self.parser.error("start must precede end")
@@ -80,29 +66,33 @@ def cmd_list(self):
else: else:
time_string = nilmdb.utils.time.timestamp_to_human time_string = nilmdb.utils.time.timestamp_to_human
for stream in streams: for argpath in self.args.path:
(path, layout, int_min, int_max, rows, time) = stream[:6] for stream in streams:
if not (fnmatch.fnmatch(path, self.args.path) and (path, layout, int_min, int_max, rows, time) = stream[:6]
fnmatch.fnmatch(layout, self.args.layout)): if not fnmatch.fnmatch(path, argpath):
continue continue
printf("%s %s\n", path, layout) if self.args.layout:
printf("%s %s\n", path, layout)
if self.args.ext:
if int_min is None or int_max is None:
printf(" interval extents: (no data)\n")
else: else:
printf(" interval extents: %s -> %s\n", printf("%s\n", path)
time_string(int_min), time_string(int_max))
printf(" total data: %d rows, %.6f seconds\n",
rows or 0,
nilmdb.utils.time.timestamp_to_seconds(time or 0))
if self.args.detail: if self.args.ext:
printed = False if int_min is None or int_max is None:
for (start, end) in self.client.stream_intervals( printf(" interval extents: (no data)\n")
path, self.args.start, self.args.end): else:
printf(" [ %s -> %s ]\n", time_string(start), time_string(end)) printf(" interval extents: %s -> %s\n",
printed = True time_string(int_min), time_string(int_max))
if not printed: printf(" total data: %d rows, %.6f seconds\n",
printf(" (no intervals)\n") rows or 0,
nilmdb.utils.time.timestamp_to_seconds(time or 0))
if self.args.detail:
printed = False
for (start, end) in self.client.stream_intervals(
path, self.args.start, self.args.end):
printf(" [ %s -> %s ]\n",
time_string(start), time_string(end))
printed = True
if not printed:
printf(" (no intervals)\n")

View File

@@ -41,10 +41,10 @@ def cmd_metadata(self):
if self.args.set is not None or self.args.update is not None: if self.args.set is not None or self.args.update is not None:
# Either set, or update # Either set, or update
if self.args.set is not None: if self.args.set is not None:
keyvals = self.args.set keyvals = map(nilmdb.utils.unicode.decode, self.args.set)
handler = self.client.stream_set_metadata handler = self.client.stream_set_metadata
else: else:
keyvals = self.args.update keyvals = map(nilmdb.utils.unicode.decode, self.args.update)
handler = self.client.stream_update_metadata handler = self.client.stream_update_metadata
# Extract key=value pairs # Extract key=value pairs
@@ -62,7 +62,9 @@ def cmd_metadata(self):
self.die("error setting/updating metadata: %s", str(e)) self.die("error setting/updating metadata: %s", str(e))
elif self.args.delete is not None: elif self.args.delete is not None:
# Delete (by setting values to empty strings) # Delete (by setting values to empty strings)
keys = self.args.delete or None keys = None
if self.args.delete:
keys = map(nilmdb.utils.unicode.decode, self.args.delete)
try: try:
data = self.client.stream_get_metadata(self.args.path, keys) data = self.client.stream_get_metadata(self.args.path, keys)
for key in data: for key in data:
@@ -72,7 +74,9 @@ def cmd_metadata(self):
self.die("error deleting metadata: %s", str(e)) self.die("error deleting metadata: %s", str(e))
else: else:
# Get (or unspecified) # Get (or unspecified)
keys = self.args.get or None keys = None
if self.args.get:
keys = map(nilmdb.utils.unicode.decode, self.args.get)
try: try:
data = self.client.stream_get_metadata(self.args.path, keys) data = self.client.stream_get_metadata(self.args.path, keys)
except nilmdb.client.ClientError as e: except nilmdb.client.ClientError as e:
@@ -81,4 +85,6 @@ def cmd_metadata(self):
# Print nonexistant keys as having empty value # Print nonexistant keys as having empty value
if value is None: if value is None:
value = "" value = ""
printf("%s=%s\n", key, value) printf("%s=%s\n",
nilmdb.utils.unicode.encode(key),
nilmdb.utils.unicode.encode(value))

View File

@@ -1,17 +1,19 @@
from nilmdb.utils.printf import * from nilmdb.utils.printf import *
import nilmdb.client import nilmdb.client
import fnmatch
def setup(self, sub): def setup(self, sub):
cmd = sub.add_parser("remove", help="Remove data", cmd = sub.add_parser("remove", help="Remove data",
description=""" description="""
Remove all data from a specified time range within a Remove all data from a specified time range within a
stream. stream. If multiple streams or wildcards are provided,
the same time range is removed from all streams.
""") """)
cmd.set_defaults(handler = cmd_remove) cmd.set_defaults(handler = cmd_remove)
group = cmd.add_argument_group("Data selection") group = cmd.add_argument_group("Data selection")
group.add_argument("path", group.add_argument("path", nargs='+',
help="Path of stream, e.g. /foo/bar", help="Path of stream, e.g. /foo/bar/*",
).completer = self.complete.path ).completer = self.complete.path
group.add_argument("-s", "--start", required=True, group.add_argument("-s", "--start", required=True,
metavar="TIME", type=self.arg_time, metavar="TIME", type=self.arg_time,
@@ -23,18 +25,31 @@ def setup(self, sub):
).completer = self.complete.time ).completer = self.complete.time
group = cmd.add_argument_group("Output format") group = cmd.add_argument_group("Output format")
group.add_argument("-q", "--quiet", action="store_true",
help="Don't display names when removing "
"from multiple paths")
group.add_argument("-c", "--count", action="store_true", group.add_argument("-c", "--count", action="store_true",
help="Output number of data points removed") help="Output number of data points removed")
return cmd return cmd
def cmd_remove(self): def cmd_remove(self):
streams = [ s[0] for s in self.client.stream_list() ]
paths = []
for path in self.args.path:
new = fnmatch.filter(streams, path)
if not new:
self.die("error: no stream matched path: %s", path)
paths.extend(new)
try: try:
count = self.client.stream_remove(self.args.path, for path in paths:
self.args.start, self.args.end) if not self.args.quiet and len(paths) > 1:
printf("Removing from %s\n", path)
count = self.client.stream_remove(path,
self.args.start, self.args.end)
if self.args.count:
printf("%d\n", count);
except nilmdb.client.ClientError as e: except nilmdb.client.ClientError as e:
self.die("error removing data: %s", str(e)) self.die("error removing data: %s", str(e))
if self.args.count:
printf("%d\n", count)
return 0 return 0

View File

@@ -19,8 +19,8 @@ from . import rocket
# Up to 256 open file descriptors at any given time. # Up to 256 open file descriptors at any given time.
# These variables are global so they can be used in the decorator arguments. # These variables are global so they can be used in the decorator arguments.
table_cache_size = 16 table_cache_size = 32
fd_cache_size = 16 fd_cache_size = 8
@nilmdb.utils.must_close(wrap_verify = False) @nilmdb.utils.must_close(wrap_verify = False)
class BulkData(object): class BulkData(object):

View File

@@ -176,7 +176,7 @@ class NilmDB(object):
raise NilmDBError("start must precede end") raise NilmDBError("start must precede end")
return (start, end) return (start, end)
@nilmdb.utils.lru_cache(size = 16) @nilmdb.utils.lru_cache(size = 64)
def _get_intervals(self, stream_id): def _get_intervals(self, stream_id):
""" """
Return a mutable IntervalSet corresponding to the given stream ID. Return a mutable IntervalSet corresponding to the given stream ID.
@@ -675,6 +675,7 @@ class NilmDB(object):
# Count how many were removed # Count how many were removed
removed += row_end - row_start removed += row_end - row_start
remaining -= row_end - row_start
if restart is not None: if restart is not None:
break break

View File

@@ -5,6 +5,9 @@
#include <ctype.h> #include <ctype.h>
#include <stdint.h> #include <stdint.h>
#define __STDC_FORMAT_MACROS
#include <inttypes.h>
/* Values missing from stdint.h */ /* Values missing from stdint.h */
#define UINT8_MIN 0 #define UINT8_MIN 0
#define UINT16_MIN 0 #define UINT16_MIN 0
@@ -19,16 +22,9 @@
typedef int64_t timestamp_t; typedef int64_t timestamp_t;
/* This code probably needs to be double-checked for the case where
sizeof(long) != 8, so enforce that here with something that will
fail at build time. We assume that the python integer type can
hold an int64_t. */
const static char __long_ok[1 - 2*!(sizeof(int64_t) ==
sizeof(long int))] = { 0 };
/* Somewhat arbitrary, just so we can use fixed sizes for strings /* Somewhat arbitrary, just so we can use fixed sizes for strings
etc. */ etc. */
static const int MAX_LAYOUT_COUNT = 128; static const int MAX_LAYOUT_COUNT = 1024;
/* Error object and constants */ /* Error object and constants */
static PyObject *ParseError; static PyObject *ParseError;
@@ -58,7 +54,7 @@ static PyObject *raise_str(int line, int col, int code, const char *string)
static PyObject *raise_int(int line, int col, int code, int64_t num) static PyObject *raise_int(int line, int col, int code, int64_t num)
{ {
PyObject *o; PyObject *o;
o = Py_BuildValue("(iiil)", line, col, code, num); o = Py_BuildValue("(iiiL)", line, col, code, (long long)num);
if (o != NULL) { if (o != NULL) {
PyErr_SetObject(ParseError, o); PyErr_SetObject(ParseError, o);
Py_DECREF(o); Py_DECREF(o);
@@ -249,11 +245,11 @@ static PyObject *Rocket_get_file_size(Rocket *self)
/**** /****
* Append from string * Append from string
*/ */
static inline long int strtol10(const char *nptr, char **endptr) { static inline long int strtoll10(const char *nptr, char **endptr) {
return strtol(nptr, endptr, 10); return strtoll(nptr, endptr, 10);
} }
static inline long int strtoul10(const char *nptr, char **endptr) { static inline long int strtoull10(const char *nptr, char **endptr) {
return strtoul(nptr, endptr, 10); return strtoull(nptr, endptr, 10);
} }
/* .append_string(count, data, offset, linenum, start, end, last_timestamp) */ /* .append_string(count, data, offset, linenum, start, end, last_timestamp) */
@@ -264,6 +260,7 @@ static PyObject *Rocket_append_string(Rocket *self, PyObject *args)
int offset; int offset;
const char *linestart; const char *linestart;
int linenum; int linenum;
long long ll1, ll2, ll3;
timestamp_t start; timestamp_t start;
timestamp_t end; timestamp_t end;
timestamp_t last_timestamp; timestamp_t last_timestamp;
@@ -280,10 +277,13 @@ static PyObject *Rocket_append_string(Rocket *self, PyObject *args)
but we need the null termination for strto*. If we had but we need the null termination for strto*. If we had
strnto* that took a length, we could use t# and not require strnto* that took a length, we could use t# and not require
a copy. */ a copy. */
if (!PyArg_ParseTuple(args, "isiilll:append_string", &count, if (!PyArg_ParseTuple(args, "isiiLLL:append_string", &count,
&data, &offset, &linenum, &data, &offset, &linenum,
&start, &end, &last_timestamp)) &ll1, &ll2, &ll3))
return NULL; return NULL;
start = ll1;
end = ll2;
last_timestamp = ll3;
/* Skip spaces, but don't skip over a newline. */ /* Skip spaces, but don't skip over a newline. */
#define SKIP_BLANK(buf) do { \ #define SKIP_BLANK(buf) do { \
@@ -372,14 +372,14 @@ static PyObject *Rocket_append_string(Rocket *self, PyObject *args)
goto extra_data_on_line; \ goto extra_data_on_line; \
break break
CS(INT8, strtol10, t64.i, t8.i, t8.u, , 1); CS(INT8, strtoll10, t64.i, t8.i, t8.u, , 1);
CS(UINT8, strtoul10, t64.u, t8.u, t8.u, , 1); CS(UINT8, strtoull10, t64.u, t8.u, t8.u, , 1);
CS(INT16, strtol10, t64.i, t16.i, t16.u, le16toh, 2); CS(INT16, strtoll10, t64.i, t16.i, t16.u, le16toh, 2);
CS(UINT16, strtoul10, t64.u, t16.u, t16.u, le16toh, 2); CS(UINT16, strtoull10, t64.u, t16.u, t16.u, le16toh, 2);
CS(INT32, strtol10, t64.i, t32.i, t32.u, le32toh, 4); CS(INT32, strtoll10, t64.i, t32.i, t32.u, le32toh, 4);
CS(UINT32, strtoul10, t64.u, t32.u, t32.u, le32toh, 4); CS(UINT32, strtoull10, t64.u, t32.u, t32.u, le32toh, 4);
CS(INT64, strtol10, t64.i, t64.i, t64.u, le64toh, 8); CS(INT64, strtoll10, t64.i, t64.i, t64.u, le64toh, 8);
CS(UINT64, strtoul10, t64.u, t64.u, t64.u, le64toh, 8); CS(UINT64, strtoull10, t64.u, t64.u, t64.u, le64toh, 8);
CS(FLOAT32, strtod, t64.d, t32.f, t32.u, le32toh, 4); CS(FLOAT32, strtod, t64.d, t32.f, t32.u, le32toh, 4);
CS(FLOAT64, strtod, t64.d, t64.d, t64.u, le64toh, 8); CS(FLOAT64, strtod, t64.d, t64.d, t64.u, le64toh, 8);
#undef CS #undef CS
@@ -397,7 +397,8 @@ static PyObject *Rocket_append_string(Rocket *self, PyObject *args)
/* Build return value and return */ /* Build return value and return */
offset = buf - data; offset = buf - data;
PyObject *o; PyObject *o;
o = Py_BuildValue("(iili)", written, offset, last_timestamp, linenum); o = Py_BuildValue("(iiLi)", written, offset,
(long long)last_timestamp, linenum);
return o; return o;
err: err:
PyErr_SetFromErrno(PyExc_OSError); PyErr_SetFromErrno(PyExc_OSError);
@@ -431,14 +432,18 @@ static PyObject *Rocket_append_binary(Rocket *self, PyObject *args)
int data_len; int data_len;
int linenum; int linenum;
int offset; int offset;
long long ll1, ll2, ll3;
timestamp_t start; timestamp_t start;
timestamp_t end; timestamp_t end;
timestamp_t last_timestamp; timestamp_t last_timestamp;
if (!PyArg_ParseTuple(args, "it#iilll:append_binary", if (!PyArg_ParseTuple(args, "it#iiLLL:append_binary",
&count, &data, &data_len, &offset, &count, &data, &data_len, &offset,
&linenum, &start, &end, &last_timestamp)) &linenum, &ll1, &ll2, &ll3))
return NULL; return NULL;
start = ll1;
end = ll2;
last_timestamp = ll3;
/* Advance to offset */ /* Advance to offset */
if (offset > data_len) if (offset > data_len)
@@ -468,7 +473,7 @@ static PyObject *Rocket_append_binary(Rocket *self, PyObject *args)
} }
/* Write binary data */ /* Write binary data */
if (fwrite(data, data_len, 1, self->file) != 1) { if (fwrite(data, self->binary_size, rows, self->file) != rows) {
PyErr_SetFromErrno(PyExc_OSError); PyErr_SetFromErrno(PyExc_OSError);
return NULL; return NULL;
} }
@@ -476,8 +481,8 @@ static PyObject *Rocket_append_binary(Rocket *self, PyObject *args)
/* Build return value and return */ /* Build return value and return */
PyObject *o; PyObject *o;
o = Py_BuildValue("(iili)", rows, offset + rows * self->binary_size, o = Py_BuildValue("(iiLi)", rows, offset + rows * self->binary_size,
last_timestamp, linenum); (long long)last_timestamp, linenum);
return o; return o;
} }
@@ -534,7 +539,7 @@ static PyObject *Rocket_extract_string(Rocket *self, PyObject *args)
if (fread(&t64.u, 8, 1, self->file) != 1) if (fread(&t64.u, 8, 1, self->file) != 1)
goto err; goto err;
t64.u = le64toh(t64.u); t64.u = le64toh(t64.u);
ret = sprintf(&str[len], "%ld", t64.i); ret = sprintf(&str[len], "%" PRId64, t64.i);
if (ret <= 0) if (ret <= 0)
goto err; goto err;
len += ret; len += ret;
@@ -556,14 +561,14 @@ static PyObject *Rocket_extract_string(Rocket *self, PyObject *args)
len += ret; \ len += ret; \
} \ } \
break break
CASE(INT8, "%hhd", t8.i, t8.u, , 1); CASE(INT8, "%" PRId8, t8.i, t8.u, , 1);
CASE(UINT8, "%hhu", t8.u, t8.u, , 1); CASE(UINT8, "%" PRIu8, t8.u, t8.u, , 1);
CASE(INT16, "%hd", t16.i, t16.u, le16toh, 2); CASE(INT16, "%" PRId16, t16.i, t16.u, le16toh, 2);
CASE(UINT16, "%hu", t16.u, t16.u, le16toh, 2); CASE(UINT16, "%" PRIu16, t16.u, t16.u, le16toh, 2);
CASE(INT32, "%d", t32.i, t32.u, le32toh, 4); CASE(INT32, "%" PRId32, t32.i, t32.u, le32toh, 4);
CASE(UINT32, "%u", t32.u, t32.u, le32toh, 4); CASE(UINT32, "%" PRIu32, t32.u, t32.u, le32toh, 4);
CASE(INT64, "%ld", t64.i, t64.u, le64toh, 8); CASE(INT64, "%" PRId64, t64.i, t64.u, le64toh, 8);
CASE(UINT64, "%lu", t64.u, t64.u, le64toh, 8); CASE(UINT64, "%" PRIu64, t64.u, t64.u, le64toh, 8);
/* These next two are a bit debatable. floats /* These next two are a bit debatable. floats
are 6-9 significant figures, so we print 7. are 6-9 significant figures, so we print 7.
Doubles are 15-19, so we print 17. This is Doubles are 15-19, so we print 17. This is
@@ -653,7 +658,7 @@ static PyObject *Rocket_extract_timestamp(Rocket *self, PyObject *args)
/* Convert and return */ /* Convert and return */
t64.u = le64toh(t64.u); t64.u = le64toh(t64.u);
return Py_BuildValue("l", t64.i); return Py_BuildValue("L", (long long)t64.i);
} }
/**** /****

View File

@@ -167,9 +167,13 @@ class Root(NilmApp):
"""Return a dictionary with the database path, """Return a dictionary with the database path,
size of the database in bytes, and free disk space in bytes""" size of the database in bytes, and free disk space in bytes"""
path = self.db.get_basepath() path = self.db.get_basepath()
usage = psutil.disk_usage(path)
dbsize = nilmdb.utils.du(path)
return { "path": path, return { "path": path,
"size": nilmdb.utils.du(path), "size": dbsize,
"free": psutil.disk_usage(path).free } "other": usage.used - dbsize,
"reserved": usage.total - usage.used - usage.free,
"free": usage.free }
class Stream(NilmApp): class Stream(NilmApp):
"""Stream-specific operations""" """Stream-specific operations"""
@@ -347,24 +351,34 @@ class Stream(NilmApp):
# /stream/remove?path=/newton/prep&start=1234567890.0&end=1234567899.0 # /stream/remove?path=/newton/prep&start=1234567890.0&end=1234567899.0
@cherrypy.expose @cherrypy.expose
@cherrypy.tools.json_in() @cherrypy.tools.json_in()
@cherrypy.tools.json_out()
@exception_to_httperror(NilmDBError)
@cherrypy.tools.CORS_allow(methods = ["POST"]) @cherrypy.tools.CORS_allow(methods = ["POST"])
@chunked_response
@response_type("application/x-json-stream")
def remove(self, path, start = None, end = None): def remove(self, path, start = None, end = None):
""" """
Remove data from the backend database. Removes all data in Remove data from the backend database. Removes all data in
the interval [start, end). Returns the number of data points the interval [start, end).
removed.
Returns the number of data points removed. Since this is a potentially
long-running operation, multiple numbers may be returned as the
data gets removed from the backend database. The total number of
points removed is the sum of all of these numbers.
""" """
(start, end) = self._get_times(start, end) (start, end) = self._get_times(start, end)
total_removed = 0
while True: if len(self.db.stream_list(path = path)) != 1:
(removed, restart) = self.db.stream_remove(path, start, end) raise cherrypy.HTTPError("404", "No such stream: " + path)
total_removed += removed
if restart is None: @workaround_cp_bug_1200
break def content(start, end):
start = restart # Note: disable chunked responses to see tracebacks from here.
return total_removed while True:
(removed, restart) = self.db.stream_remove(path, start, end)
yield json.dumps(removed) + "\r\n"
if restart is None:
break
start = restart
return content(start, end)
# /stream/intervals?path=/newton/prep # /stream/intervals?path=/newton/prep
# /stream/intervals?path=/newton/prep&start=1234567890.0&end=1234567899.0 # /stream/intervals?path=/newton/prep&start=1234567890.0&end=1234567899.0

View File

@@ -13,3 +13,5 @@ import nilmdb.utils.time
import nilmdb.utils.iterator import nilmdb.utils.iterator
import nilmdb.utils.interval import nilmdb.utils.interval
import nilmdb.utils.lock import nilmdb.utils.lock
import nilmdb.utils.sort
import nilmdb.utils.unicode

View File

@@ -1,4 +1,5 @@
import os import os
import errno
from math import log from math import log
def human_size(num): def human_size(num):
@@ -16,10 +17,17 @@ def human_size(num):
return '1 byte' return '1 byte'
def du(path): def du(path):
"""Like du -sb, returns total size of path in bytes.""" """Like du -sb, returns total size of path in bytes. Ignore
size = os.path.getsize(path) errors that might occur if we encounter broken symlinks or
if os.path.isdir(path): files in the process of being removed."""
for thisfile in os.listdir(path): try:
filepath = os.path.join(path, thisfile) size = os.path.getsize(path)
size += du(filepath) if os.path.isdir(path):
return size for thisfile in os.listdir(path):
filepath = os.path.join(path, thisfile)
size += du(filepath)
return size
except OSError as e: # pragma: no cover
if e.errno != errno.ENOENT:
raise
return 0

18
nilmdb/utils/sort.py Normal file
View File

@@ -0,0 +1,18 @@
import re
def sort_human(items, key = None):
"""Human-friendly sort (/stream/2 before /stream/10)"""
def to_num(val):
try:
return int(val)
except Exception:
return val
def human_key(text):
if key:
text = key(text)
# Break into character and numeric chunks.
chunks = re.split(r'([0-9]+)', text)
return [ to_num(c) for c in chunks ]
return sorted(items, key = human_key)

22
nilmdb/utils/unicode.py Normal file
View File

@@ -0,0 +1,22 @@
def encode(u):
"""Try to encode something from Unicode to a string using the
default encoding. If it fails, try encoding as UTF-8."""
if not isinstance(u, unicode):
return u
try:
return u.encode()
except UnicodeEncodeError:
return u.encode("utf-8")
def decode(s):
"""Try to decode someting from string to Unicode using the
default encoding. If it fails, try decoding as UTF-8."""
if isinstance(s, unicode):
return s
try:
return s.decode()
except UnicodeDecodeError:
try:
return s.decode("utf-8")
except UnicodeDecodeError:
return s # best we can do

View File

@@ -105,16 +105,19 @@ class TestClient(object):
client.http.post("/stream/list") client.http.post("/stream/list")
client = nilmdb.client.Client(url = testurl) client = nilmdb.client.Client(url = testurl)
# Create three streams # Create four streams
client.stream_create("/newton/prep", "float32_8") client.stream_create("/newton/prep", "float32_8")
client.stream_create("/newton/raw", "uint16_6") client.stream_create("/newton/raw", "uint16_6")
client.stream_create("/newton/zzz/rawnotch", "uint16_9") client.stream_create("/newton/zzz/rawnotch2", "uint16_9")
client.stream_create("/newton/zzz/rawnotch11", "uint16_9")
# Verify we got 3 streams # Verify we got 4 streams in the right order
eq_(client.stream_list(), [ ["/newton/prep", "float32_8"], eq_(client.stream_list(), [ ["/newton/prep", "float32_8"],
["/newton/raw", "uint16_6"], ["/newton/raw", "uint16_6"],
["/newton/zzz/rawnotch", "uint16_9"] ["/newton/zzz/rawnotch2", "uint16_9"],
["/newton/zzz/rawnotch11", "uint16_9"]
]) ])
# Match just one type or one path # Match just one type or one path
eq_(client.stream_list(layout="uint16_6"), eq_(client.stream_list(layout="uint16_6"),
[ ["/newton/raw", "uint16_6"] ]) [ ["/newton/raw", "uint16_6"] ])
@@ -327,6 +330,10 @@ class TestClient(object):
2525.169921875, 8350.83984375, 3724.699951171875, 2525.169921875, 8350.83984375, 3724.699951171875,
1355.3399658203125, 2039.0)) 1355.3399658203125, 2039.0))
# Just get some coverage
with assert_raises(ClientError) as e:
client.http.post("/stream/remove", { "path": "/none" })
client.close() client.close()
def test_client_06_generators(self): def test_client_06_generators(self):
@@ -613,8 +620,12 @@ class TestClient(object):
with client.stream_insert_context("/empty/test", end = 950): with client.stream_insert_context("/empty/test", end = 950):
pass pass
# Equal start and end is OK as long as there's no data
with client.stream_insert_context("/empty/test", start=9, end=9):
pass
# Try various things that might cause problems # Try various things that might cause problems
with client.stream_insert_context("/empty/test", 1000, 1050): with client.stream_insert_context("/empty/test", 1000, 1050) as ctx:
ctx.finalize() # inserts [1000, 1050] ctx.finalize() # inserts [1000, 1050]
ctx.finalize() # nothing ctx.finalize() # nothing
ctx.finalize() # nothing ctx.finalize() # nothing

View File

@@ -88,7 +88,7 @@ class TestCmdline(object):
sys.exit(0) sys.exit(0)
except SystemExit as e: except SystemExit as e:
exitcode = e.code exitcode = e.code
captured = outfile.getvalue() captured = nilmdb.utils.unicode.decode(outfile.getvalue())
self.captured = captured self.captured = captured
self.exitcode = exitcode self.exitcode = exitcode
@@ -245,8 +245,10 @@ class TestCmdline(object):
self.contain("Client version: " + nilmdb.__version__) self.contain("Client version: " + nilmdb.__version__)
self.contain("Server version: " + test_server.version) self.contain("Server version: " + test_server.version)
self.contain("Server database path") self.contain("Server database path")
self.contain("Server database size") self.contain("Server disk space used by NilmDB")
self.contain("Server database free space") self.contain("Server disk space used by other")
self.contain("Server disk space reserved")
self.contain("Server disk space free")
def test_04_createlist(self): def test_04_createlist(self):
# Basic stream tests, like those in test_client. # Basic stream tests, like those in test_client.
@@ -300,38 +302,19 @@ class TestCmdline(object):
# Verify we got those 3 streams and they're returned in # Verify we got those 3 streams and they're returned in
# alphabetical order. # alphabetical order.
self.ok("list") self.ok("list -l")
self.match("/newton/prep float32_8\n" self.match("/newton/prep float32_8\n"
"/newton/raw uint16_6\n" "/newton/raw uint16_6\n"
"/newton/zzz/rawnotch uint16_9\n") "/newton/zzz/rawnotch uint16_9\n")
# Match just one type or one path. Also check # Match just one type or one path. Also check
# that --path is optional # that --path is optional
self.ok("list --path /newton/raw") self.ok("list --layout /newton/raw")
self.match("/newton/raw uint16_6\n")
self.ok("list /newton/raw")
self.match("/newton/raw uint16_6\n")
self.fail("list -p /newton/raw /newton/raw")
self.contain("too many paths")
self.ok("list --layout uint16_6")
self.match("/newton/raw uint16_6\n") self.match("/newton/raw uint16_6\n")
# Wildcard matches # Wildcard matches
self.ok("list --layout uint16*") self.ok("list *zzz*")
self.match("/newton/raw uint16_6\n" self.match("/newton/zzz/rawnotch\n")
"/newton/zzz/rawnotch uint16_9\n")
self.ok("list --path *zzz* --layout uint16*")
self.match("/newton/zzz/rawnotch uint16_9\n")
self.ok("list *zzz* --layout uint16*")
self.match("/newton/zzz/rawnotch uint16_9\n")
self.ok("list --path *zzz* --layout float32*")
self.match("")
# reversed range # reversed range
self.fail("list /newton/prep --start 2020-01-01 --end 2000-01-01") self.fail("list /newton/prep --start 2020-01-01 --end 2000-01-01")
@@ -497,28 +480,28 @@ class TestCmdline(object):
self.ok("list --detail") self.ok("list --detail")
lines_(self.captured, 8) lines_(self.captured, 8)
self.ok("list --detail --path *prep") self.ok("list --detail *prep")
lines_(self.captured, 4) lines_(self.captured, 4)
self.ok("list --detail --path *prep --start='23 Mar 2012 10:02'") self.ok("list --detail *prep --start='23 Mar 2012 10:02'")
lines_(self.captured, 3) lines_(self.captured, 3)
self.ok("list --detail --path *prep --start='23 Mar 2012 10:05'") self.ok("list --detail *prep --start='23 Mar 2012 10:05'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15'") self.ok("list --detail *prep --start='23 Mar 2012 10:05:15'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.contain("10:05:15.000") self.contain("10:05:15.000")
self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15.50'") self.ok("list --detail *prep --start='23 Mar 2012 10:05:15.50'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.contain("10:05:15.500") self.contain("10:05:15.500")
self.ok("list --detail --path *prep --start='23 Mar 2012 19:05:15.50'") self.ok("list --detail *prep --start='23 Mar 2012 19:05:15.50'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.contain("no intervals") self.contain("no intervals")
self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15.50'" self.ok("list --detail *prep --start='23 Mar 2012 10:05:15.50'"
+ " --end='23 Mar 2012 10:05:15.51'") + " --end='23 Mar 2012 10:05:15.51'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.contain("10:05:15.500") self.contain("10:05:15.500")
@@ -527,15 +510,15 @@ class TestCmdline(object):
lines_(self.captured, 8) lines_(self.captured, 8)
# Verify the "raw timestamp" output # Verify the "raw timestamp" output
self.ok("list --detail --path *prep --timestamp-raw " self.ok("list --detail *prep --timestamp-raw "
"--start='23 Mar 2012 10:05:15.50'") "--start='23 Mar 2012 10:05:15.50'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.contain("[ 1332497115500000 -> 1332497160000000 ]") self.contain("[ 1332497115500000 -> 1332497160000000 ]")
# bad time # bad time
self.fail("list --detail --path *prep -T --start='9332497115.612'") self.fail("list --detail *prep -T --start='9332497115.612'")
# good time # good time
self.ok("list --detail --path *prep -T --start='1332497115.612'") self.ok("list --detail *prep -T --start='1332497115.612'")
lines_(self.captured, 2) lines_(self.captured, 2)
self.contain("[ 1332497115612000 -> 1332497160000000 ]") self.contain("[ 1332497115612000 -> 1332497160000000 ]")
@@ -620,6 +603,14 @@ class TestCmdline(object):
self.ok("extract -c /newton/prep --start 2000-01-01 --end 2020-01-01") self.ok("extract -c /newton/prep --start 2000-01-01 --end 2020-01-01")
self.match("43200\n") self.match("43200\n")
# test binary mode
self.fail("extract -c -B /newton/prep -s min -e max")
self.contain("binary cannot be combined")
self.fail("extract -m -B /newton/prep -s min -e max")
self.contain("binary cannot be combined")
self.ok("extract -B /newton/prep -s min -e max")
eq_(len(self.captured), 43200 * (8 + 8*4))
# markup for 3 intervals, plus extra markup lines whenever we had # markup for 3 intervals, plus extra markup lines whenever we had
# a "restart" from the nilmdb.stream_extract function # a "restart" from the nilmdb.stream_extract function
self.ok("extract -m /newton/prep --start 2000-01-01 --end 2020-01-01") self.ok("extract -m /newton/prep --start 2000-01-01 --end 2020-01-01")
@@ -639,7 +630,7 @@ class TestCmdline(object):
# Try nonexistent stream # Try nonexistent stream
self.fail("remove /no/such/foo --start 2000-01-01 --end 2020-01-01") self.fail("remove /no/such/foo --start 2000-01-01 --end 2020-01-01")
self.contain("No stream at path") self.contain("no stream matched path")
# empty or backward ranges return errors # empty or backward ranges return errors
self.fail("remove /newton/prep --start 2020-01-01 --end 2000-01-01") self.fail("remove /newton/prep --start 2020-01-01 --end 2000-01-01")
@@ -667,9 +658,14 @@ class TestCmdline(object):
"--start '23 Mar 2022 20:00:30' " + "--start '23 Mar 2022 20:00:30' " +
"--end '23 Mar 2022 20:00:31'") "--end '23 Mar 2022 20:00:31'")
self.match("0\n") self.match("0\n")
self.ok("remove -c /newton/prep /newton/pre* " +
"--start '23 Mar 2022 20:00:30' " +
"--end '23 Mar 2022 20:00:31'")
self.match("Removing from /newton/prep\n0\n" +
"Removing from /newton/prep\n0\n")
# Make sure we have the data we expect # Make sure we have the data we expect
self.ok("list --detail /newton/prep") self.ok("list -l --detail /newton/prep")
self.match("/newton/prep float32_8\n" + self.match("/newton/prep float32_8\n" +
" [ Fri, 23 Mar 2012 10:00:00.000000 +0000" " [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
" -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n" " -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n"
@@ -704,7 +700,7 @@ class TestCmdline(object):
self.match("24000\n") self.match("24000\n")
# See the missing chunks in list output # See the missing chunks in list output
self.ok("list --detail /newton/prep") self.ok("list --layout --detail /newton/prep")
self.match("/newton/prep float32_8\n" + self.match("/newton/prep float32_8\n" +
" [ Fri, 23 Mar 2012 10:00:00.000000 +0000" " [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
" -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n" " -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n"
@@ -718,7 +714,7 @@ class TestCmdline(object):
# Remove all data, verify it's missing # Remove all data, verify it's missing
self.ok("remove /newton/prep --start 2000-01-01 --end 2020-01-01") self.ok("remove /newton/prep --start 2000-01-01 --end 2020-01-01")
self.match("") # no count requested this time self.match("") # no count requested this time
self.ok("list --detail /newton/prep") self.ok("list -l --detail /newton/prep")
self.match("/newton/prep float32_8\n" + self.match("/newton/prep float32_8\n" +
" (no intervals)\n") " (no intervals)\n")
@@ -736,16 +732,16 @@ class TestCmdline(object):
self.contain("too few arguments") self.contain("too few arguments")
self.fail("destroy /no/such/stream") self.fail("destroy /no/such/stream")
self.contain("No stream at path") self.contain("no stream matched path")
self.fail("destroy -R /no/such/stream") self.fail("destroy -R /no/such/stream")
self.contain("No stream at path") self.contain("no stream matched path")
self.fail("destroy asdfasdf") self.fail("destroy asdfasdf")
self.contain("No stream at path") self.contain("no stream matched path")
# From previous tests, we have: # From previous tests, we have:
self.ok("list") self.ok("list -l")
self.match("/newton/prep float32_8\n" self.match("/newton/prep float32_8\n"
"/newton/raw uint16_6\n" "/newton/raw uint16_6\n"
"/newton/zzz/rawnotch uint16_9\n") "/newton/zzz/rawnotch uint16_9\n")
@@ -761,13 +757,13 @@ class TestCmdline(object):
lines_(self.captured, 7) lines_(self.captured, 7)
# Destroy for real # Destroy for real
self.ok("destroy -R /newton/prep") self.ok("destroy -R /n*/prep")
self.ok("list") self.ok("list -l")
self.match("/newton/raw uint16_6\n" self.match("/newton/raw uint16_6\n"
"/newton/zzz/rawnotch uint16_9\n") "/newton/zzz/rawnotch uint16_9\n")
self.ok("destroy /newton/zzz/rawnotch") self.ok("destroy /newton/zzz/rawnotch")
self.ok("list") self.ok("list -l")
self.match("/newton/raw uint16_6\n") self.match("/newton/raw uint16_6\n")
self.ok("destroy /newton/raw") self.ok("destroy /newton/raw")
@@ -786,18 +782,17 @@ class TestCmdline(object):
self.ok("list") self.ok("list")
self.contain(path) self.contain(path)
# Make sure it was created empty # Make sure it was created empty
self.ok("list --detail --path " + path) self.ok("list --detail " + path)
self.contain("(no intervals)") self.contain("(no intervals)")
def test_12_unicode(self): def test_12_unicode(self):
# Unicode paths. # Unicode paths.
self.ok("destroy /newton/asdf/qwer") self.ok("destroy /newton/asdf/qwer")
self.ok("destroy /newton/prep") self.ok("destroy /newton/prep /newton/raw")
self.ok("destroy /newton/raw")
self.ok("destroy /newton/zzz") self.ok("destroy /newton/zzz")
self.ok(u"create /düsseldorf/raw uint16_6") self.ok(u"create /düsseldorf/raw uint16_6")
self.ok("list --detail") self.ok("list -l --detail")
self.contain(u"/düsseldorf/raw uint16_6") self.contain(u"/düsseldorf/raw uint16_6")
self.contain("(no intervals)") self.contain("(no intervals)")
@@ -883,7 +878,7 @@ class TestCmdline(object):
du_before = nilmdb.utils.diskusage.du(testdb) du_before = nilmdb.utils.diskusage.du(testdb)
# Make sure we have the data we expect # Make sure we have the data we expect
self.ok("list --detail") self.ok("list -l --detail")
self.match("/newton/prep float32_8\n" + self.match("/newton/prep float32_8\n" +
" [ Fri, 23 Mar 2012 10:00:00.000000 +0000" " [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
" -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n" " -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n"
@@ -919,7 +914,7 @@ class TestCmdline(object):
self.match("3600\n") self.match("3600\n")
# See the missing chunks in list output # See the missing chunks in list output
self.ok("list --detail") self.ok("list -l --detail")
self.match("/newton/prep float32_8\n" + self.match("/newton/prep float32_8\n" +
" [ Fri, 23 Mar 2012 10:00:00.000000 +0000" " [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
" -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n" " -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n"
@@ -1043,7 +1038,7 @@ class TestCmdline(object):
else: else:
raise AssertionError("data not found at " + seek) raise AssertionError("data not found at " + seek)
# Verify "list" output # Verify "list" output
self.ok("list") self.ok("list -l")
self.match("/" + "/".join(components) + " float32_8\n") self.match("/" + "/".join(components) + " float32_8\n")
# Lots of renames # Lots of renames

View File

@@ -28,7 +28,10 @@ def setup_module():
recursive_unlink(testdb) recursive_unlink(testdb)
# Start web app on a custom port # Start web app on a custom port
test_db = nilmdb.utils.serializer_proxy(nilmdb.server.NilmDB)(testdb) test_db = nilmdb.utils.serializer_proxy(nilmdb.server.NilmDB)(
testdb, bulkdata_args = { "file_size" : 16384,
"files_per_dir" : 3 } )
test_server = nilmdb.server.Server(test_db, host = "127.0.0.1", test_server = nilmdb.server.Server(test_db, host = "127.0.0.1",
port = 32180, stoppable = False, port = 32180, stoppable = False,
fast_shutdown = True, fast_shutdown = True,
@@ -130,6 +133,15 @@ class TestNumpyClient(object):
[4, 5]]])) [4, 5]]]))
in_("wrong number of dimensions", str(e.exception)) in_("wrong number of dimensions", str(e.exception))
# Wrong number of fields
with assert_raises(ValueError) as e:
client.stream_insert_numpy("/test/1",
np.array([[0, 1, 2],
[1, 2, 3],
[3, 4, 5],
[4, 5, 6]]))
in_("wrong number of fields", str(e.exception))
# Unstructured # Unstructured
client.stream_create("/test/2", "float32_8") client.stream_create("/test/2", "float32_8")
client.stream_insert_numpy( client.stream_insert_numpy(
@@ -170,6 +182,17 @@ class TestNumpyClient(object):
assert(np.array_equal(a,b)) assert(np.array_equal(a,b))
assert(np.array_equal(a,c)) assert(np.array_equal(a,c))
# Make sure none of the files are greater than 16384 bytes as
# we configured with the bulkdata_args above.
datapath = os.path.join(testdb, "data")
for (dirpath, dirnames, filenames) in os.walk(datapath):
for f in filenames:
fn = os.path.join(dirpath, f)
size = os.path.getsize(fn)
if size > 16384:
raise AssertionError(sprintf("%s is too big: %d > %d\n",
fn, size, 16384))
nilmdb.client.numpyclient.StreamInserterNumpy._max_data = old_max_data nilmdb.client.numpyclient.StreamInserterNumpy._max_data = old_max_data
client.close() client.close()
@@ -286,8 +309,25 @@ class TestNumpyClient(object):
with client.stream_insert_numpy_context("/empty/test", end = 950): with client.stream_insert_numpy_context("/empty/test", end = 950):
pass pass
# Equal start and end is OK as long as there's no data
with assert_raises(ClientError) as e:
with client.stream_insert_numpy_context("/empty/test",
start=9, end=9) as ctx:
ctx.insert([[9, 9]])
ctx.finalize()
in_("have data to send, but invalid start/end times", str(e.exception))
with client.stream_insert_numpy_context("/empty/test",
start=9, end=9) as ctx:
pass
# reusing a context object is bad
with assert_raises(Exception) as e:
ctx.insert([[9, 9]])
# Try various things that might cause problems # Try various things that might cause problems
with client.stream_insert_numpy_context("/empty/test", 1000, 1050): with client.stream_insert_numpy_context("/empty/test",
1000, 1050) as ctx:
ctx.finalize() # inserts [1000, 1050] ctx.finalize() # inserts [1000, 1050]
ctx.finalize() # nothing ctx.finalize() # nothing
ctx.finalize() # nothing ctx.finalize() # nothing