Spaces:
Sleeping
Sleeping
| 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) | |