SQLite数据库（www.sqlite.org）是一个轻型的数据库引擎，应用非常广泛。我们使用SQLite来存储采集到的通讯数据，一帧通讯原始数据加上时间戳后作为一条记录存入数据库中。在这个项目中，只需要用到数据库的存储功能，不需要查询之类的（到本地上位机上解析数据时才会用到），所以下面也只介绍存储相关的部分，其他功能可以参考这个网址：
https://www.runoob.com/sqlite/sqlite-python.html
程序说明
用下面这些函数就可以插入记录完成存储，还需要解决的是SQL语句：sql_create_table、sql_insert_record，以及记录集合：records_list。
1 import sqlite3 2 3 # 连接到SQLite数据库：DB_NAME，如果不存在就创建 4 conn = sqlite3.connect(DB_NAME) 5 # 创建一个游标Cursor，可以通过Cursor来执行SQL语句 6 cursor = conn.cursor() 7 # 创建记录表：records 8 cursor.execute(sql_create_table) 9 # 插入多条记录 10 cursor.executemany(sql_insert_record, records_list) 11 # 关闭Cursor 12 cursor.close() 13 # 提交事务 14 conn.commit() 15 # 关闭数据库连接 16 conn.close()
>> SQL语句：sql_create_table、sql_insert_record
sql_create_table
在数据库中建一个表records，表中记录的结构为：单条记录结构：id（自增），时间戳（text），从数据库创建开始的累计时间（real），通讯原始数据（400个，int）
sql_insert_record
插入记录时只需要：时间戳、累计时间、通讯数据*400
1 def init_sql_sentences(param_num): 2 global sql_create_table 3 global sql_insert_record 4 5 # sql_create_table 6 sql_create_table = 'create table if not exists records(' 7 sql_create_table += 'id integer primary key autoincrement, ' 8 sql_create_table += 'datetime text, ' 9 sql_create_table += 'runningtime real, ' 10 for i in range(0, param_num - 1): 11 sql_create_table += 'param' + str(i) + ' int, ' 12 sql_create_table += 'param' + str(param_num - 1) + ' int)' 13 14 # sql_insert_record 15 sql_insert_record = 'insert into records values (null, ?, ?, ' 16 for i in range(0, param_num - 1): 17 sql_insert_record += '?, ' 18 sql_insert_record += '?)'
>> 记录集合：records_list
按照记录的格式，将时间戳、通讯数据等存入单条记录record_buff[]，然后再把record_buff[]添加到到记录集records_list[]，这里使用了tuple，把record_buff[]转换成元组，这样就负荷cursor.executemany()函数要求的格式了。
1 records_list = [] 2 record_buff = [] 3 record_buff.append('') 4 record_buff.append(0.0) 5 for i in range(0, DB_PARAMS_LENGTH): 6 record_buff.append(0) 7 8 #### 9 record_buff[0] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) 10 record_buff[1] = round((time.time() - start_time), 1) 11 for i in range(0, DB_PARAMS_LENGTH): 12 record_buff[i + 2] = comn_data[i] 13 comn_data[i] = 0 14 records_list.append(tuple(record_buff[:]))
用模拟通讯数据，生成100条记录，插入数据库，等待0.4s。循环10次以后退出。
pi@raspberrypi:~ $ sudo nano test_sqlite.py
1 # -*- coding:utf-8 -*- 2 import time 3 import sqlite3 4 import os 5 from datetime import datetime 6 7 DEFAULT_DIR = '/home/pi/project/' 8 DB_DIR = '/home/pi/project/data/' 9 DB_TMP_DIR = '/home/pi/project/data/tmp/' 10 DB_UPDATING_DIR = '/home/pi/project/data/updating/' 11 DB_UPDATED_DIR = '/home/pi/project/data/updated/' 12 13 DB_NAME = '' 14 15 PRODUCT_TYPE = 'NULL' 16 DEVICE_ID = 999 17 DB_PARAMS_LENGTH = 400 18 19 sql_create_table = '' 20 sql_insert_record = '' 21 22 def init_sql_sentences(param_num): 23 #### 24 25 def main(): 26 global PRODUCT_TYPE 27 global DEVICE_ID 28 global DB_PARAMS_LENGTH 29 30 record_count = 0 31 insert_count = 0 32 33 record_buff = [] 34 records_list = [] 35 comn_data = [0 for i in range(DB_PARAMS_LENGTH)] 36 37 record_buff.append('') 38 record_buff.append(0.0) 39 for i in range(0, DB_PARAMS_LENGTH): 40 record_buff.append(0) 41 42 if not os.path.exists(DEFAULT_DIR): 43 os.mkdir(DEFAULT_DIR) 44 if not os.path.exists(DB_DIR): 45 os.mkdir(DB_DIR) 46 if not os.path.exists(DB_TMP_DIR): 47 os.mkdir(DB_TMP_DIR) 48 if not os.path.exists(DB_UPDATING_DIR): 49 os.mkdir(DB_UPDATING_DIR) 50 if not os.path.exists(DB_UPDATED_DIR): 51 os.mkdir(DB_UPDATED_DIR) 52 53 init_sql_sentences(DB_PARAMS_LENGTH) 54 55 DB_NAME = DB_TMP_DIR+PRODUCT_TYPE + '_' + str(DEVICE_ID).rjust(3,'0') + '_' 56 + time.strftime('%Y-%m-%d_%H%M%S', time.localtime()) + '.db3' 57 58 print('>> sqlite test start...') 59 print(DB_NAME) 60 61 start_time = time.time() 62 while insert_count < 10: 63 # set fake data 64 for i in range(0, DB_PARAMS_LENGTH): 65 comn_data[i] = i 66 comn_data[0] = record_count 67 68 record_buff[0] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) 69 record_buff[1] = round((time.time() - start_time), 1) 70 for i in range(0, DB_PARAMS_LENGTH): 71 record_buff[i + 2] = comn_data[i] 72 comn_data[i] = 0 73 74 records_list.append(tuple(record_buff[:])) 75 record_count = record_count + 1 76 77 if((record_count % 100) == 0): 78 conn = sqlite3.connect(DB_NAME) 79 cursor = conn.cursor() 80 cursor.execute(sql_create_table) 81 cursor.executemany(sql_insert_record, records_list) 82 cursor.close() 83 conn.commit() 84 conn.close() 85 86 records_list = [] 87 records_need_save = False 88 89 print('>> sqlite insert records: ' + 90 str(insert_count * 100) + ' - ' + str(insert_count * 100 + 99)) 91 insert_count = insert_count + 1 92 time.sleep(0.4) 93 94 print('>> sqlite test finish...') 95 96 if __name__ == '__main__': 97 main()
运行结果
运行test_sqlite.py，显示总共插入10次（每次有100条记录）
pi@raspberrypi:~ $ python test_sqlite.py
查看数据库是否存在，显示新生成一个数据库：NULL_999_2020-04-23_160903.db3
pi@raspberrypi:~ $ cd /home/pi/project/data/tmp/
pi@raspberrypi:~/project/data/tmp $ ls -lh
在同一无线网络系啊，使用FileZilla软件登陆树莓派，把数据库文件从树莓派中拷贝到电脑上。
主机：raspberrypi.local，用户名：pi，密码：raspberry，端口：22，点击 快速连接
远程站点中，进入/home/pi/project/data/tmp目录
将下面的数据库文件复制到本地站点的一个目录下（双击或者拖拽都可以）：
用SQLite Expert打开这个文件，点击records表，可以看到里面的数据：
扩展内容：存储逻辑
>> 存储位置
创建3个文件夹：<临时>文件夹、<待上传>文件夹、<已上传>文件夹
| 临时 | 新收到通讯数据时，在该文件夹下建立数据库，持续写入数据； 持续写入1小时/持续1分钟未收到新数据，关闭该数据库，移动至<待上传>； 上电后首先将该目录下的数据库文件移动至<待上传> |
| 待上传 | 数据上传服务程序每隔30s扫描一次该目录 如果有文件，则开始上传第1个文件，上传完成后将该文件移动至<已上传> 如果采集装置离线，则该目录为实际存储目录 |
| 已上传 | 如果采集装置在线，则该目录为实际存储目录 |
>> 批量写入
为了提高写入效率，可以把记录先放在内存里，攒够N条记录（100条）后再一次性写入到数据库中。当然如果较长的一段时间（10s）没有收到新的数据帧，就会认为被监听的设备掉线，这时把已经有的记录写入到数据库中。
>> 定时打包
连续采集较长时间（1个小时），或者连续一段时间（≥1分钟）没有收到新的数据帧，就结束往当前数据库写入记录，相当于把这段时间的数据打包成一个数据库，便于后续使用。把这个数据库移动到待上传的文件夹中，然后再新建一个数据库继续采集。