小米手环数据自动化导出工具

# date_data_extraction.py 读取sqlite数据 import sqlite3 as db # sqlite3数据库连接的包 import pandas as pd import numpy as np import ast # 字符串转为字典的包 import datetime import json # 控制台输出不省略 pd.set_option('display.max_columns', 1000) pd.set_option('display.width', 500) # pd.set_option('display.max_colwidth', 1000) # 从SQLite文件中读取数据 db_path = 'C:\\Users\\17588\\Documents\\Programming\\appdb\\health2\\apps\\com.xiaomi.hm.health\\db\\origin_db_d294ad468810c0e8650961d77f20a9d3' def readFronSqllite(db_path, exectCmd): conn = db.connect(db_path) # 该 API 打开一个到 SQLite 数据库文件 database 的链接，如果数据库成功打开，则返回一个连接对象 cursor = conn.cursor() # 该例程创建一个 cursor，将在 Python 数据库编程中用到。 conn.row_factory = db.Row # 可访问列信息 cursor.execute(exectCmd) # 该例程执行一个 SQL 语句 rows = cursor.fetchall() # 该例程获取查询结果集中所有（剩余）的行，返回一个列表。当没有可用的行时，则返回一个空的列表。 table = pd.DataFrame(rows, columns=['DATE', 'SUMMARY']) return table # print(rows[0][2]) # 选择某一列数据 # 解析ARPA 单帧信息，我的代码没有用到这块 def readfromAppaFrame(ARPAFrame): subARPA = ARPAFrame.split(',') print(subARPA) if __name__ == "__main__": table = readFronSqllite( "C:\\Users\\17588\\Documents\\Programming\\appdb\\health3\\apps\\com.xiaomi.hm.health\\db\\origin_db_d294ad468810c0e8650961d77f20a9d3", "select DATE,SUMMARY from DATE_DATA order by date") # 以下空列表用于保存将要加入dataframe的每列数据 st = [] ed = [] dp = [] lt = [] wk = [] dt = [] ss = [] rhr = [] ttl = [] dis = [] cal = [] pai = [] # 我这里有6个列来保存每日3段小睡的数据，实际上这里最好用动态分配内存的方式来做会更加灵活 o_st1=[] o_st2=[] o_st3=[] o_ed1=[] o_ed2=[] o_ed3=[] for SUMMARY in table["SUMMARY"]: # 循环遍历每一行的SUMMARY # 保持字典各个层级的相关数据到空列表 st.append(datetime.datetime.fromtimestamp(ast.literal_eval(SUMMARY)["slp"]["st"])) ed.append(datetime.datetime.fromtimestamp(ast.literal_eval(SUMMARY)["slp"]["ed"])) dp.append(ast.literal_eval(SUMMARY)["slp"]["dp"]) lt.append(ast.literal_eval(SUMMARY)["slp"]["lt"]) wk.append(ast.literal_eval(SUMMARY)["slp"]["wk"]) ss.append(ast.literal_eval(SUMMARY)["slp"]["ss"]) rhr.append(ast.literal_eval(SUMMARY)["slp"]["rhr"]) ttl.append(ast.literal_eval(SUMMARY)["stp"]["ttl"]) dis.append(ast.literal_eval(SUMMARY)["stp"]["dis"]) cal.append(ast.literal_eval(SUMMARY)["stp"]["cal"]) if(ast.literal_eval(SUMMARY).get("pai")): pai.append(ast.literal_eval(SUMMARY)["pai"]["tp"]) else: pai.append(0) # 由于不是每天都有REM数据，所以需要进行一个判断，没有REM的保存为0 if (ast.literal_eval(SUMMARY)["slp"].get("dt")): dt.append(ast.literal_eval(SUMMARY)["slp"]["dt"]) else: dt.append(0) # 由于不是每天都有小睡数据，所以需要进行一个判断，没有REM的保存为0 if (ast.literal_eval(SUMMARY)["slp"].get("odd_stage")): ii=0 for i,stage in enumerate(ast.literal_eval(SUMMARY)["slp"]["odd_stage"]): if i==0: o_st1.append(stage["start"]) o_ed1.append(stage["stop"]) ii = i elif i==1: o_st2.append(stage["start"]) o_ed2.append(stage["stop"]) ii = i elif i==2: o_st3.append(stage["start"]) o_ed3.append(stage["stop"]) ii = i if ii==0: o_st2.append(0) o_st3.append(0) o_ed2.append(0) o_ed3.append(0) elif ii==1: o_st3.append(0) o_ed3.append(0) else: o_st1.append(0) o_st2.append(0) o_st3.append(0) o_ed1.append(0) o_ed2.append(0) o_ed3.append(0) # datafram新增列，把我们提取的数据拷贝进去 table["st"] = st table["ed"] = ed table["dp"] = dp table["lt"] = lt table["wk"] = wk table["dt"] = dt table["ss"] = ss table["rhr"] = rhr table["ttl"] = ttl table["dis"] = dis table["cal"] = cal table["pai"] = pai table["o_st1"] = o_st1 table["o_ed1"] = o_ed1 table["o_st2"] = o_st2 table["o_ed2"] = o_ed2 table["o_st3"] = o_st3 table["o_ed3"] = o_ed3 # 将datafram保存成excle，后面的数据分析将在excel里面进行 table.to_csv("sleep.csv",index=False)