Skip to content
Snippets Groups Projects
Database.py 5.7 KiB
Newer Older
  • Learn to ignore specific revisions
  • from datetime import datetime
    
    from enum import Enum
    
    Robert Goldmann's avatar
    Robert Goldmann committed
    from typing import Dict, List
    
    
    from TheCodeLabs_BaseUtils import DefaultLogger
    
    from logic import Constants
    
    LOGGER = DefaultLogger().create_logger_if_not_exists(Constants.APP_NAME)
    
    
    
    class FetchType(Enum):
        NONE = 1
        ONE = 2
        ALL = 3
    
    
    
    class Database:
        TABLE_DEVICE = 'device'
        TABLE_SENSOR = 'sensor'
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        TABLE_MEASUREMENT = 'measurement'
    
        DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
    
    
        @staticmethod
        def namedtuple_factory(cursor, row):
            """
            Returns sqlite rows as dicts.
            """
            d = {}
            for idx, col in enumerate(cursor.description):
                d[col[0]] = row[idx]
            return d
    
    
        def __init__(self, databasePath):
            self._databasePath = databasePath
            self.__create_database()
    
        def __create_database(self):
            LOGGER.debug('Creating database tables...')
    
            self.__query(f'''CREATE TABLE IF NOT EXISTS {self.TABLE_DEVICE} (
    
                                id INTEGER PRIMARY KEY AUTOINCREMENT, 
    
                                name TEXT NOT NULL)''', fetch_type=FetchType.NONE)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
            self.__query(f'''CREATE TABLE IF NOT EXISTS {self.TABLE_SENSOR} (
    
                             id INTEGER PRIMARY KEY AUTOINCREMENT,
                             device_id INTEGER,
                             name TEXT NOT NULL, 
    
    Robert Goldmann's avatar
    Robert Goldmann committed
                             type TEXT NOT NULL)''', fetch_type=FetchType.NONE)
            self.__query(f'''CREATE TABLE IF NOT EXISTS {self.TABLE_MEASUREMENT} (
                             id INTEGER PRIMARY KEY AUTOINCREMENT,
                             sensor_id INTEGER,
    
                             value TEXT NOT NULL,
    
                             timestamp TEXT NOT NULL)''', fetch_type=FetchType.NONE)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def __query(self, query, *args, fetch_type=FetchType.ALL):
    
            connection = sqlite3.connect(self._databasePath)
            connection.row_factory = Database.namedtuple_factory
    
    
    Robert Goldmann's avatar
    Robert Goldmann committed
            with connection:
                cursor = connection.cursor()
                try:
                    cursor.execute(query, args)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
                    if fetch_type == FetchType.ONE:
                        return cursor.fetchone()
                    if fetch_type == FetchType.ALL:
                        return cursor.fetchall()
                finally:
                    cursor.close()
    
        def __get_current_datetime(self):
            return datetime.strftime(datetime.now(), self.DATE_FORMAT)
    
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def get_all_devices(self) -> List[Dict[str, str]]:
    
            return self.__query(f'SELECT * FROM {self.TABLE_DEVICE} ORDER BY name', fetch_type=FetchType.ALL)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def get_device(self, deviceID: int) -> Dict[str, str] or None:
            return self.__query(f'SELECT * FROM {self.TABLE_DEVICE} WHERE id = ?', deviceID, fetch_type=FetchType.ONE)
    
        def get_device_by_name(self, deviceName: str) -> Dict[str, str] or None:
            return self.__query(f'SELECT * FROM {self.TABLE_DEVICE} WHERE name = ?', deviceName, fetch_type=FetchType.ONE)
    
        def add_device(self, deviceName: str):
    
            LOGGER.debug(f'Inserting new device "{deviceName}"')
            self.__query(f'INSERT INTO {self.TABLE_DEVICE}(name) VALUES(?)', deviceName, fetch_type=FetchType.NONE)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def get_all_sensors(self) -> List[Dict[str, str]]:
    
            return self.__query(f'SELECT * FROM {self.TABLE_SENSOR} ORDER BY device_id, name', fetch_type=FetchType.ALL)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def get_all_sensors_for_device(self, deviceID: int) -> List[Dict[str, str]]:
    
            return self.__query(f'SELECT * FROM {self.TABLE_SENSOR} WHERE device_id = ? ORDER BY name',
    
    Robert Goldmann's avatar
    Robert Goldmann committed
                                deviceID,
    
                                fetch_type=FetchType.ALL)
    
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def get_sensor(self, sensorID: int) -> Dict[str, str] or None:
            return self.__query(f'SELECT * FROM {self.TABLE_SENSOR} WHERE id = ?',
                                sensorID,
                                fetch_type=FetchType.ONE)
    
        def get_sensor_by_name_and_device_id(self, deviceID: int, sensorName: str) -> Dict[str, str] or None:
    
            return self.__query(f'SELECT * FROM {self.TABLE_SENSOR} WHERE device_id = ? AND name = ?',
    
    Robert Goldmann's avatar
    Robert Goldmann committed
                                deviceID, sensorName,
    
                                fetch_type=FetchType.ONE)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def add_sensor(self, deviceID: int, name: str, sensorType: str):
            LOGGER.debug(f'Inserting new "{sensorType}" sensor "{name}" for device "{deviceID}"')
            self.__query(f'INSERT INTO {self.TABLE_SENSOR}(name, device_id, type) '
                         f'VALUES(?, ?, ?)',
                         name, deviceID, sensorType,
    
                         fetch_type=FetchType.NONE)
    
    Robert Goldmann's avatar
    Robert Goldmann committed
        def get_all_measurements(self) -> List[Dict[str, str]]:
            return self.__query(f'SELECT * FROM {self.TABLE_MEASUREMENT} ORDER BY sensor_id, date(timestamp)',
                                fetch_type=FetchType.ALL)
    
        def get_measurement(self, measurementID: int) -> Dict[str, str] or None:
            return self.__query(f'SELECT * FROM {self.TABLE_MEASUREMENT} WHERE id = ?',
                                measurementID,
                                fetch_type=FetchType.ALL)
    
        def get_all_measurements_for_sensor(self, sensorID: int) -> List[Dict[str, str]]:
            return self.__query(f'SELECT * FROM {self.TABLE_MEASUREMENT} WHERE sensor_id = ? '
                                f'ORDER BY sensor_id, date(timestamp)',
                                sensorID,
                                fetch_type=FetchType.ALL)
    
        def add_measurement(self, sensorID: int, value: str):
            sensor = self.get_sensor(sensorID)
            LOGGER.debug(f'Inserting new measurement for sensor "{sensor["name"]}" '
    
    Robert Goldmann's avatar
    Robert Goldmann committed
                         f'(value: "{value}", device_id "{sensor["device_id"]}")')
    
    Robert Goldmann's avatar
    Robert Goldmann committed
            self.__query(f'INSERT INTO {self.TABLE_MEASUREMENT}(sensor_id, value, timestamp ) VALUES(?, ?, ?)',
                         sensorID, value, self.__get_current_datetime(),
    
                         fetch_type=FetchType.NONE)