system-monitor/database/main.py

import os
import sqlite3

from models import HardwareType, Hardware, SensorType, Sensor, SensorValue


directory_path = os.path.join(
    os.getenv('appdata'),
    'csasq',
    'Hardware Monitor',
)
file_path = os.path.join(
    directory_path,
    'data.sql',
)


def connect():
    os.makedirs(
        name=directory_path,
        exist_ok=True,
    )
    return sqlite3.connect(
        database=file_path,
    )


with connect() as connection:
    cursor = connection.cursor()
    sql = '''
    create table if not exists settings (
        name,
        value,
        primary key (name)
    );
    
    create table if not exists hardware_types (
        id                  integer,
        name,
        primary key (id),
        unique (name)
    );
    
    create table if not exists hardware (
        id                  integer,
        identifier,
        name,
        hardware_type_id,
        primary key (id),
        foreign key (hardware_type_id) references hardware_types on delete cascade on update cascade,
        unique (identifier)
    );
    
    create table if not exists sensor_types (
        id                  integer,
        name,
        primary key (id),
        unique (name)
    );
    
    create table if not exists sensors (
        id                  integer,
        hardware_id,
        sensor_type_id,
        identifier,
        name,
        'index',
        is_default_hidden,
        enabled,
        primary key (id),
        foreign key (hardware_id) references hardware on delete cascade on update cascade,
        foreign key (sensor_type_id) references sensor_types on delete cascade on update cascade,
        unique (identifier, sensor_type_id)
    );
    
    create table if not exists sensor_values (
        sensor_id,
        timestamp,
        value,
        foreign key (sensor_id) references sensors
    );
    
    pragma foreign_keys = on;
    '''
    cursor.executescript(sql)


def get_sensors() -> list[Sensor]:
    with connect() as connection:
        cursor = connection.cursor()
        sql = '''
        select
            sensors.id,
            hardware.id,
            hardware.identifier,
            hardware.name,
            hardware_types.id,
            hardware_types.name,
            sensor_types.id,
            sensor_types.name,
            sensors.identifier,
            sensors.name,
            sensors.'index',
            sensors.is_default_hidden,
            sensors.enabled
        from
            sensors
                left outer join hardware on
                    sensors.hardware_id = hardware.id
                left outer join hardware_types on
                    hardware.hardware_type_id = hardware_types.id
                left outer join sensor_types on
                    sensors.sensor_type_id = sensor_types.id;
        '''
        cursor.execute(sql)
        records = cursor.fetchall()
    return list(
        Sensor(
            id=sensor_id,
            hardware=Hardware(
                id=sensor_hardware_id,
                identifier=sensor_hardware_identifier,
                name=sensor_hardware_name,
                hardware_type=HardwareType(
                    id=sensor_hardware_type_id,
                    name=sensor_hardware_type_name,
                ),
            ),
            sensor_type=SensorType(
                id=sensor_type_id,
                name=sensor_type_name,
            ),
            identifier=sensor_identifier,
            name=sensor_name,
            index=sensor_index,
            is_default_hidden=sensor_is_default_hidden,
            enabled=sensor_enabled,
        )
        for
        sensor_id,
        sensor_hardware_id,
        sensor_hardware_identifier,
        sensor_hardware_name,
        sensor_hardware_type_id,
        sensor_hardware_type_name,
        sensor_type_id,
        sensor_type_name,
        sensor_identifier,
        sensor_name,
        sensor_index,
        sensor_is_default_hidden,
        sensor_enabled,
        in
        records
    )


def get_sensor_values() -> list[SensorValue]:
    with connect() as connection:
        cursor = connection.cursor()
        sql = '''
        select
            sensor_values.sensor_id,
            sensor_values.timestamp,
            sensor_values.value
        from
            sensor_values
        where
            sensor_values.sensor_id in ?;
        '''
        cursor.execute(
            sql,
            (
                [15, 16, 17],
            ),
        )
        records, = cursor.fetchall()
    return list(
        SensorValue(
            sensor_id=sensor_id,
            timestamp=timestamp,
            value=value,
        )
        for
        sensor_id,
        timestamp,
        value,
        in
        records
    )