homecontrol: use sqlite with dataset

.gitignore

@@ -1,3 +1,7 @@
 Pipfile.lock
 config.json
 .idea/*
+db.sqlite
+__pycache__
+*.swp
+db.sqlite

Client.py

@@ -20,8 +20,15 @@ class Sensor(Client):
         self.values = {}
         super().__init__(id_s, type_s)
 
-    def get_values(self):
-        return self.values
+    def get_values(self, single=False):
+        ret = {}
+        ret["id"] = self.id_c
+        ret["type"] = self.type_c
+        if single:
+            ret["values"] = self.values[next(iter(self.values))]
+        else:
+            ret["values"] = self.values
+        return ret
 
     def add_value(self, value):
         ts = time.time()
@@ -34,7 +41,6 @@ class Sensor(Client):
 
 class Actor(Client):
     def __init__(self, id_a, type_a, levels):
-        self.id_a = id_a
         self.type_a = type_a
         self.levels = levels
         self.queue = queue.Queue()
@@ -42,11 +48,11 @@ class Actor(Client):
 
 
     def get_info(self):
-        return {"id": self.id_a, "type": self.type_a, "ts": self.ts,
+        return {"id": self.id_c, "type": self.type_a, "ts": self.ts,
                 "levels": self.levels, "level": self.level}
 
     def add_queue(self, command, data):
-        print("add_queue: {\"command\": %s, \"data\": %s}" % (command, data))
+        # print("add_queue: {\"command\": %s, \"data\": %s}" % (command, data))
         self.queue.put({"command": command, "data": data})
 
     def get_queue(self):

Pipfile

@@ -8,3 +8,4 @@ name = "pypi"
 [packages]
 requests = {version="*"}
 flask = {version="*"}
+dataset = {version="*"}

homecontrol.py

@@ -3,11 +3,14 @@ from flask import Flask, request, json, jsonify, abort, make_response
 import logging
 import argparse
 import time
-import Client
+import dataset
+import sys
 
 CONFIG_FILE = 'config.json'
+DB_FILE = 'db.sqlite'
+LIMIT=100
 PORT = 5000
-TIMEOUT_CLIENT = 10
+# TIMEOUT_CLIENT = 10
 core = None
 
 # sensors
@@ -20,6 +23,8 @@ def setup():
                         help='listening port')
     parser.add_argument('-c', '--config', dest='config', type=str,
                         help='config file', default=CONFIG_FILE)
+    parser.add_argument('-f', '--dbfile', dest='dbfile', type=str,
+                        help='database file', default=DB_FILE)
     parser.add_argument('-d', '--debug', dest='debug', action='store_true',
                         help='debug mode')
     parser.add_argument('-D', '--debugflask', dest='debugflask', action='store_true',
@@ -33,16 +38,21 @@ def setup():
     try:
         config_file = open(args.config, 'r')
         config = json.load(config_file)
-    except Exception as ex:
-        logger.error('Exception Type:%s, args:\n%s' % (type(ex).__name__, ex.args))
+    except:
+        pass
 
     # fill new keys with defaults
     if not config.get('port'):
         config['port'] = PORT
 
+    if not config.get('dbfile'):
+        config['dbfile'] = DB_FILE
+
     # overwrite with arguments
     if args.port:
         config['port'] = args.port
+    if args.dbfile:
+        config['dbfile'] = args.dbfile
 
     # save to file
     with open(args.config, 'w') as config_file:
@@ -50,92 +60,141 @@ def setup():
 
     return args, config
 
+class Logger:
+    def __init__(self, name, debug):
+        FORMAT='%(asctime)-8s %(name)12s::%(levelname)-7s %(message)s'
+        DATEFMT='%H:%M:%S'
+        logging.basicConfig(level=logging.INFO, format=FORMAT, datefmt=DATEFMT)
+        self.logger = logging.getLogger(name)
+        if debug:
+            self.logger.setLevel(logging.DEBUG)
+
+    def getLogger(self):
+        return self.logger
+
 
 class Core:
-    def __init__(self, debug):
-        self.logger = logging.getLogger("Core")
-        if args.debug:
-            self.logger.setLevel(logging.DEBUG)
+    def __init__(self, dbfile, debug):
+        self.logger = Logger("Core", debug).getLogger()
+        self.logger.info("initialization starting...")
+        self.actor_queue = {}
+        self.db = dataset.connect('sqlite:///db.sqlite')
+        self.logger.info("initialization complete.")
+
+
+    def actor_add_level(self, actorId, actorType, maxLevel, level):
+        ret = {}
+        if not self.db["actors"].find_one(actorId=actorId):
+            self.db['actors'].insert({"actorId": actorId, "actorType": actorType,
+                "maxLevel": maxLevel})
+        self.db['actor_levels'].insert({"actorId": actorId, "ts": time.time(), "level":
+            level})
+
+        return ret
+
+    def actor_add_queue(self, actorId, command, data):
+        self.actor_queue[actorId] = {"command": command, "data": data}
+
+
+    def actor_get_actors(self):
+        ret = {}
+        for s in self.db["actors"]:
+            actorId = s["actorId"]
+            ret[actorId] = self.actor_get_info(actorId)
+        return ret
+
+
+    def actor_get_info(self, actorId):
+        ret = {}
+        q = self.db["actors"].find_one(actorId=actorId)
+        ret["actorId"] = actorId
+        ret["maxLevel"] = "0x%x" % q["maxLevel"]
+        ret["actorType"] = q["actorType"]
+        q = self.db["actor_levels"].find_one(actorId=actorId, order_by="-ts")
+        ret["ts"] = q["ts"]
+        ret["level"] = "0x%x" % q["level"]
+        return ret
+
+
+    def actor_get_levels(self, actorId, limit=None):
+        ret = self.actor_get_info(actorId)
+
+        levels = {}
+        if limit:
+            query = self.db["actor_levels"].find(actorId=actorId, _limit=limit)
         else:
-            self.logger.setLevel(logging.INFO)
+            query = self.db["actor_levels"].find(actorId=actorId)
+        k = query.keys
+        for q in query:
+            levels[q["ts"]] = "0x%x" % q["level"]
+        ret["levels"] = levels
 
-        self.actors = {}
-        self.sensors = {}
-
-    def actors_get(self):
-        ret = {}
-        # remove inactive actors
-        for a in list(self.actors):
-            if time.time() - self.actors[a].ts > TIMEOUT_CLIENT:
-                self.logger.info("removing inactive actor: %s" % a)
-                self.actors.pop(a)
-
-        for a in self.actors:
-            ret[a] = self.actors[a].get_info()
         return ret
 
-    def sensors_get(self):
-        ret = {}
-        # remove inactive sensors
-        for s in list(self.sensors):
-            if time.time() - self.sensors[s].ts > TIMEOUT_CLIENT:
-                self.logger.info("removing inactive sensor: %s" % s)
-                self.sensors.pop(s)
 
-        for s in self.sensors:
-            ret[s] = self.sensors[s].get_info()
+    def actor_get_queue(self, actorId):
+        ret = {}
+        if actorId in self.actor_queue:
+            ret = self.actor_queue.pop(actorId)
         return ret
 
-    def update(self):
-        # TODO
-        pass
+    def sensor_get_info(self, sensorId):
+        ret = {}
+        q = self.db["sensors"].find_one(sensorId=sensorId)
+        ret["sensorId"] = sensorId
+        ret["sensorType"] = q["sensorType"]
+        q = self.db["sensor_values"].find_one(sensorId=sensorId, order_by="-ts")
+        ret["ts"] = q["ts"]
+        return ret
+
+
+    def sensor_get_values(self, sensorId, limit=None):
+        ret = self.sensor_get_info(sensorId)
+
+        values = {}
+        if limit:
+            query = self.db["sensor_values"].find(sensorId=sensorId, _limit=limit)
+        else:
+            query = self.db["sensor_values"].find(sensorId=sensorId)
+        k = query.keys
+        for q in query:
+            values[q["ts"]] = q["value"]
+        ret["values"] = values
+
+        return ret
+
+
+    def sensor_get_sensors(self):
+        ret = {}
+        for s in self.db["sensors"]:
+            sensorId = s["sensorId"]
+            ret[sensorId] = self.sensor_get_info(sensorId)
+        return ret
+
+
+    def sensor_add_value(self, sensorId, sensorType, value):
+        ret = {}
+        if not self.db["sensors"].find_one(sensorId=sensorId):
+            self.db['sensors'].insert({"sensorId": sensorId, "sensorType": sensorType})
+        self.db['sensor_values'].insert({"sensorId": sensorId, "ts": time.time(), "value":
+            value})
+
+        return ret
 
 
 app = Flask(__name__)
 
 
-@app.route("/actors/get", methods=['GET'])
-def actors_get():
-    ret = core.actors_get()
-    return make_response(jsonify(ret), 200)
-
-
-@app.route("/actors/update", methods=['POST'])
-def actors_update():
+@app.route("/actor/command", methods=['POST'])
+def actor_command():
     ret = {}
     try:
         content = request.json
-        # logger.debug("/actors/update: %s" % content)
-        id_a = content.get("id")
-        level = content.get("level")
-        if id_a not in core.actors:
-            type_a = content.get("type")
-            levels = content.get("levels")
-            core.actors[id_a] = Client.Actor(id_a, type_a, levels)
-        actor = core.actors[id_a]
-        actor.update()
-        actor.level = level
-        ret = actor.get_queue()
-    except Exception as ex:
-        logger.error('Exception Type:%s, args:\n%s' % (type(ex).__name__, ex.args))
-        abort(400)
-
-    return make_response(jsonify(ret), 200)
-
-
-@app.route("/actors/command", methods=['POST'])
-def actors_command():
-    ret = {}
-    try:
-        content = request.json
-        # logger.debug("/actors/command: %s" % content)
-        id_a = content.get("id")
+        actorId = content.get("id")
         command = content.get("command")
         data = content.get("data")
-        if id_a in core.actors:
-            actor = core.actors[id_a]
-            actor.add_queue(command, data)
-
+        if core.db["actors"].find_one(actorId=actorId):
+            core.actor_add_queue(actorId, command, data)
     except Exception as ex:
         logger.error('Exception Type:%s, args:\n%s' % (type(ex).__name__, ex.args))
         abort(400)
@@ -143,49 +202,83 @@ def actors_command():
     return make_response(jsonify(ret), 200)
 
 
-@app.route("/actors/get_level/<id_a>", methods=['GET'])
-def actors_get_level(id_a):
+@app.route("/actor/get", methods=['GET'])
+def actor_get():
+    ret = core.actor_get_actors()
+    return make_response(jsonify(ret), 200)
+
+
+@app.route("/actor/get_levels/<actorId>", methods=['GET'])
+def actor_get_level(actorId):
     ret = {}
-    # logger.debug("actors/get_level/%s" % id_a)
-    if id_a in core.actors:
-        ret = core.actors[id_a].get_level()
+    if request.args.get("limit"):
+        limit=int(request.args.get("limit"))
     else:
-        logger.debug("id_a: %s not in core.actors" % id_a)
-    return make_response(jsonify(ret), 200)
-
-
-@app.route("/sensors/get", methods=['GET'])
-def sensors_get():
-    ret = core.sensors_get()
-    return make_response(jsonify(ret), 200)
-
-
-@app.route("/sensors/get_values/<id_s>", methods=['GET'])
-def sensors_get_values(id_s):
-    ret = {}
-    logger.debug("sensors/get_values/%s" % id_s)
-    if id_s in core.sensors:
-        ret = core.sensors[id_s].get_values()
+        limit=None
+    logger.debug("actor/get_levels/%s, limit: %s" % (actorId, limit))
+    if actorId == "all":
+        for q in core.db["actors"]:
+            s = q["actorId"]
+            ret[s] = core.actor_get_levels(s, limit=limit)
+    elif core.db["actors"].find_one(actorId=actorId):
+        ret[actorId] = core.actor_get_levels(actorId, limit=limit)
     else:
-        logger.debug("id_s: %s not in core.sensors" % id_s)
+        abort(404)
     return make_response(jsonify(ret), 200)
 
 
-@app.route("/sensors/update", methods=['POST'])
-def sensors_add_value():
+@app.route("/actor/update", methods=['POST'])
+def actor_update():
     ret = {}
     try:
         content = request.json
-        # logger.info("sensors_update: %s" % (content))
-        id_s = content.get("id")
+        actorId = content.get("id")
+        actorType = content.get("type")
+        level = content.get("level")
+        maxLevel = content.get("maxLevel")
+        core.actor_add_level(actorId, actorType, maxLevel, level)
+        ret = core.actor_get_queue(actorId)
+    except Exception as ex:
+        logger.error('Exception Type:%s, args:\n%s' % (type(ex).__name__, ex.args))
+        abort(400)
+
+    return make_response(jsonify(ret), 200)
+
+
+@app.route("/sensor/get", methods=['GET'])
+def sensor_get_sensors():
+    ret = core.sensor_get_sensors()
+    return make_response(jsonify(ret), 200)
+
+
+@app.route("/sensor/get_values/<sensorId>", methods=['GET'])
+def sensor_get_values(sensorId):
+    ret = {}
+    if request.args.get("limit"):
+        limit=int(request.args.get("limit"))
+    else:
+        limit=None
+    logger.debug("sensor/get_values/%s, limit: %s" % (sensorId, limit))
+    if sensorId == "all":
+        for q in core.db["sensors"]:
+            s = q["sensorId"]
+            ret[s] = core.sensor_get_values(s, limit=limit)
+    elif core.db["sensors"].find_one(sensorId=sensorId):
+        ret[sensorId] = core.sensor_get_values(sensorId, limit=limit)
+    else:
+        abort(404)
+    return make_response(jsonify(ret), 200)
+
+
+@app.route("/sensor/update", methods=['POST'])
+def sensor_add_value():
+    ret = {}
+    try:
+        content = request.json
+        sensorId = content.get("id")
+        sensorType = content.get("type")
         value = content.get("value")
-        if id_s not in core.sensors:
-            type_s = content.get("type")
-            core.sensors[id_s] = Client.Sensor(id_s, type_s, NUM_VALUES)
-            logger.info("created sensor: %s" % (core.sensors[id_s].get_info()))
-        sensor = core.sensors[id_s]
-        sensor.update()
-        ret = sensor.add_value(value)
+        ret = core.sensor_add_value(sensorId, sensorType, value)
     except Exception as ex:
         logger.error('Exception Type:%s, args:\n%s' % (type(ex).__name__, ex.args))
         abort(400)
@@ -194,14 +287,7 @@ def sensors_add_value():
 
 if __name__ == "__main__":
     args, config = setup()
-    core = Core(args.debug)
-    FORMAT='%(asctime)-8s %(name)12s::%(levelname)-7s %(message)s'
-    DATEFMT='%H:%M:%S'
-    logging.basicConfig(level=logging.WARNING, format=FORMAT, datefmt=DATEFMT)
-    logger = logging.getLogger("homecontrol")
-    if args.debug:
-        logger.setLevel(logging.DEBUG)
-    else:
-        logger.setLevel(logging.WARNING)
+    core = Core(config['dbfile'], args.debug)
+    logger = Logger("main", args.debug).getLogger()
 
     app.run(host='0.0.0.0', port=config['port'], debug=args.debugflask)

scripts/actor_command.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+ID=${1:-1}
+LEVEL=${2:-65280}
+COMMAND=set_level
+DATA='{"id":"'$ID'", "command":"'$COMMAND'", "data":{"level":"'$LEVEL'"}}'
+
+REQ=POST
+URL=http://localhost:5000/actor/command
+echo "$REQ $URL --> $DATA"
+read -p "continue?" -n 1 cont
+curl \
+    -X $REQ $URL \
+    -H "Content-Type: application/json" \
+    -d "$DATA"
+

scripts/actor_get.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+REQ=GET
+URL=http://localhost:5000/actor/get
+
+echo "$REQ $URL"
+read -p "continue?" -n 1 foo
+curl \
+  -X $REQ $URL \
+  -H "Content-Type: application/json"

scripts/actor_get_levels.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+ACTORID=${1:-11111111}
+LIMIT=${2:-0}
+REQ=GET
+URL=http://localhost:5000/actor/get_levels/$ACTORID?limit=$LIMIT
+
+echo "$REQ $URL"
+read -p "continue?" -n 1 foo
+curl \
+  -X $REQ $URL \
+  -H "Content-Type: application/json"

scripts/sensor_get.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+REQ=GET
+URL=http://localhost:5000/sensor/get
+
+echo "$REQ $URL"
+read -p "continue?" -n 1 foo
+curl \
+  -X $REQ $URL \
+  -H "Content-Type: application/json"

scripts/sensor_get_values.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+SENSORID=${1:-11111111}
+LIMIT=${2:-0}
+REQ=GET
+URL=http://localhost:5000/sensor/get_values/$SENSORID?limit=$LIMIT
+# URL=http://localhost:5000/sensor/get_values/$SENSORID
+
+echo "$REQ $URL"
+read -p "continue?" -n 1 foo
+curl \
+  -X $REQ $URL \
+  -H "Content-Type: application/json"

scripts/sensor_update.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+REQ=POST
+URL=http://localhost:5000/sensor/update
+
+ID=${1:-11111111}
+VALUE=${2:-20.2}
+TYPE=${3:-temperature}
+data="{\"id\":\"$ID\", \"type\":\"$TYPE\", \"value\":$VALUE}"
+echo "$REQ $URL $data"
+read -p "continue?" -n 1 foo
+
+curl \
+  -X $REQ $URL \
+  -H "Content-Type: application/json" \
+  -d "$data"