Compare commits
17 Commits
nilmdb-1.7
...
nilmdb-1.8
Author | SHA1 | Date | |
---|---|---|---|
bd7bdb2eb8 | |||
840cd2fd13 | |||
bbd59c8b50 | |||
405c110fd7 | |||
274adcd856 | |||
a1850c9c2c | |||
6cd28b67b1 | |||
d6d215d53d | |||
e02143ddb2 | |||
e275384d03 | |||
a6a67ec15c | |||
fc43107307 | |||
90633413bb | |||
c7c3aff0fb | |||
e2347c954e | |||
222a5c6c53 | |||
1ca2c143e5 |
@@ -19,12 +19,12 @@ Then, set up Apache with a configuration like:
|
||||
|
||||
<VirtualHost>
|
||||
WSGIScriptAlias /nilmdb /home/nilm/nilmdb.wsgi
|
||||
WSGIApplicationGroup nilmdb-appgroup
|
||||
WSGIProcessGroup nilmdb-procgroup
|
||||
WSGIDaemonProcess nilmdb-procgroup threads=32 user=nilm group=nilm
|
||||
<Location /nilmdb>
|
||||
WSGIProcessGroup nilmdb-procgroup
|
||||
WSGIApplicationGroup nilmdb-appgroup
|
||||
|
||||
# Access control example:
|
||||
<Location /nilmdb>
|
||||
Order deny,allow
|
||||
Deny from all
|
||||
Allow from 1.2.3.4
|
||||
|
@@ -58,6 +58,11 @@ class Client(object):
|
||||
return self.http.get("dbinfo")
|
||||
|
||||
def stream_list(self, path = None, layout = None, extended = False):
|
||||
"""Return a sorted list of [path, layout] lists. If 'path' or
|
||||
'layout' are specified, only return streams that match those
|
||||
exact values. If 'extended' is True, the returned lists have
|
||||
extended info, e.g.: [path, layout, extent_min, extent_max,
|
||||
total_rows, total_seconds."""
|
||||
params = {}
|
||||
if path is not None:
|
||||
params["path"] = path
|
||||
@@ -69,6 +74,7 @@ class Client(object):
|
||||
return nilmdb.utils.sort.sort_human(streams, key = lambda s: s[0])
|
||||
|
||||
def stream_get_metadata(self, path, keys = None):
|
||||
"""Get stream metadata"""
|
||||
params = { "path": path }
|
||||
if keys is not None:
|
||||
params["key"] = keys
|
||||
|
@@ -1,5 +1,6 @@
|
||||
from nilmdb.utils.printf import *
|
||||
import nilmdb.utils.time
|
||||
from nilmdb.utils.interval import Interval
|
||||
|
||||
import fnmatch
|
||||
import argparse
|
||||
@@ -42,6 +43,8 @@ def setup(self, sub):
|
||||
group = cmd.add_argument_group("Misc options")
|
||||
group.add_argument("-T", "--timestamp-raw", action="store_true",
|
||||
help="Show raw timestamps when printing times")
|
||||
group.add_argument("-o", "--optimize", action="store_true",
|
||||
help="Optimize (merge adjacent) intervals")
|
||||
|
||||
return cmd
|
||||
|
||||
@@ -58,9 +61,16 @@ def cmd_intervals(self):
|
||||
time_string = nilmdb.utils.time.timestamp_to_human
|
||||
|
||||
try:
|
||||
for (start, end) in self.client.stream_intervals(
|
||||
self.args.path, self.args.start, self.args.end, self.args.diff):
|
||||
printf("[ %s -> %s ]\n", time_string(start), time_string(end))
|
||||
intervals = ( Interval(start, end) for (start, end) in
|
||||
self.client.stream_intervals(self.args.path,
|
||||
self.args.start,
|
||||
self.args.end,
|
||||
self.args.diff) )
|
||||
if self.args.optimize:
|
||||
intervals = nilmdb.utils.interval.optimize(intervals)
|
||||
for i in intervals:
|
||||
printf("[ %s -> %s ]\n", time_string(i.start), time_string(i.end))
|
||||
|
||||
except nilmdb.client.ClientError as e:
|
||||
self.die("error listing intervals: %s", str(e))
|
||||
|
||||
|
@@ -17,126 +17,26 @@ import decorator
|
||||
import psutil
|
||||
import traceback
|
||||
|
||||
from nilmdb.server.serverutil import (
|
||||
chunked_response,
|
||||
response_type,
|
||||
workaround_cp_bug_1200,
|
||||
exception_to_httperror,
|
||||
CORS_allow,
|
||||
json_to_request_params,
|
||||
json_error_page,
|
||||
cherrypy_start,
|
||||
cherrypy_stop,
|
||||
bool_param,
|
||||
)
|
||||
|
||||
# Add CORS_allow tool
|
||||
cherrypy.tools.CORS_allow = cherrypy.Tool('on_start_resource', CORS_allow)
|
||||
|
||||
class NilmApp(object):
|
||||
def __init__(self, db):
|
||||
self.db = db
|
||||
|
||||
# Decorators
|
||||
def chunked_response(func):
|
||||
"""Decorator to enable chunked responses."""
|
||||
# Set this to False to get better tracebacks from some requests
|
||||
# (/stream/extract, /stream/intervals).
|
||||
func._cp_config = { 'response.stream': True }
|
||||
return func
|
||||
|
||||
def response_type(content_type):
|
||||
"""Return a decorator-generating function that sets the
|
||||
response type to the specified string."""
|
||||
def wrapper(func, *args, **kwargs):
|
||||
cherrypy.response.headers['Content-Type'] = content_type
|
||||
return func(*args, **kwargs)
|
||||
return decorator.decorator(wrapper)
|
||||
|
||||
@decorator.decorator
|
||||
def workaround_cp_bug_1200(func, *args, **kwargs): # pragma: no cover
|
||||
"""Decorator to work around CherryPy bug #1200 in a response
|
||||
generator.
|
||||
|
||||
Even if chunked responses are disabled, LookupError or
|
||||
UnicodeError exceptions may still be swallowed by CherryPy due to
|
||||
bug #1200. This throws them as generic Exceptions instead so that
|
||||
they make it through.
|
||||
"""
|
||||
exc_info = None
|
||||
try:
|
||||
for val in func(*args, **kwargs):
|
||||
yield val
|
||||
except (LookupError, UnicodeError):
|
||||
# Re-raise it, but maintain the original traceback
|
||||
exc_info = sys.exc_info()
|
||||
new_exc = Exception(exc_info[0].__name__ + ": " + str(exc_info[1]))
|
||||
raise new_exc, None, exc_info[2]
|
||||
finally:
|
||||
del exc_info
|
||||
|
||||
def exception_to_httperror(*expected):
|
||||
"""Return a decorator-generating function that catches expected
|
||||
errors and throws a HTTPError describing it instead.
|
||||
|
||||
@exception_to_httperror(NilmDBError, ValueError)
|
||||
def foo():
|
||||
pass
|
||||
"""
|
||||
def wrapper(func, *args, **kwargs):
|
||||
exc_info = None
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except expected:
|
||||
# Re-raise it, but maintain the original traceback
|
||||
exc_info = sys.exc_info()
|
||||
new_exc = cherrypy.HTTPError("400 Bad Request", str(exc_info[1]))
|
||||
raise new_exc, None, exc_info[2]
|
||||
finally:
|
||||
del exc_info
|
||||
# We need to preserve the function's argspecs for CherryPy to
|
||||
# handle argument errors correctly. Decorator.decorator takes
|
||||
# care of that.
|
||||
return decorator.decorator(wrapper)
|
||||
|
||||
# Custom CherryPy tools
|
||||
|
||||
def CORS_allow(methods):
|
||||
"""This does several things:
|
||||
|
||||
Handles CORS preflight requests.
|
||||
Adds Allow: header to all requests.
|
||||
Raise 405 if request.method not in method.
|
||||
|
||||
It is similar to cherrypy.tools.allow, with the CORS stuff added.
|
||||
"""
|
||||
request = cherrypy.request.headers
|
||||
response = cherrypy.response.headers
|
||||
|
||||
if not isinstance(methods, (tuple, list)): # pragma: no cover
|
||||
methods = [ methods ]
|
||||
methods = [ m.upper() for m in methods if m ]
|
||||
if not methods: # pragma: no cover
|
||||
methods = [ 'GET', 'HEAD' ]
|
||||
elif 'GET' in methods and 'HEAD' not in methods: # pragma: no cover
|
||||
methods.append('HEAD')
|
||||
response['Allow'] = ', '.join(methods)
|
||||
|
||||
# Allow all origins
|
||||
if 'Origin' in request:
|
||||
response['Access-Control-Allow-Origin'] = request['Origin']
|
||||
|
||||
# If it's a CORS request, send response.
|
||||
request_method = request.get("Access-Control-Request-Method", None)
|
||||
request_headers = request.get("Access-Control-Request-Headers", None)
|
||||
if (cherrypy.request.method == "OPTIONS" and
|
||||
request_method and request_headers):
|
||||
response['Access-Control-Allow-Headers'] = request_headers
|
||||
response['Access-Control-Allow-Methods'] = ', '.join(methods)
|
||||
# Try to stop further processing and return a 200 OK
|
||||
cherrypy.response.status = "200 OK"
|
||||
cherrypy.response.body = ""
|
||||
cherrypy.request.handler = lambda: ""
|
||||
return
|
||||
|
||||
# Reject methods that were not explicitly allowed
|
||||
if cherrypy.request.method not in methods:
|
||||
raise cherrypy.HTTPError(405)
|
||||
|
||||
cherrypy.tools.CORS_allow = cherrypy.Tool('on_start_resource', CORS_allow)
|
||||
|
||||
# Helper for json_in tool to process JSON data into normal request
|
||||
# parameters.
|
||||
def json_to_request_params(body):
|
||||
cherrypy.lib.jsontools.json_processor(body)
|
||||
if not isinstance(cherrypy.request.json, dict):
|
||||
raise cherrypy.HTTPError(415)
|
||||
cherrypy.request.params.update(cherrypy.request.json)
|
||||
|
||||
# CherryPy apps
|
||||
class Root(NilmApp):
|
||||
"""Root application for NILM database"""
|
||||
@@ -206,10 +106,10 @@ class Stream(NilmApp):
|
||||
layout parameter, just list streams that match the given path
|
||||
or layout.
|
||||
|
||||
If extent is not given, returns a list of lists containing
|
||||
the path and layout: [ path, layout ]
|
||||
If extended is missing or zero, returns a list of lists
|
||||
containing the path and layout: [ path, layout ]
|
||||
|
||||
If extended is provided, returns a list of lists containing
|
||||
If extended is true, returns a list of lists containing
|
||||
extended info: [ path, layout, extent_min, extent_max,
|
||||
total_rows, total_seconds ]. More data may be added.
|
||||
"""
|
||||
@@ -322,6 +222,8 @@ class Stream(NilmApp):
|
||||
little-endian and matches the database types (including an
|
||||
int64 timestamp).
|
||||
"""
|
||||
binary = bool_param(binary)
|
||||
|
||||
# Important that we always read the input before throwing any
|
||||
# errors, to keep lengths happy for persistent connections.
|
||||
# Note that CherryPy 3.2.2 has a bug where this fails for GET
|
||||
@@ -446,6 +348,10 @@ class Stream(NilmApp):
|
||||
little-endian and matches the database types (including an
|
||||
int64 timestamp).
|
||||
"""
|
||||
binary = bool_param(binary)
|
||||
markup = bool_param(markup)
|
||||
count = bool_param(count)
|
||||
|
||||
(start, end) = self._get_times(start, end)
|
||||
|
||||
# Check path and get layout
|
||||
@@ -573,70 +479,14 @@ class Server(object):
|
||||
|
||||
def json_error_page(self, status, message, traceback, version):
|
||||
"""Return a custom error page in JSON so the client can parse it"""
|
||||
errordata = { "status" : status,
|
||||
"message" : message,
|
||||
"traceback" : traceback }
|
||||
# Don't send a traceback if the error was 400-499 (client's fault)
|
||||
try:
|
||||
code = int(status.split()[0])
|
||||
if not self.force_traceback:
|
||||
if code >= 400 and code <= 499:
|
||||
errordata["traceback"] = ""
|
||||
except Exception: # pragma: no cover
|
||||
pass
|
||||
# Override the response type, which was previously set to text/html
|
||||
cherrypy.serving.response.headers['Content-Type'] = (
|
||||
"application/json;charset=utf-8" )
|
||||
# Undo the HTML escaping that cherrypy's get_error_page function applies
|
||||
# (cherrypy issue 1135)
|
||||
for k, v in errordata.iteritems():
|
||||
v = v.replace("<","<")
|
||||
v = v.replace(">",">")
|
||||
v = v.replace("&","&")
|
||||
errordata[k] = v
|
||||
return json.dumps(errordata, separators=(',',':'))
|
||||
return json_error_page(status, message, traceback, version,
|
||||
self.force_traceback)
|
||||
|
||||
def start(self, blocking = False, event = None):
|
||||
|
||||
if not self.embedded: # pragma: no cover
|
||||
# Handle signals nicely
|
||||
if hasattr(cherrypy.engine, "signal_handler"):
|
||||
cherrypy.engine.signal_handler.subscribe()
|
||||
if hasattr(cherrypy.engine, "console_control_handler"):
|
||||
cherrypy.engine.console_control_handler.subscribe()
|
||||
|
||||
# Cherrypy stupidly calls os._exit(70) when it can't bind the
|
||||
# port. At least try to print a reasonable error and continue
|
||||
# in this case, rather than just dying silently (as we would
|
||||
# otherwise do in embedded mode)
|
||||
real_exit = os._exit
|
||||
def fake_exit(code): # pragma: no cover
|
||||
if code == os.EX_SOFTWARE:
|
||||
fprintf(sys.stderr, "error: CherryPy called os._exit!\n")
|
||||
else:
|
||||
real_exit(code)
|
||||
os._exit = fake_exit
|
||||
cherrypy.engine.start()
|
||||
os._exit = real_exit
|
||||
|
||||
# Signal that the engine has started successfully
|
||||
if event is not None:
|
||||
event.set()
|
||||
|
||||
if blocking:
|
||||
try:
|
||||
cherrypy.engine.wait(cherrypy.engine.states.EXITING,
|
||||
interval = 0.1, channel = 'main')
|
||||
except (KeyboardInterrupt, IOError): # pragma: no cover
|
||||
cherrypy.engine.log('Keyboard Interrupt: shutting down bus')
|
||||
cherrypy.engine.exit()
|
||||
except SystemExit: # pragma: no cover
|
||||
cherrypy.engine.log('SystemExit raised: shutting down bus')
|
||||
cherrypy.engine.exit()
|
||||
raise
|
||||
cherrypy_start(blocking, event, self.embedded)
|
||||
|
||||
def stop(self):
|
||||
cherrypy.engine.exit()
|
||||
cherrypy_stop()
|
||||
|
||||
# Use a single global nilmdb.server.NilmDB and nilmdb.server.Server
|
||||
# instance since the database can only be opened once. For this to
|
||||
|
214
nilmdb/server/serverutil.py
Normal file
214
nilmdb/server/serverutil.py
Normal file
@@ -0,0 +1,214 @@
|
||||
"""Miscellaneous decorators and other helpers for running a CherryPy
|
||||
server"""
|
||||
|
||||
import cherrypy
|
||||
import sys
|
||||
import os
|
||||
import decorator
|
||||
import simplejson as json
|
||||
|
||||
# Helper to parse parameters into booleans
|
||||
def bool_param(s):
|
||||
"""Return a bool indicating whether parameter 's' was True or False,
|
||||
supporting a few different types for 's'."""
|
||||
try:
|
||||
ss = s.lower()
|
||||
if ss in [ "0", "false", "f", "no", "n" ]:
|
||||
return False
|
||||
if ss in [ "1", "true", "t", "yes", "y" ]:
|
||||
return True
|
||||
except Exception:
|
||||
return bool(s)
|
||||
raise cherrypy.HTTPError("400 Bad Request",
|
||||
"can't parse parameter: " + ss)
|
||||
|
||||
# Decorators
|
||||
def chunked_response(func):
|
||||
"""Decorator to enable chunked responses."""
|
||||
# Set this to False to get better tracebacks from some requests
|
||||
# (/stream/extract, /stream/intervals).
|
||||
func._cp_config = { 'response.stream': True }
|
||||
return func
|
||||
|
||||
def response_type(content_type):
|
||||
"""Return a decorator-generating function that sets the
|
||||
response type to the specified string."""
|
||||
def wrapper(func, *args, **kwargs):
|
||||
cherrypy.response.headers['Content-Type'] = content_type
|
||||
return func(*args, **kwargs)
|
||||
return decorator.decorator(wrapper)
|
||||
|
||||
@decorator.decorator
|
||||
def workaround_cp_bug_1200(func, *args, **kwargs): # pragma: no cover
|
||||
"""Decorator to work around CherryPy bug #1200 in a response
|
||||
generator.
|
||||
|
||||
Even if chunked responses are disabled, LookupError or
|
||||
UnicodeError exceptions may still be swallowed by CherryPy due to
|
||||
bug #1200. This throws them as generic Exceptions instead so that
|
||||
they make it through.
|
||||
"""
|
||||
exc_info = None
|
||||
try:
|
||||
for val in func(*args, **kwargs):
|
||||
yield val
|
||||
except (LookupError, UnicodeError):
|
||||
# Re-raise it, but maintain the original traceback
|
||||
exc_info = sys.exc_info()
|
||||
new_exc = Exception(exc_info[0].__name__ + ": " + str(exc_info[1]))
|
||||
raise new_exc, None, exc_info[2]
|
||||
finally:
|
||||
del exc_info
|
||||
|
||||
def exception_to_httperror(*expected):
|
||||
"""Return a decorator-generating function that catches expected
|
||||
errors and throws a HTTPError describing it instead.
|
||||
|
||||
@exception_to_httperror(NilmDBError, ValueError)
|
||||
def foo():
|
||||
pass
|
||||
"""
|
||||
def wrapper(func, *args, **kwargs):
|
||||
exc_info = None
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except expected:
|
||||
# Re-raise it, but maintain the original traceback
|
||||
exc_info = sys.exc_info()
|
||||
new_exc = cherrypy.HTTPError("400 Bad Request", str(exc_info[1]))
|
||||
raise new_exc, None, exc_info[2]
|
||||
finally:
|
||||
del exc_info
|
||||
# We need to preserve the function's argspecs for CherryPy to
|
||||
# handle argument errors correctly. Decorator.decorator takes
|
||||
# care of that.
|
||||
return decorator.decorator(wrapper)
|
||||
|
||||
# Custom CherryPy tools
|
||||
|
||||
def CORS_allow(methods):
|
||||
"""This does several things:
|
||||
|
||||
Handles CORS preflight requests.
|
||||
Adds Allow: header to all requests.
|
||||
Raise 405 if request.method not in method.
|
||||
|
||||
It is similar to cherrypy.tools.allow, with the CORS stuff added.
|
||||
|
||||
Add this to CherryPy with:
|
||||
cherrypy.tools.CORS_allow = cherrypy.Tool('on_start_resource', CORS_allow)
|
||||
"""
|
||||
request = cherrypy.request.headers
|
||||
response = cherrypy.response.headers
|
||||
|
||||
if not isinstance(methods, (tuple, list)): # pragma: no cover
|
||||
methods = [ methods ]
|
||||
methods = [ m.upper() for m in methods if m ]
|
||||
if not methods: # pragma: no cover
|
||||
methods = [ 'GET', 'HEAD' ]
|
||||
elif 'GET' in methods and 'HEAD' not in methods: # pragma: no cover
|
||||
methods.append('HEAD')
|
||||
response['Allow'] = ', '.join(methods)
|
||||
|
||||
# Allow all origins
|
||||
if 'Origin' in request:
|
||||
response['Access-Control-Allow-Origin'] = request['Origin']
|
||||
|
||||
# If it's a CORS request, send response.
|
||||
request_method = request.get("Access-Control-Request-Method", None)
|
||||
request_headers = request.get("Access-Control-Request-Headers", None)
|
||||
if (cherrypy.request.method == "OPTIONS" and
|
||||
request_method and request_headers):
|
||||
response['Access-Control-Allow-Headers'] = request_headers
|
||||
response['Access-Control-Allow-Methods'] = ', '.join(methods)
|
||||
# Try to stop further processing and return a 200 OK
|
||||
cherrypy.response.status = "200 OK"
|
||||
cherrypy.response.body = ""
|
||||
cherrypy.request.handler = lambda: ""
|
||||
return
|
||||
|
||||
# Reject methods that were not explicitly allowed
|
||||
if cherrypy.request.method not in methods:
|
||||
raise cherrypy.HTTPError(405)
|
||||
|
||||
|
||||
# Helper for json_in tool to process JSON data into normal request
|
||||
# parameters.
|
||||
def json_to_request_params(body):
|
||||
cherrypy.lib.jsontools.json_processor(body)
|
||||
if not isinstance(cherrypy.request.json, dict):
|
||||
raise cherrypy.HTTPError(415)
|
||||
cherrypy.request.params.update(cherrypy.request.json)
|
||||
|
||||
# Used as an "error_page.default" handler
|
||||
def json_error_page(status, message, traceback, version,
|
||||
force_traceback = False):
|
||||
"""Return a custom error page in JSON so the client can parse it"""
|
||||
errordata = { "status" : status,
|
||||
"message" : message,
|
||||
"traceback" : traceback }
|
||||
# Don't send a traceback if the error was 400-499 (client's fault)
|
||||
try:
|
||||
code = int(status.split()[0])
|
||||
if not force_traceback:
|
||||
if code >= 400 and code <= 499:
|
||||
errordata["traceback"] = ""
|
||||
except Exception: # pragma: no cover
|
||||
pass
|
||||
# Override the response type, which was previously set to text/html
|
||||
cherrypy.serving.response.headers['Content-Type'] = (
|
||||
"application/json;charset=utf-8" )
|
||||
# Undo the HTML escaping that cherrypy's get_error_page function applies
|
||||
# (cherrypy issue 1135)
|
||||
for k, v in errordata.iteritems():
|
||||
v = v.replace("<","<")
|
||||
v = v.replace(">",">")
|
||||
v = v.replace("&","&")
|
||||
errordata[k] = v
|
||||
return json.dumps(errordata, separators=(',',':'))
|
||||
|
||||
# Start/stop CherryPy standalone server
|
||||
def cherrypy_start(blocking = False, event = False, embedded = False):
|
||||
"""Start the CherryPy server, handling errors and signals
|
||||
somewhat gracefully."""
|
||||
|
||||
if not embedded: # pragma: no cover
|
||||
# Handle signals nicely
|
||||
if hasattr(cherrypy.engine, "signal_handler"):
|
||||
cherrypy.engine.signal_handler.subscribe()
|
||||
if hasattr(cherrypy.engine, "console_control_handler"):
|
||||
cherrypy.engine.console_control_handler.subscribe()
|
||||
|
||||
# Cherrypy stupidly calls os._exit(70) when it can't bind the
|
||||
# port. At least try to print a reasonable error and continue
|
||||
# in this case, rather than just dying silently (as we would
|
||||
# otherwise do in embedded mode)
|
||||
real_exit = os._exit
|
||||
def fake_exit(code): # pragma: no cover
|
||||
if code == os.EX_SOFTWARE:
|
||||
fprintf(sys.stderr, "error: CherryPy called os._exit!\n")
|
||||
else:
|
||||
real_exit(code)
|
||||
os._exit = fake_exit
|
||||
cherrypy.engine.start()
|
||||
os._exit = real_exit
|
||||
|
||||
# Signal that the engine has started successfully
|
||||
if event is not None:
|
||||
event.set()
|
||||
|
||||
if blocking:
|
||||
try:
|
||||
cherrypy.engine.wait(cherrypy.engine.states.EXITING,
|
||||
interval = 0.1, channel = 'main')
|
||||
except (KeyboardInterrupt, IOError): # pragma: no cover
|
||||
cherrypy.engine.log('Keyboard Interrupt: shutting down bus')
|
||||
cherrypy.engine.exit()
|
||||
except SystemExit: # pragma: no cover
|
||||
cherrypy.engine.log('SystemExit raised: shutting down bus')
|
||||
cherrypy.engine.exit()
|
||||
raise
|
||||
|
||||
# Stop CherryPy server
|
||||
def cherrypy_stop():
|
||||
cherrypy.engine.exit()
|
@@ -1,5 +1,6 @@
|
||||
"""Interval. Like nilmdb.server.interval, but re-implemented here
|
||||
in plain Python so clients have easier access to it.
|
||||
in plain Python so clients have easier access to it, and with a few
|
||||
helper functions.
|
||||
|
||||
Intervals are half-open, ie. they include data points with timestamps
|
||||
[start, end)
|
||||
@@ -34,6 +35,10 @@ class Interval:
|
||||
return ("[" + nilmdb.utils.time.timestamp_to_string(self.start) +
|
||||
" -> " + nilmdb.utils.time.timestamp_to_string(self.end) + ")")
|
||||
|
||||
def human_string(self):
|
||||
return ("[ " + nilmdb.utils.time.timestamp_to_human(self.start) +
|
||||
" -> " + nilmdb.utils.time.timestamp_to_human(self.end) + " ]")
|
||||
|
||||
def __cmp__(self, other):
|
||||
"""Compare two intervals. If non-equal, order by start then end"""
|
||||
return cmp(self.start, other.start) or cmp(self.end, other.end)
|
||||
@@ -53,18 +58,11 @@ class Interval:
|
||||
raise IntervalError("not a subset")
|
||||
return Interval(start, end)
|
||||
|
||||
def set_difference(a, b):
|
||||
"""
|
||||
Compute the difference (a \\ b) between the intervals in 'a' and
|
||||
the intervals in 'b'; i.e., the ranges that are present in 'self'
|
||||
but not 'other'.
|
||||
|
||||
'a' and 'b' must both be iterables.
|
||||
|
||||
Returns a generator that yields each interval in turn.
|
||||
Output intervals are built as subsets of the intervals in the
|
||||
first argument (a).
|
||||
"""
|
||||
def _interval_math_helper(a, b, op, subset = True):
|
||||
"""Helper for set_difference, intersection functions,
|
||||
to compute interval subsets based on a math operator on ranges
|
||||
present in A and B. Subsets are computed from A, or new intervals
|
||||
are generated if subset = False."""
|
||||
# Iterate through all starts and ends in sorted order. Add a
|
||||
# tag to the iterator so that we can figure out which one they
|
||||
# were, after sorting.
|
||||
@@ -79,28 +77,71 @@ def set_difference(a, b):
|
||||
# At each point, evaluate which type of end it is, to determine
|
||||
# how to build up the output intervals.
|
||||
a_interval = None
|
||||
b_interval = None
|
||||
in_a = False
|
||||
in_b = False
|
||||
out_start = None
|
||||
for (ts, k, i) in nilmdb.utils.iterator.imerge(a_iter, b_iter):
|
||||
if k == 0:
|
||||
# start a interval
|
||||
a_interval = i
|
||||
if b_interval is None:
|
||||
out_start = ts
|
||||
in_a = True
|
||||
elif k == 1:
|
||||
# start b interval
|
||||
b_interval = i
|
||||
if out_start is not None and out_start != ts:
|
||||
yield a_interval.subset(out_start, ts)
|
||||
out_start = None
|
||||
in_b = True
|
||||
elif k == 2:
|
||||
# end a interval
|
||||
if out_start is not None and out_start != ts:
|
||||
yield a_interval.subset(out_start, ts)
|
||||
out_start = None
|
||||
a_interval = None
|
||||
in_a = False
|
||||
elif k == 3:
|
||||
# end b interval
|
||||
b_interval = None
|
||||
if a_interval:
|
||||
in_b = False
|
||||
include = op(in_a, in_b)
|
||||
if include and out_start is None:
|
||||
out_start = ts
|
||||
elif not include:
|
||||
if out_start is not None and out_start != ts:
|
||||
if subset:
|
||||
yield a_interval.subset(out_start, ts)
|
||||
else:
|
||||
yield Interval(out_start, ts)
|
||||
out_start = None
|
||||
|
||||
def set_difference(a, b):
|
||||
"""
|
||||
Compute the difference (a \\ b) between the intervals in 'a' and
|
||||
the intervals in 'b'; i.e., the ranges that are present in 'self'
|
||||
but not 'other'.
|
||||
|
||||
'a' and 'b' must both be iterables.
|
||||
|
||||
Returns a generator that yields each interval in turn.
|
||||
Output intervals are built as subsets of the intervals in the
|
||||
first argument (a).
|
||||
"""
|
||||
return _interval_math_helper(a, b, (lambda a, b: a and not b))
|
||||
|
||||
def intersection(a, b):
|
||||
"""
|
||||
Compute the intersection between the intervals in 'a' and the
|
||||
intervals in 'b'; i.e., the ranges that are present in both 'a'
|
||||
and 'b'.
|
||||
|
||||
'a' and 'b' must both be iterables.
|
||||
|
||||
Returns a generator that yields each interval in turn.
|
||||
Output intervals are built as subsets of the intervals in the
|
||||
first argument (a).
|
||||
"""
|
||||
return _interval_math_helper(a, b, (lambda a, b: a and b))
|
||||
|
||||
def optimize(it):
|
||||
"""
|
||||
Given an iterable 'it' with intervals, optimize them by joining
|
||||
together intervals that are adjacent in time, and return a generator
|
||||
that yields the new intervals.
|
||||
"""
|
||||
saved_int = None
|
||||
for interval in it:
|
||||
if saved_int is not None:
|
||||
if saved_int.end == interval.start:
|
||||
interval.start = saved_int.start
|
||||
else:
|
||||
yield saved_int
|
||||
saved_int = interval
|
||||
if saved_int is not None:
|
||||
yield saved_int
|
||||
|
@@ -91,6 +91,20 @@ def serializer_proxy(obj_or_type):
|
||||
r = SerializerCallProxy(self.__call_queue, attr, self)
|
||||
return r
|
||||
|
||||
# For an interable object, on __iter__(), save the object's
|
||||
# iterator and return this proxy. On next(), call the object's
|
||||
# iterator through this proxy.
|
||||
def __iter__(self):
|
||||
attr = getattr(self.__object, "__iter__")
|
||||
self.__iter = SerializerCallProxy(self.__call_queue, attr, self)()
|
||||
return self
|
||||
def next(self):
|
||||
return SerializerCallProxy(self.__call_queue,
|
||||
self.__iter.next, self)()
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.__getattr__("__getitem__")(key)
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
"""Call this to instantiate the type, if a type was passed
|
||||
to serializer_proxy. Otherwise, pass the call through."""
|
||||
|
@@ -60,7 +60,7 @@ def rate_to_period(hz, cycles = 1):
|
||||
def parse_time(toparse):
|
||||
"""
|
||||
Parse a free-form time string and return a nilmdb timestamp
|
||||
(integer seconds since epoch). If the string doesn't contain a
|
||||
(integer microseconds since epoch). If the string doesn't contain a
|
||||
timestamp, the current local timezone is assumed (e.g. from the TZ
|
||||
env var).
|
||||
"""
|
||||
|
@@ -1,7 +1,14 @@
|
||||
import sys
|
||||
|
||||
if sys.version_info[0] >= 3: # pragma: no cover (future Python3 compat)
|
||||
text_type = str
|
||||
else:
|
||||
text_type = unicode
|
||||
|
||||
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):
|
||||
if not isinstance(u, text_type):
|
||||
return u
|
||||
try:
|
||||
return u.encode()
|
||||
@@ -11,7 +18,7 @@ def encode(u):
|
||||
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):
|
||||
if isinstance(s, text_type):
|
||||
return s
|
||||
try:
|
||||
return s.decode()
|
||||
|
@@ -354,10 +354,6 @@ class TestClient(object):
|
||||
with assert_raises(ServerError) as e:
|
||||
client.http.get_gen("http://nosuchurl.example.com./").next()
|
||||
|
||||
# Trigger a curl error in generator
|
||||
with assert_raises(ServerError) as e:
|
||||
client.http.get_gen("http://nosuchurl.example.com./").next()
|
||||
|
||||
# Check 404 for missing streams
|
||||
for function in [ client.stream_intervals, client.stream_extract ]:
|
||||
with assert_raises(ClientError) as e:
|
||||
@@ -396,20 +392,16 @@ class TestClient(object):
|
||||
headers())
|
||||
|
||||
# Extract
|
||||
x = http.get("stream/extract",
|
||||
{ "path": "/newton/prep",
|
||||
"start": "123",
|
||||
"end": "124" })
|
||||
x = http.get("stream/extract", { "path": "/newton/prep",
|
||||
"start": "123", "end": "124" })
|
||||
if "transfer-encoding: chunked" not in headers():
|
||||
warnings.warn("Non-chunked HTTP response for /stream/extract")
|
||||
if "content-type: text/plain;charset=utf-8" not in headers():
|
||||
raise AssertionError("/stream/extract is not text/plain:\n" +
|
||||
headers())
|
||||
|
||||
x = http.get("stream/extract",
|
||||
{ "path": "/newton/prep",
|
||||
"start": "123",
|
||||
"end": "124",
|
||||
x = http.get("stream/extract", { "path": "/newton/prep",
|
||||
"start": "123", "end": "124",
|
||||
"binary": "1" })
|
||||
if "transfer-encoding: chunked" not in headers():
|
||||
warnings.warn("Non-chunked HTTP response for /stream/extract")
|
||||
@@ -417,6 +409,21 @@ class TestClient(object):
|
||||
raise AssertionError("/stream/extract is not binary:\n" +
|
||||
headers())
|
||||
|
||||
# Make sure a binary of "0" is really off
|
||||
x = http.get("stream/extract", { "path": "/newton/prep",
|
||||
"start": "123", "end": "124",
|
||||
"binary": "0" })
|
||||
if "content-type: application/octet-stream" in headers():
|
||||
raise AssertionError("/stream/extract is not text:\n" +
|
||||
headers())
|
||||
|
||||
# Invalid parameters
|
||||
with assert_raises(ClientError) as e:
|
||||
x = http.get("stream/extract", { "path": "/newton/prep",
|
||||
"start": "123", "end": "124",
|
||||
"binary": "asdfasfd" })
|
||||
in_("can't parse parameter", str(e.exception))
|
||||
|
||||
client.close()
|
||||
|
||||
def test_client_08_unicode(self):
|
||||
|
@@ -1011,6 +1011,18 @@ class TestCmdline(object):
|
||||
self.match("[ Thu, 01 Jan 2004 00:00:00.000000 +0000 -"
|
||||
"> Sat, 01 Jan 2005 00:00:00.000000 +0000 ]\n")
|
||||
|
||||
# optimize
|
||||
self.ok("insert -s 01-01-2002 -e 01-01-2004 /diff/1 /dev/null")
|
||||
self.ok("intervals /diff/1")
|
||||
self.match("[ Sat, 01 Jan 2000 00:00:00.000000 +0000 -"
|
||||
"> Thu, 01 Jan 2004 00:00:00.000000 +0000 ]\n"
|
||||
"[ Thu, 01 Jan 2004 00:00:00.000000 +0000 -"
|
||||
"> Sat, 01 Jan 2005 00:00:00.000000 +0000 ]\n")
|
||||
self.ok("intervals /diff/1 --optimize")
|
||||
self.ok("intervals /diff/1 -o")
|
||||
self.match("[ Sat, 01 Jan 2000 00:00:00.000000 +0000 -"
|
||||
"> Sat, 01 Jan 2005 00:00:00.000000 +0000 ]\n")
|
||||
|
||||
self.ok("destroy -R /diff/1")
|
||||
self.ok("destroy -R /diff/2")
|
||||
|
||||
|
@@ -59,6 +59,14 @@ class TestInterval:
|
||||
self.test_interval_intersect()
|
||||
Interval = NilmdbInterval
|
||||
|
||||
# Other helpers in nilmdb.utils.interval
|
||||
i = [ UtilsInterval(1,2), UtilsInterval(2,3), UtilsInterval(4,5) ]
|
||||
eq_(list(nilmdb.utils.interval.optimize(i)),
|
||||
[ UtilsInterval(1,3), UtilsInterval(4,5) ])
|
||||
eq_(UtilsInterval(1234567890123456, 1234567890654321).human_string(),
|
||||
"[ Fri, 13 Feb 2009 18:31:30.123456 -0500 -> " +
|
||||
"Fri, 13 Feb 2009 18:31:30.654321 -0500 ]")
|
||||
|
||||
def test_interval(self):
|
||||
# Test Interval class
|
||||
os.environ['TZ'] = "America/New_York"
|
||||
@@ -226,13 +234,16 @@ class TestInterval:
|
||||
x = makeset("[--)") & 1234
|
||||
|
||||
def do_test(a, b, c, d):
|
||||
# a & b == c
|
||||
# a & b == c (using nilmdb.server.interval)
|
||||
ab = IntervalSet()
|
||||
for x in b:
|
||||
for i in (a & x):
|
||||
ab += i
|
||||
eq_(ab,c)
|
||||
|
||||
# a & b == c (using nilmdb.utils.interval)
|
||||
eq_(IntervalSet(nilmdb.utils.interval.intersection(a,b)), c)
|
||||
|
||||
# a \ b == d
|
||||
eq_(IntervalSet(nilmdb.utils.interval.set_difference(a,b)), d)
|
||||
|
||||
@@ -302,6 +313,17 @@ class TestInterval:
|
||||
eq_(nilmdb.utils.interval.set_difference(
|
||||
a.intersection(list(c)[0]), b.intersection(list(c)[0])), d)
|
||||
|
||||
# Fill out test coverage for non-subsets
|
||||
def diff2(a,b, subset):
|
||||
return nilmdb.utils.interval._interval_math_helper(
|
||||
a, b, (lambda a, b: b and not a), subset=subset)
|
||||
with assert_raises(nilmdb.utils.interval.IntervalError):
|
||||
list(diff2(a,b,True))
|
||||
list(diff2(a,b,False))
|
||||
|
||||
# Empty second set
|
||||
eq_(nilmdb.utils.interval.set_difference(a, IntervalSet()), a)
|
||||
|
||||
# Empty second set
|
||||
eq_(nilmdb.utils.interval.set_difference(a, IntervalSet()), a)
|
||||
|
||||
|
@@ -62,6 +62,28 @@ class Base(object):
|
||||
eq_(self.foo.val, 20)
|
||||
eq_(self.foo.init_thread, self.foo.test_thread)
|
||||
|
||||
class ListLike(object):
|
||||
def __init__(self):
|
||||
self.thread = threading.current_thread().name
|
||||
self.foo = 0
|
||||
|
||||
def __iter__(self):
|
||||
eq_(threading.current_thread().name, self.thread)
|
||||
self.foo = 0
|
||||
return self
|
||||
|
||||
def __getitem__(self, key):
|
||||
eq_(threading.current_thread().name, self.thread)
|
||||
return key
|
||||
|
||||
def next(self):
|
||||
eq_(threading.current_thread().name, self.thread)
|
||||
if self.foo < 5:
|
||||
self.foo += 1
|
||||
return self.foo
|
||||
else:
|
||||
raise StopIteration
|
||||
|
||||
class TestUnserialized(Base):
|
||||
def setUp(self):
|
||||
self.foo = Foo()
|
||||
@@ -84,3 +106,9 @@ class TestSerializer(Base):
|
||||
sp(sp(Foo("x"))).t()
|
||||
sp(sp(Foo)("x")).t()
|
||||
sp(sp(Foo))("x").t()
|
||||
|
||||
def test_iter(self):
|
||||
sp = nilmdb.utils.serializer_proxy
|
||||
i = sp(ListLike)()
|
||||
eq_(list(i), [1,2,3,4,5])
|
||||
eq_(i[3], 3)
|
||||
|
Reference in New Issue
Block a user