从数据到图表：树莓派Flask-SQLite物联网仪表盘构建

从DHT22传感器采集实时数据，存入轻量级SQLite数据库，通过Flask构建Web仪表盘可视化展示。

1. 引言

上一教程树莓派Flask Web服务器中我们学会了通过Web前端暴露传感器数据。下一逻辑步骤是持久化读数以便后续分析。

2. 物料清单

• 树莓派3 – $32.00
• DHT22温湿度传感器 – $9.95
• 4.7 kΩ上拉电阻

3. 安装SQLite

SQLite是单板计算机理想数据库引擎：无服务器、轻量、全开源。安装命令：

sudo apt-get install sqlite3

创建工作目录和数据库文件：

mkdir Sensors_Database
cd Sensors_Database
sqlite3 sensorsData.db

SQLite提示符出现后，输入.quit返回Shell。

4. 创建DHT数据表

表DHT_data存储时间戳、温度、湿度。

sqlite3 sensorsData.db
BEGIN;
CREATE TABLE DHT_data (timestamp DATETIME, temp NUMERIC, hum NUMERIC);
COMMIT;

或用Python：

import sqlite3
con = sqlite3.connect('sensorsData.db')
with con:
    cur = con.cursor()
    cur.execute('DROP TABLE IF EXISTS DHT_data')
    cur.execute('CREATE TABLE DHT_data(timestamp DATETIME, temp NUMERIC, hum NUMERIC)')

5. 插入示例数据

SQLite手动插入：

INSERT INTO DHT_data VALUES(datetime('now'), 20.5, 30);

Python批量插入：

with con:
    cur.execute("INSERT INTO DHT_data VALUES(datetime('now'), 20.5, 30)")
    cur.execute("INSERT INTO DHT_data VALUES(datetime('now'), 25.8, 40)")
    cur.execute("INSERT INTO DHT_data VALUES(datetime('now'), 30.3, 50)")

6. DHT22接线

VCC接3.3V，Data接GPIO 16，GND接地，VCC与Data间接4.7kΩ上拉电阻。

7. 安装Adafruit DHT库

从GitHub下载安装：

sudo python3 setup.py install

8. 读取与记录数据

最小脚本读取传感器、四舍五入、写入数据库：

import time, sqlite3, Adafruit_DHT
DB_NAME = 'sensorsData.db'
SAMPLE_FREQ = 120
DHT_SENSOR = Adafruit_DHT.DHT22
DHT_PIN = 16

def read_and_log():
    humidity, temperature = Adafruit_DHT.read_retry(DHT_SENSOR, DHT_PIN)
    if humidity is not None and temperature is not None:
        humidity = round(humidity)
        temperature = round(temperature, 1)
        conn = sqlite3.connect(DB_NAME)
        cur = conn.cursor()
        cur.execute("INSERT INTO DHT_data VALUES(datetime('now'), ?, ?)", (temperature, humidity))
        conn.commit()
        conn.close()

for _ in range(3):
    read_and_log()
    time.sleep(SAMPLE_FREQ)

9. 持续记录

实时记录器，每分钟采样，在专用终端运行：

import time, sqlite3, Adafruit_DHT
DB = 'sensorsData.db'
SAMPLE_FREQ = 60
while True:
    humidity, temperature = Adafruit_DHT.read_retry(Adafruit_DHT.DHT22, 16)
    if humidity is not None and temperature is not None:
        conn = sqlite3.connect(DB)
        cur = conn.cursor()
        cur.execute("INSERT INTO DHT_data VALUES(datetime('now'), ?, ?)", (round(temperature,1), round(humidity)))
        conn.commit()
        conn.close()
    time.sleep(SAMPLE_FREQ)

10. 数据库查询

常用查询：

• 所有行：SELECT * FROM DHT_data
• 最新条目：SELECT * FROM DHT_data ORDER BY timestamp DESC LIMIT 1
• 温度>30°C：SELECT * FROM DHT_data WHERE temp > 30
• 湿度=29%：SELECT * FROM DHT_data WHERE hum = 29

11. Flask Web服务器

安装Flask：sudo apt-get install python3-flask。创建项目结构：

mkdir dhtWebServer
cd dhtWebServer
mkdir templates static

appDhtWebServer.py：

from flask import Flask, render_template
import sqlite3
app = Flask(__name__)

def fetch_latest():
    conn = sqlite3.connect('../sensorsData.db')
    cur = conn.cursor()
    cur.execute('SELECT * FROM DHT_data ORDER BY timestamp DESC LIMIT 1')
    row = cur.fetchone()
    conn.close()
    return row

@app.route('/')
def index():
    time, temp, hum = fetch_latest()
    return render_template('index.html', time=time, temp=temp, hum=hum)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=80, debug=False)

模板templates/index.html：

<!DOCTYPE html>
<html>
<head>
<title>DHT传感器仪表盘</title>
<link rel='stylesheet' href='../static/style.css'>
</head>
<body>
<h1>DHT传感器数据</h1>
<p>温度: {{ temp }} °C</p>
<p>湿度: {{ hum }} %</p>
<p>最新读数: {{ time }}</p>
<a href='/' class='button'>刷新</a>
</body>
</html>

12. 前端增强：JustGage仪表盘

下载JustGage，将justgage.js和raphael-2.1.4.min.js置于static。更新模板渲染动态仪表：

<div id='g1'></div>
<div id='g2'></div>
<script src='../static/raphael-2.1.4.min.js'></script>
<script src='../static/justgage.js'></script>
<script>
new JustGage({id:'g1', value:{{ temp }}, min:-10, max:50, title:'温度', label:'°C'});
new JustGage({id:'g2', value:{{ hum }}, min:0, max:100, title:'湿度', label:'%'});
</script>

13. Matplotlib历史趋势图

安装Matplotlib：sudo apt-get install python3-matplotlib。新建Flask应用appDhtWebHist.py动态生成PNG图片。

from flask import Flask, render_template, make_response
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas
from matplotlib.figure import Figure
import sqlite3, io
app = Flask(__name__)
DB = '../sensorsData.db'

def get_latest():
    conn = sqlite3.connect(DB)
    cur = conn.cursor()
    cur.execute('SELECT * FROM DHT_data ORDER BY timestamp DESC LIMIT 1')
    row = cur.fetchone()
    conn.close()
    return row

def get_history(samples):
    conn = sqlite3.connect(DB)
    cur = conn.cursor()
    cur.execute('SELECT * FROM DHT_data ORDER BY timestamp DESC LIMIT ?', (samples,))
    rows = cur.fetchall()
    conn.close()
    rows.reverse()
    times, temps, hums = zip(*rows)
    return times, temps, hums

@app.route('/')
def index():
    time, temp, hum = get_latest()
    return render_template('index.html', time=time, temp=temp, hum=hum, samples=100)

@app.route('/plot/temp')
def plot_temp():
    times, temps, _ = get_history(100)
    fig = Figure()
    ax = fig.add_subplot(1,1,1)
    ax.plot(range(len(temps)), temps)
    ax.set_title('温度 (°C)')
    canvas = FigureCanvas(fig)
    output = io.BytesIO()
    canvas.print_png(output)
    return make_response(output.getvalue(), 200, {'Content-Type':'image/png'})

@app.route('/plot/hum')
def plot_hum():
    _, _, hums = get_history(100)
    fig = Figure()
    ax = fig.add_subplot(1,1,1)
    ax.plot(range(len(hums)), hums)
    ax.set_title('湿度 (%)')
    canvas = FigureCanvas(fig)
    output = io.BytesIO()
    canvas.print_png(output)
    return make_response(output.getvalue(), 200, {'Content-Type':'image/png'})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=80, debug=False)

模板templates/index.html包含两个<img>标签请求PNG路由。

14. 结语

这个端到端项目演示了如何将廉价传感器转化为数据记录与监控平台：采集、持久化、查询、可视化，全部通过本地Flask Web服务器完成。所有源码见GitHub，更多示例访问MJRoBot.org。