sleep_data / analysis_db.py
Luis Wu
draw predict
78de124
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)