analysis_db.py

import sqlite3
import sys

from util.debug import debug_print
from draw_db import draw_lists, min_max_list, up_down_list
from util.date_util import format_date

MIN_FROM = 0
MIN_TO = 25
LIGHT_FROM = 0
LIGHT_TO = 5
ACC_FROM = 0
ACC_TO = 10


def get_filename(file_obj):
    return file_obj.name


def read_db(db_file, min_from, min_to, light_from, light_to, acc_from, acc_to):
    db_name = get_filename(db_file)
    global MIN_FROM
    global MIN_TO
    global LIGHT_FROM
    global LIGHT_TO
    global ACC_FROM
    global ACC_TO
    MIN_FROM = min_from
    MIN_TO = min_to
    LIGHT_FROM = light_from
    LIGHT_TO = light_to
    ACC_FROM = acc_from
    ACC_TO = acc_to
    return read_section(db_name, 'tb_section')


def read_section(db_name, tb_name='tb_section'):
    res = []
    debug_print('read_section, db_name = ', db_name, ', tb_name = ', tb_name)
    conn = sqlite3.connect(db_name)

    if not tb_name:
        table_cursor = conn.cursor()
        table_cursor.execute(' SELECT name FROM sqlite_master WHERE type=\'table\'; ')
        table_names = table_cursor.fetchall()
        print('table_names = ', str(table_names))
        return

    section_cursor = conn.execute(' SELECT section_id, section_date, section_end_date, section_mark FROM ' + tb_name)
    for section_row in section_cursor:
        section_id = section_row[0]
        section_date = section_row[1]
        section_end_date = section_row[2]
        section_mark = section_row[3]

        if section_mark == -1:
            print('section_id = ', section_id, ', section_date = ', str(section_date), ', section_end_date = ',
                  str(section_end_date))
            plot = read_sample(conn, section_id, tb_name)
            res.append(plot)
    conn.close()

    for idx in range(10):
        if idx >= len(res):
            res.append(None)

    return res


def clamp(_value, _min, _max):
    if _value < _min:
        return _min
    elif _value > _max:
        return _max
    else:
        return _value


def clamp_list(_list, _min, _max):
    return [clamp(x, _min, _max) for x in _list]


def predict_sleeping(min_list, light_list, acc_list):
    res = []

    cnt = min(len(min_list), len(light_list), len(acc_list))
    for idx in range(cnt):
        min_from_to = MIN_FROM <= min_list[idx] <= MIN_TO
        light_from_to = LIGHT_FROM <= light_list[idx] <= LIGHT_TO
        acc_from_to = ACC_FROM <= acc_list[idx] <= ACC_TO
        sleeping = 1 if min_from_to and light_from_to and acc_from_to else 0
        res.append(sleeping)

    return res


def read_sample(conn, section_id, tb_name):
    sql_str = ' SELECT section_id, _sample_id, _sample_end_id, section_date, section_end_date FROM ' + str(
        tb_name) + ' WHERE section_id == ' + str(section_id)
    section_cursor = conn.execute(sql_str)
    for section_row in section_cursor:
        section_id = section_row[0]
        _sample_id = section_row[1]
        _sample_end_id = section_row[2]
        section_date = section_row[3]
        section_end_date = section_row[4]
        debug_print('section_id = ', section_id, ', _sample_id = ', _sample_id, ', _sample_end_id = ',
                    _sample_end_id)

        date_list = []
        max_list = []
        min_list = []
        acc_list = []
        light_list = []
        screen_list = []
        predict_list = []
        sample_sql = "SELECT _id, date, max, min, acc_max_dx, acc_max_dy, acc_max_dz, avg, max_snoring FROM sample_table WHERE _id >= " + str(
            _sample_id) + ' AND ' + ' _id <= ' + str(_sample_end_id)
        sample_cursor = conn.execute(sample_sql)
        for sample_row in sample_cursor:
            _id = sample_row[0]
            date = sample_row[1]
            _max = sample_row[2]
            _min = sample_row[3]
            acc_max = sample_row[4] + sample_row[5] + sample_row[6]
            light = sample_row[7]
            screen = sample_row[8]
            debug_print('_id = ', _id, 'date = ', format_date(date))

            date_list.append(format_date(date))
            max_list.append(_max)
            min_list.append(_min)
            acc_list.append(acc_max)
            light_list.append(light)
            screen_list.append(screen)

        title_str = str(section_id) + ', ' + str(format_date(section_date)) + ', ' + str(format_date(section_end_date))

        if len(light_list) > 0:
            light_list = clamp_list(light_list, 0, 100)

        if len(screen_list) > 0:
            screen_list = min_max_list(screen_list, 1, 11)

        if len(min_list) > 0 and len(light_list) > 0 and len(acc_list) > 0:
            predict_list = predict_sleeping(min_list, light_list, acc_list)

        return draw_lists(title_str, date_list, max=max_list, min=min_list, acc=acc_list, screen=screen_list,
                          light=light_list, predict=predict_list)


if __name__ == '__main__':
    argv = sys.argv[1:]
    argc = len(sys.argv[1:])
    # print('sys.argv[1:] = ', argv, ', ', str(len(argv)))

    if argc < 1:
        print('USAGE: python analysis_db.py xxx.db')
        exit(1)

    db_uri = argv[0] if argc >= 1 else ''
    tb_name = argv[1] if argc >= 2 else ''
    read_section(db_uri, tb_name)