树莓派ZeroW的Python中使用SQLite数据库

>> [目录] 数据远程采集 Step by Step

SQLite数据库（www.sqlite.org）是一个轻型的数据库引擎，应用非常广泛。我们使用SQLite来存储采集到的通讯数据，一帧通讯原始数据加上时间戳后作为一条记录存入数据库中。在这个项目中，只需要用到数据库的存储功能，不需要查询之类的（到本地上位机上解析数据时才会用到），所以下面也只介绍存储相关的部分，其他功能可以参考这个网址：

https://www.runoob.com/sqlite/sqlite-python.html

程序说明

---

用下面这些函数就可以插入记录完成存储，还需要解决的是SQL语句：sql_create_table、sql_insert_record，以及记录集合：records_list。

import sqlite3

# 连接到SQLite数据库：DB_NAME，如果不存在就创建
conn = sqlite3.connect(DB_NAME)
# 创建一个游标Cursor，可以通过Cursor来执行SQL语句
cursor = conn.cursor()
# 创建记录表：records
cursor.execute(sql_create_table)
# 插入多条记录
cursor.executemany(sql_insert_record, records_list)
# 关闭Cursor
cursor.close()
# 提交事务
conn.commit()
# 关闭数据库连接
conn.close()

>> SQL语句：sql_create_table、sql_insert_record

sql_create_table

在数据库中建一个表records，表中记录的结构为：单条记录结构：id（自增），时间戳（text），从数据库创建开始的累计时间（real），通讯原始数据（400个，int）

sql_insert_record

插入记录时只需要：时间戳、累计时间、通讯数据*400

def init_sql_sentences(param_num):
  global sql_create_table
  global sql_insert_record

  # sql_create_table
  sql_create_table = 'create table if not exists records('
  sql_create_table += 'id integer primary key autoincrement, '
  sql_create_table += 'datetime text, '
  sql_create_table += 'runningtime real, '
  for i in range(0, param_num - 1):
    sql_create_table += 'param' + str(i) + ' int, '
  sql_create_table += 'param' + str(param_num - 1) + ' int)'

  # sql_insert_record
  sql_insert_record = 'insert into records values (null, ?, ?, '
  for i in range(0, param_num - 1):
    sql_insert_record += '?, '
  sql_insert_record += '?)'

>> 记录集合：records_list

按照记录的格式，将时间戳、通讯数据等存入单条记录record_buff[]，然后再把record_buff[]添加到到记录集records_list[]，这里使用了tuple，把record_buff[]转换成元组，这样就负荷cursor.executemany()函数要求的格式了。

records_list = []
record_buff = []
record_buff.append('')
record_buff.append(0.0)
for i in range(0, DB_PARAMS_LENGTH):
  record_buff.append(0)

####
record_buff[0] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
record_buff[1] = round((time.time() - start_time), 1)
for i in range(0, DB_PARAMS_LENGTH):
  record_buff[i + 2] = comn_data[i]
  comn_data[i] = 0
records_list.append(tuple(record_buff[:]))

创建测试文件test_sqlite.py

用模拟通讯数据，生成100条记录，插入数据库，等待0.4s。循环10次以后退出。

pi@raspberrypi:~ $ sudo nano test_sqlite.py

# -*- coding:utf-8 -*-
import time
import sqlite3
import os
from datetime import datetime

DEFAULT_DIR         = '/home/pi/project/'
DB_DIR              = '/home/pi/project/data/'
DB_TMP_DIR          = '/home/pi/project/data/tmp/'
DB_UPDATING_DIR     = '/home/pi/project/data/updating/'
DB_UPDATED_DIR      = '/home/pi/project/data/updated/'

DB_NAME             = ''

PRODUCT_TYPE        = 'NULL'
DEVICE_ID           = 999
DB_PARAMS_LENGTH    = 400

sql_create_table    = ''
sql_insert_record   = ''

def init_sql_sentences(param_num):
  ####

def main():
  global PRODUCT_TYPE
  global DEVICE_ID
  global DB_PARAMS_LENGTH

  record_count = 0
  insert_count = 0

  record_buff = []
  records_list = []
  comn_data = [0 for i in range(DB_PARAMS_LENGTH)]

  record_buff.append('')
  record_buff.append(0.0)
  for i in range(0, DB_PARAMS_LENGTH):
    record_buff.append(0)

  if not os.path.exists(DEFAULT_DIR):
    os.mkdir(DEFAULT_DIR)
  if not os.path.exists(DB_DIR):
    os.mkdir(DB_DIR)
  if not os.path.exists(DB_TMP_DIR):
    os.mkdir(DB_TMP_DIR)
  if not os.path.exists(DB_UPDATING_DIR):
    os.mkdir(DB_UPDATING_DIR)
  if not os.path.exists(DB_UPDATED_DIR):
    os.mkdir(DB_UPDATED_DIR)

  init_sql_sentences(DB_PARAMS_LENGTH)

  DB_NAME = DB_TMP_DIR+PRODUCT_TYPE + '_' + str(DEVICE_ID).rjust(3,'0') + '_' 
            + time.strftime('%Y-%m-%d_%H%M%S', time.localtime()) + '.db3'

  print('>> sqlite test start...')
  print(DB_NAME)

  start_time = time.time()
  while insert_count < 10:
    # set fake data
    for i in range(0, DB_PARAMS_LENGTH):
      comn_data[i] = i
    comn_data[0] = record_count

    record_buff[0] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
    record_buff[1] = round((time.time() - start_time), 1)
    for i in range(0, DB_PARAMS_LENGTH):
      record_buff[i + 2] = comn_data[i]
      comn_data[i] = 0

    records_list.append(tuple(record_buff[:]))
    record_count = record_count + 1

    if((record_count % 100) == 0):
      conn = sqlite3.connect(DB_NAME)
      cursor = conn.cursor()
      cursor.execute(sql_create_table)
      cursor.executemany(sql_insert_record, records_list)
      cursor.close()
      conn.commit()
      conn.close()

      records_list = []
      records_need_save = False

      print('>> sqlite insert records: ' + 
             str(insert_count * 100) + ' - ' + str(insert_count * 100 + 99))
      insert_count = insert_count + 1
      time.sleep(0.4)

  print('>> sqlite test finish...')

if __name__ == '__main__':
  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分钟）没有收到新的数据帧，就结束往当前数据库写入记录，相当于把这段时间的数据打包成一个数据库，便于后续使用。把这个数据库移动到待上传的文件夹中，然后再新建一个数据库继续采集。