smart_brain/function/db.py

from function.modules import heating_function
import os
import sqlite3

db_file = os.path.join(os.path.dirname(__file__), '..', 'database.db')

db_mapping_radiator = {
    0: heating_function.KEY_AWAY_MODE,
    1: heating_function.KEY_SUMMER_MODE,
    2: heating_function.KEY_USER_TEMPERATURE_SETPOINT,
    3: heating_function.KEY_TEMPERATURE_SETPOINT
}


def get_radiator_data(topic):
    db_data = __storage__().get_radiator_data(topic)
    rv = {}
    for index in db_mapping_radiator:
        rv[db_mapping_radiator[index]] = db_data[index]
    return rv


def set_radiator_data(device, key, data):
    if key in db_mapping_radiator.values():
        db_data = []
        for index in range(0, len(db_mapping_radiator)):
            db_data.append(device.get(db_mapping_radiator[index]))
        return __storage__().store_radiator_data(device.heating_valve.topic, db_data)


class __storage__(object):
    def __init__(self):
        self.conn = sqlite3.connect(db_file)
        self.c = self.conn.cursor()
        with self.conn:
            self.c.execute("""CREATE TABLE IF NOT EXISTS radiator (
                    topic text PRIMARY KEY,
                    away_mode integer,
                    summer_mode integer,
                    user_temperatur_setpoint real,
                    temperatur_setpoint real
                )""")

    def store_radiator_data(self, topic, target_data):
        try:
            with self.conn:
                self.c.execute(
                    'INSERT INTO radiator VALUES (?, ?, ?, ?, ?)', [topic] + target_data)
        except sqlite3.IntegrityError:
            db_data = self.get_radiator_data(topic)
            if db_data != target_data:
                with self.conn:
                    self.c.execute(
                        'UPDATE radiator SET away_mode = ?, summer_mode = ?, user_temperatur_setpoint = ?, temperatur_setpoint = ? WHERE topic = ?', target_data + [topic])

    def get_radiator_data(self, topic):
        """ returns a list [away_mode, summer_mode, user_temperatur_setpoint, temperatur_setpoint] or [None, None, None, None]"""
        self.c.execute("SELECT * FROM radiator WHERE topic=?", (topic, ))
        data = self.c.fetchone()
        if data is not None:
            data = list(data)
            data[1] = data[1] == 1
            data[2] = data[2] == 1
            return data[1:]
        else:
            return [None, None, None, None]

    def __del__(self):
        self.conn.close()