1. 树莓派 DHT22 温湿度落库为什么我最后选了 SQLite树莓派上用 DHT22 采集温湿度再把数据写进本地 SQLite这套链路听起来简单但真正跑起来会碰到几个很实际的问题传感器读数偶尔校验失败、时间戳到底用采集时刻还是入库时刻、数据库文件放哪、后面怎么查。我试过把数据先写 CSV结果查一周的平均温湿度得自己写脚本遍历后来换成 SQLite一条AVG就出来了。这篇聚焦的是「树莓派 DHT22 本地 SQLite」的完整可跑通链路同时把 TaoToken 统一 Key 接入的config.toml配置骨架和 Cline 侧接入示例给出来。适合谁手上有一块树莓派、一个 DHT22AM2302、想先把数据稳稳落在本地、后面再考虑上云或做分析的开发者。核心检索词就三个树莓派、SQLite、DHT22 温湿度数据。为什么是 SQLite 而不是 MySQL 或 InfluxDB单个 DHT22 几分钟采一条写入频率极低SQLite 是文件型数据库Python 标准库自带sqlite3零安装零服务整个库就是一个.db文件备份复制都简单。MySQL 要装服务、配用户权限对单传感器场景是杀鸡用牛刀InfluxDB 适合海量高并发时序数据这里完全发挥不出优势。数据量大了、要多节点汇总了再换也不迟。下面从建表、采集写入、验证查询一路走完中间穿插 TaoToken 的 Key 配置骨架方便你后面接 AI 辅助编码或做数据分析时统一管理凭证。2. TaoToken 前置统一 Key 与 config.toml 骨架在写采集脚本之前先把凭证管理这件事理清楚。树莓派上跑的东西一多API Key 散落在各个脚本里很容易乱。TaoToken 提供统一 Key 和 API 通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它的作用是让你用一个 Key 走统一通道不用在每个工具里分别配不同厂商的凭证。我建议在树莓派上建一个独立目录放配置比如~/.config/taotoken/config.toml权限设成600别让同机其他用户读到。下面这份config.toml骨架可以直接复制把api_key换成你在控制台生成的即可# ~/.config/taotoken/config.toml # TaoToken 统一 Key 配置骨架 # 控制台生成 Key: https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite [default] # 统一 API Key所有工具共用这一份 api_key sk-替换成你的Key # API 基础地址不要带末尾斜杠 base_url https://taotoken.net/api # 请求超时秒树莓派网络一般给宽一点 timeout 60 # 模型对话场景做数据分析、问 SQL 写法时用 [chat] model claude-sonnet-4-20250514 max_tokens 4096 # 长期编码 / Agent 场景Cline 这类工具走这个 [coding] model claude-sonnet-4-20250514 max_tokens 8192生成 Key 的入口在控制台的 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。拿到 Key 后先别急着写进脚本用环境变量兜一层更安全# 写入 shell 配置避免明文散落在代码里 echo export TAOTOKEN_API_KEYsk-你的Key ~/.bashrc source ~/.bashrc # 验证变量已生效 echo $TAOTOKEN_API_KEY | head -c 8这样脚本里读os.environ[TAOTOKEN_API_KEY]就行config.toml里也可以留空由环境变量覆盖。Cline 侧接入时在设置里选 Anthropic 兼容模式Base URL 填https://taotoken.net/apiAPI Key 填同一个模型名按config.toml里的写。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到字段对不上可以对照查。注意config.toml里不要提交到 Git加进.gitignore。树莓派如果多人用chmod 600是底线。3. 可复制配置建表语句与采集写入脚本先把数据库模块写好。核心就三件事建表、写入、查询。时间戳用 TEXT 存 ISO 格式YYYY-MM-DD HH:MM:SS精确到秒排序和范围查询都方便。温度湿度用 REAL。建表时顺手建个时间索引后面按时间段查会快很多。#!/usr/bin/env python3 # dht22_db.py — DHT22 温湿度 SQLite 存储模块 import sqlite3 import os from datetime import datetime CREATE_TABLE_SQL CREATE TABLE IF NOT EXISTS dht22 ( id INTEGER PRIMARY KEY AUTOINCREMENT, temperature REAL NOT NULL, humidity REAL NOT NULL, recorded_at TEXT NOT NULL ) CREATE_INDEX_SQL CREATE INDEX IF NOT EXISTS idx_dht22_recorded_at ON dht22 (recorded_at DESC) INSERT_SQL INSERT INTO dht22 (temperature, humidity, recorded_at) VALUES (?, ?, ?) class Dht22Database: def __init__(self, db_pathNone): if db_path is None: script_dir os.path.dirname(os.path.abspath(__file__)) db_path os.path.join(script_dir, sensor_data.db) self.db_path db_path db_dir os.path.dirname(db_path) if db_dir and not os.path.exists(db_dir): os.makedirs(db_dir, exist_okTrue) self._init_tables() def _get_connection(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row # WAL 模式提升并发写入性能 conn.execute(PRAGMA journal_modeWAL) return conn def _init_tables(self): conn self._get_connection() try: cur conn.execute( SELECT name FROM sqlite_master WHERE typetable AND namedht22 ) if cur.fetchone() is None: conn.execute(CREATE_TABLE_SQL) conn.execute(CREATE_INDEX_SQL) conn.commit() print(f[数据库] 首次初始化 {self.db_path}) finally: conn.close() def insert_now(self, temperature, humidity): 用当前时间写入一条记录返回时间戳 now datetime.now().strftime(%Y-%m-%d %H:%M:%S) conn self._get_connection() try: conn.execute(INSERT_SQL, (temperature, humidity, now)) conn.commit() finally: conn.close() return now def query_recent(self, limit10): conn self._get_connection() try: cur conn.execute( SELECT id, temperature, humidity, recorded_at FROM dht22 ORDER BY recorded_at DESC LIMIT ?, (limit,), ) return [dict(r) for r in cur.fetchall()] finally: conn.close() def avg_range(self, start, end): conn self._get_connection() try: cur conn.execute( SELECT AVG(temperature), AVG(humidity) FROM dht22 WHERE recorded_at ? AND recorded_at ?, (start, end), ) row cur.fetchone() return { avg_temperature: round(row[0], 2) if row[0] else None, avg_humidity: round(row[1], 2) if row[1] else None, } finally: conn.close()采集脚本负责读 DHT22 并调用上面的模块落库。DHT22 用 Adafruit 的 CircuitPython 库内核级脉冲计数比纯 Python bit-banging 稳。读失败要重试连续失败就跳过本轮别让一条坏数据把整个循环卡死。#!/usr/bin/env python3 # dht22_collect.py — 读取 DHT22 并写入 SQLite import time import sys import board import adafruit_dht from dht22_db import Dht22Database DHT22_PIN board.D4 # GPIO4物理引脚 7 MAX_RETRIES 5 RETRY_DELAY 2.0 REPORT_INTERVAL 300 # 5 分钟采一次 def read_sensor(sensor): for attempt in range(1, MAX_RETRIES 1): try: t sensor.temperature h sensor.humidity if t is None or h is None: raise RuntimeError(读数为空) if not (-40.0 t 80.0) or not (0.0 h 100.0): raise RuntimeError(f读数越界: {t}C {h}%) return {temperature: round(t, 2), humidity: round(h, 2)} except RuntimeError as e: print(f[重试 {attempt}/{MAX_RETRIES}] {e}) if attempt MAX_RETRIES: time.sleep(RETRY_DELAY) return None def main(): sensor adafruit_dht.DHT22(DHT22_PIN, use_pulseioTrue) db Dht22Database() print(f[启动] 数据库: {db.db_path}) try: while True: data read_sensor(sensor) if data is None: print([跳过] 本轮读取失败) else: ts db.insert_now(data[temperature], data[humidity]) print(f[写入] {ts} | {data[temperature]}C | {data[humidity]}%) time.sleep(REPORT_INTERVAL) except KeyboardInterrupt: print(\n[退出] 收到中断信号) finally: sensor.exit() print([清理] 传感器资源已释放) if __name__ __main__: main()依赖安装就两条命令树莓派上先确认 I2C/GPIO 权限正常pip install adafruit-circuitpython-dht # 如果提示缺少底层库 sudo apt-get install -y libgpiod24. 验证请求一次采集写入与查询确认脚本写完别急着挂后台先手动跑一次确认从传感器读数到落库整条链路通。第一步单独测数据库模块不接传感器也能验证建表和写入逻辑python3 -c from dht22_db import Dht22Database db Dht22Database() ts db.insert_now(25.8, 67.9) print(写入时间戳:, ts) print(最近记录:, db.query_recent(3)) 预期输出类似[数据库] 首次初始化 /home/pi/sensor/sensor_data.db 写入时间戳: 2026-06-14 01:55:00 最近记录: [{id: 1, temperature: 25.8, humidity: 67.9, recorded_at: 2026-06-14 01:55:00}]第二步接上 DHT22 跑采集脚本观察一轮写入python3 dht22_collect.py正常会看到[写入] 2026-06-14 02:00:00 | 25.6C | 68.2%这样的行。按 CtrlC 退出会打印清理信息。第三步用查询脚本确认数据真的在库里并且聚合查询能跑python3 -c from dht22_db import Dht22Database db Dht22Database() print(总数:, len(db.query_recent(9999))) print(平均:, db.avg_range(2026-06-14 00:00:00, 2026-06-14 23:59:59)) 如果平均温湿度能正常返回说明建表、写入、索引、聚合查询全通了。这时候再考虑用systemd或nohup把采集脚本挂后台长期跑。想用 AI 辅助写查询语句或分析数据可以在模型对话入口 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 里把表结构贴进去让它帮你生成 SQL长期在树莓派上做编码和 Agent 任务走 Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。5. 本篇常见错排查报错RuntimeError: Timed out waiting for PulseIn messageDHT22 没响应。先查接线VCC 接 3.3VDATA 接 GPIO4GND 接 GNDDATA 和 VCC 之间最好加个 4.7k 到 10k 的上拉电阻。树莓派上跑脚本要确保有 GPIO 访问权限必要时sudo跑一次确认是不是权限问题。报错ModuleNotFoundError: No module named boardAdafruit 库没装全。pip install adafruit-circuitpython-dht之后还要确认libgpiod2在系统里sudo apt-get install -y libgpiod2。数据库文件找不到或写不进去Dht22Database()默认把.db放在脚本同目录。如果脚本在只读目录或权限受限目录会静默失败或报unable to open database file。显式传路径比如Dht22Database(/home/pi/sensor/sensor_data.db)并确认目录可写。时间戳排序不对检查写入时用的是不是YYYY-MM-DD HH:MM:SS格式。如果混入了其他格式字符串排序会乱。统一用datetime.now().strftime(%Y-%m-%d %H:%M:%S)。连续读取失败后脚本卡死read_sensor里重试次数和间隔要设合理别无限重试。连续失败就跳过本轮等下一个采集周期。另外sensor.exit()一定要在finally里调用否则下次启动可能因为 GPIO 被占用而失败。Cline 接入报 401 或模型不存在检查config.toml里base_url是不是https://taotoken.net/api末尾不要带斜杠API Key 有没有多余空格模型名是否和控制台里可用的一致。接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里有字段对照表。6. 接入与排障入口树莓派本地 SQLite 这条链路跑通之后后面无非是加查询脚本、做可视化、或者把数据同步到别处。凭证这块统一走 TaoToken一个 Key 管住所有工具省得每个脚本里塞一份。需要生成或轮换 Key 去 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入细节和字段说明看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。遇到传感器读数和数据库写入的问题先把采集脚本单独跑一轮确认是硬件层还是存储层再对症处理。