From 862282ce99760832d3e9e5b4b1171b861105e004 Mon Sep 17 00:00:00 2001 From: Ebus-at-dockstar Date: Mon, 25 Mar 2013 10:24:28 +0100 Subject: move old stuff away --- heap/ebus/webhdf/__init__.py | 142 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 heap/ebus/webhdf/__init__.py (limited to 'heap/ebus/webhdf/__init__.py') diff --git a/heap/ebus/webhdf/__init__.py b/heap/ebus/webhdf/__init__.py new file mode 100644 index 0000000..44b5ab4 --- /dev/null +++ b/heap/ebus/webhdf/__init__.py @@ -0,0 +1,142 @@ +# -*- coding: utf-8 -*- +import os +import json +import datetime +import time +import itertools +import traceback +import logging +logging.basicConfig(level=logging.DEBUG) + +import numpy + +import bottle + +import ebus.datastore + +datastore = ebus.datastore.Datastore("hdf-data") +app = bottle.Bottle("ebus") + +lastflushes = dict() + +def get_lastflush(name): + if name not in lastflushes: + lastflushes[name] = -1 + + return lastflushes[name] + +@app.route('/') +def index_file(): + return static_files("index.html") + +@app.route('/static/:filename#.+#') +def static_files(filename): + return bottle.static_file(filename, root=os.path.join(os.path.dirname(__file__),"static")) + + +@app.get('/sensor/:name') +def sensor_data_get(name): + try: + table = datastore.getTable(name) + with datastore: + data = table.readSorted(sortby="timestamp", checkCSI=True, start=0, stop=1,step=-1).tolist()[0] + return {'sensor':name,'error':None,'data':data} + except Exception,e: + return {'sensor':name,'data':None, 'error':str(e)} + +@app.put('/sensor/:name') +@app.put('/sensor/:name/:timestamp') +def sensor_data_put(name,timestamp=None): + FLUSH_INTERVAL=240 #sec + global lastflush + + if not timestamp: timestamp = int(time.time()) + try: + value = bottle.request.POST.value + type = bottle.request.POST.type + + if type == "int": + klass = ebus.datastore.ValueInt + elif type == "float": + klass = ebus.datastore.ValueFloat + elif type == "string": + klass = ebus.datastore.ValueString + elif type == "": + klass = None + else: + return {'error':'INVALID_TYPE', msg:'Type {0} is invalid'.format(type)} + + msg = "Storing {0} of type {1} with timestamp {2} to {3}".format(value,type,timestamp,name) + logging.info(msg) + + datastore.addValue(name, timestamp, value, klass) + + if time.time() - FLUSH_INTERVAL > get_lastflush(name): + table = datastore.getTable(name) + with datastore.fileLock: + datastore.flush_table(name,table,close=False,index=True) + lastflushes[name] = time.time() + + return {'error':None,'msg':msg} + except Exception,e: + traceback.print_exc() + logging.error("Error: " + "{0} name={1} value={2} type={3}".format(e, name, value, type)) + return {'error':str(e),'msg':str(e)} + + + +@app.route('/sensor/:name/:startdate/:enddate') +def sensor_name_start_end(name,startdate,enddate): + SAMPLING_STEPSIZE=500 + NUMBER_OF_VALUES=250 + try: + startdate, enddate = int(startdate), int(enddate) + logging.info("/sensor/ start={0} end={1}".format(startdate, enddate)) + table=datastore.getTable(name) + with datastore: + i = table.where("(timestamp >= startdate) & (timestamp <= enddate)", + condvars={'startdate':startdate,'enddate':enddate,'timestamp':table.cols.timestamp}, + step=SAMPLING_STEPSIZE) + timestamps = [] + try: + for x in range(20): + i.next() + timestamps.append(i['timestamp']) + except: pass + + if len(timestamps) > 10: + diff = map(lambda (x1,x2): (x2-x1)/SAMPLING_STEPSIZE, zip(timestamps[:-1], timestamps[1:])) + diff_avg = numpy.average(diff) + time_period = enddate - startdate + samples = time_period / diff_avg + step = numpy.ceil(samples / float(NUMBER_OF_VALUES)) + + data = [(x['timestamp'], x['value']) + for x in table.where("(timestamp >= startdate) & (timestamp <= enddate)", step=step)] + logging.info("diff={0} samples={1} step={2} len={3} ({4})".format(diff_avg, samples, step, len(data),name)) + else: + # Not enough data, sampling approach useless, deliver all data + data = [(x['timestamp'], x['value']) for x in table.where("(timestamp >= startdate) & (timestamp <= enddate)")] + return {'sensor':name, 'error':None,'data':data} + except Exception,e: + logging.error("Error: " + str(e) + str(type(e))) + return {'sensor':name,'data':None, 'error':str(e)} + +@app.route('/avg/:name/:startdate') +@app.route('/avg/:name/:startdate/:period') +def sensor_avg_start(name, startdate, period=60*15): #15min + AVG_STEPSIZE=50 + try: + startdate, enddate = int(startdate), int(time.time()) + logging.info("/avg/ start={0} end={1}".format(startdate, enddate)) + table=datastore.getTable(name) + with datastore: + sel_rows = table.where("(timestamp >= startdate) & (timestamp <=enddate)", step=AVG_STEPSIZE) + f_group = range(startdate, enddate, period) + data = map(lambda (group_id, grouped_rows): (group_id, numpy.average([row['value'] for row in grouped_rows])), + itertools.groupby(sel_rows, lambda t: (t['timestamp']/period)*period)) + data = map(lambda (timestamp,value): (timestamp, value), data) + return {'sensor':name, 'error':None,'data':data} + except Exception,e: + return {'sensor':name, 'error':str(e), 'data':None} + -- cgit v1.2.1