Skip to content
Snippets Groups Projects
Select Git revision
  • b2426ec4a80c0a35734389ee3409d12c6128ce3e
  • master default
  • v2.21.0
  • v2.20.1
  • v2.20.0
  • v2.19.0
  • v2.18.1
  • v2.18.0
  • v2.17.0
  • v2.16.0
  • v2.15.0
  • v2.14.0
  • v2.13.1
  • v2.13.0
  • v2.12.0
  • v2.11.0
  • v2.10.0
  • v2.9.0
  • v2.8.0
  • v2.7.0
  • v2.6.0
  • v2.5.0
22 results

Parameters.py

Blame
  • Database.py 5.70 KiB
    import sqlite3
    from datetime import datetime
    from enum import Enum
    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'
        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)
            self.__query(f'''CREATE TABLE IF NOT EXISTS {self.TABLE_SENSOR} (
                             id INTEGER PRIMARY KEY AUTOINCREMENT,
                             device_id INTEGER,
                             name TEXT NOT NULL, 
                             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)
    
        def __query(self, query, *args, fetch_type=FetchType.ALL):
            connection = sqlite3.connect(self._databasePath)
            connection.row_factory = Database.namedtuple_factory
    
            with connection:
                cursor = connection.cursor()
                try:
                    cursor.execute(query, args)
    
                    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)
    
        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)
    
        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)
    
        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)
    
        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',
                                deviceID,
                                fetch_type=FetchType.ALL)
    
        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 = ?',
                                deviceID, sensorName,
                                fetch_type=FetchType.ONE)
    
        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)
    
        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"]}" '
                         f'(value: "{value}", device_id "{sensor["device_id"]}")')
            self.__query(f'INSERT INTO {self.TABLE_MEASUREMENT}(sensor_id, value, timestamp ) VALUES(?, ?, ?)',
                         sensorID, value, self.__get_current_datetime(),
                         fetch_type=FetchType.NONE)