YUWEN.LOG
嵌入式

Journal Article

ESP32物联网全栈方案：MQTT多传感器采集 + FastAPI + Tauri桌面应用

ESP32物联网全栈项目：BH1750/SHT35/MQ-2三传感器采集，Python FastAPI内嵌MQTT代理，SQLAlchemy持久化，React+ECharts+SVG地图拓扑可视化，JWT鉴权后台管理。

21 min6 views
PythonMySQLC++编程基础Java

ESP32物联网全栈方案：MQTT多传感器采集 + FastAPI + Tauri桌面应用

项目地址：GitHub - esp32bishe

系统全景

code
┌──────────────────────────────────────────────────────┐
│   Web 前端 (my-app)          │  Tauri 桌面应用 (sensorapp) │
│   React 19 + Vite + ECharts │  React + Rust + Tauri v2    │
│   15个功能模块               │  原生窗口 + 系统级能力       │
├──────────────────────────────────────────────────────┤
│              Python FastAPI 后端                      │
│   内嵌 MQTT Broker (amqtt) + MQTT 订阅 (paho-mqtt)   │
│   SQLAlchemy + PyMySQL + JWT + APScheduler           │
├──────────────────────────────────────────────────────┤
│   ESP32 设备端 (C++ / ESP-IDF / FreeRTOS)            │
│   BH1750 + SHT35 + MQ-2 + SD 环形缓存                │
└──────────────────────────────────────────────────────┘

第一层：ESP32 设备端

传感器与硬件

传感器总线引脚数据
BH1750I2C0SDA=GPIO33, SCL=GPIO32光照 (lux)
SHT35I2C1SDA=GPIO16, SCL=GPIO17温度 + 湿度
MQ-2ADCGPIO34/35烟雾/VOC
SD卡SPI—离线环形缓存

GPIO4 控制 I2C0 电源，降低功耗。采样周期可通过 MQTT 远程配置（默认 CONFIG_SENSOR_TASK_PERIOD_MS）。

FreeRTOS 四任务架构

cpp
// 1. 光照采集：独立 I2C0 周期采样，发送到 lux_queue
xTaskCreate(sensor_lux_task,     "lux",     4096, NULL, 5, NULL);

// 2. 温湿度采集：独立 I2C1 周期采样，发送到 climate_queue  
xTaskCreate(sensor_climate_task, "climate", 4096, NULL, 5, NULL);

// 3. 合流上报：从队列收数据 → JSON 序列化 → MQTT publish
//    同时处理 backfill 补发确认 (BIT_BACKFILL_PUBLISHED)
xTaskCreate(app_logic_task, "logic", 8192, NULL, 3, NULL);

// 4. SD日志：断网时写环形缓存，定期输出最近5条日志
xTaskCreate(sd_log_task, "sdlog", 4096, NULL, 1, NULL);

MQTT 双工通信

发布主题（设备 → Broker）：

Topic内容说明
esp32bishe/sensor/lux光照值独立上报
esp32bishe/sensor/climate温度+湿度独立上报
esp32bishe/sensor/gas烟雾/VOCMQ-2 独立通道
esp32bishe/sensor/all全传感器合并主数据通道

订阅主题（Broker → 设备）：

模式Topic
控制device/{id}/control
心跳device/{id}/hello
补发device/{id}/backfill

SD 卡离线 + backfill 补发协议

断网时数据写入 SD 环形缓冲区，恢复后补发。补发使用自定义二进制协议：

cpp
static constexpr uint32_t BACKFILL_MAGIC = 0x314C4642u;  // "BFL1" LE
static constexpr uint16_t BACKFILL_VERSION = 1;
static constexpr size_t   BACKFILL_BATCH_MAX_RECORDS = 24;

// delta-pack + Base64 → JSON → MQTT publish
// 服务器收到后解包恢复时间序列

补发流程：backfill publish → 等待 MQTT_EVENT_PUBLISHED（ACK，5s超时）→ 设 BIT_BACKFILL_PUBLISHED

第二层：Python FastAPI 后端

依赖一览

text
fastapi>=0.115.0          # Web 框架
uvicorn[standard]>=0.32.0 # ASGI 服务器
amqtt>=0.2.0              # 内嵌 MQTT Broker（无需 Mosquitto）
paho-mqtt>=2.0.0          # MQTT 订阅客户端
sqlalchemy>=2.0.0         # ORM
pymysql>=1.1.0            # MySQL 驱动
apscheduler>=3.10.0       # 定时任务（统计/告警/清理）
python-jose>=3.3.0        # JWT 鉴权
python-multipart>=0.0.9   # 表单解析
pydantic-settings>=2.0.0  # 配置管理

启动流程（lifespan）

python
# server/app/main.py
@asynccontextmanager
async def lifespan(app: FastAPI):
    init_engine()                          # SQLAlchemy 引擎
    create_tables()                        # 自动建表
    seed_if_needed()                       # 种子数据 + 全局默认设置
    ensure_global_settings()
    
    # 1. 内嵌 MQTT Broker（amqtt，端口 1883）
    broker_task = asyncio.create_task(run_broker_until_cancelled())
    
    # 2. MQTT 订阅线程（paho-mqtt）
    subscriber = threading.Thread(target=run_subscriber)
    subscriber.start()
    
    # 3. 定时调度器（APScheduler）
    start_scheduler()
    
    notify_ws_start(app)  # WebSocket 通知（可选）
    
    yield
    
    # 关闭
    shutdown_scheduler()
    broker_task.cancel()

关键优势：amqtt 在 FastAPI 进程内启动 Broker，开发/部署无需额外安装 Mosquitto。

API 路由全貌

code
/api/v1/
├── auth           ← JWT 登录/注册（python-jose）
├── health         ← 健康检查
├── sensors        ← 实时传感器数据（SensorStore 内存缓存）
├── zones          ← 区域 CRUD + 阈值 JSON 配置
│   └── {zone_id}/detail  ← 区域内设备、最新值、最近告警
├── settings       ← 全局设置（高/低温阈值、上传间隔）
├── devices        ← 设备元数据（zone_id、label、sensor_enabled）
├── history        ← 历史数据查询（时间范围 + 分区/设备过滤）
├── export/csv     ← CSV 导出（支持大文件流式输出）
└── alerts         ← 告警日志 + 模拟告警（warn/critical）

数据持久化

python
# server/app/services/persistence.py
# paho-mqtt on_message 回调 → SQLAlchemy → MySQL
def dispatch_mqtt_message(topic: str, payload: str):
    data = json.loads(payload)
    with SessionLocal() as session:
        record = SensorData(
            device_id=data["device_id"],
            temperature=data.get("temperature"),
            humidity=data.get("humidity"),
            light=data.get("lux"),
            voc=data.get("voc"),
            timestamp=...
        )
        session.add(record)
        session.commit()
        sensor_store.update(device_id, values)  # 同时更新内存缓存

数据库表结构

sql
-- 时序数据
CREATE TABLE sensor_data (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    device_id VARCHAR(64),
    temperature FLOAT, humidity FLOAT,
    light FLOAT, voc FLOAT,
    timestamp DATETIME, INDEX idx_ts (timestamp)
);

-- 设备管理
CREATE TABLE devices (
    id BIGINT PRIMARY KEY, device_id VARCHAR(64) UNIQUE,
    zone_id VARCHAR(64), label VARCHAR(128),
    sensor_enabled BOOLEAN DEFAULT TRUE,
    upload_interval_ms INT, coordinates VARCHAR(128)
);

-- 告警日志
CREATE TABLE alert_logs (
    id BIGINT PRIMARY KEY, device_id VARCHAR(64),
    alert_type VARCHAR(32), level ENUM('warn','critical'),
    metric VARCHAR(32), message TEXT, created_at DATETIME
);

第三层：React Web 前端（my-app）

完整模块树（15+ 个模块）

code
src/
├── pages/                    # 页面路由
│   ├── LoginPage.tsx         # JWT 登录（含 OAuth 扩展点）
│   ├── DashboardPage.tsx     # 主面板：地图 + 曲线
│   ├── ZonesDevicesPage.tsx  # 区域/设备管理（CRUD + 阈值）
│   └── SettingsPage.tsx      # 全局设置
│
├── components/               # 可复用组件
│   ├── ZoneMap.tsx           # SVG 地图拓扑（绿/黄/红状态着色）
│   ├── DateRangePicker.tsx   # 时间范围选择
│   ├── ThresholdConfigForm.tsx  # 阈值配置表单（JSON 编辑器）
│   ├── StatusBadge.tsx       # 状态标签（含单元测试）
│   ├── ErrorBoundary.tsx     # React 错误边界
│   ├── LoadingSpinner.tsx    # 加载指示器
│   └── EmptyState.tsx        # 空数据兜底 UI
│
├── api.ts                    # ★ 核心 API 层（495 行）
├── store/                    # 状态管理
├── hooks/                    # 自定义 Hooks
├── layout/AdminLayout.tsx    # 管理后台布局
├── i18n/                     # 国际化
├── theme/                    # 主题系统
├── security/                 # 安全模块
├── performance/              # 性能优化
├── notifications/            # 通知系统
├── search/                   # 搜索功能
├── logger/                   # 前端日志
├── charts/                   # 图表配置
├── dev-tools/                # 开发工具
├── test/                     # 测试
└── styles/                   # 样式

api.ts 核心设计

ApiCache — 客户端缓存层：

typescript
class ApiCache {
  private cache = new Map<string, CacheEntry<unknown>>()
  private defaultTTL = 5 * 60 * 1000  // 默认 5 分钟

  set<T>(key: string, data: T, ttl?: number): void
  get<T>(key: string): T | null
  delete(key: string): void
  cleanup(): void  // 每 60 秒自动清除过期条目
}

fetchWithRetry — 指数退避重试：

typescript
async function fetchWithRetry(url, options, retries = 3): Promise<Response> {
  // 5xx 服务器错误 → 重试（延迟 1s, 2s, 3s）
  // 401 → 不重试，直接返回
  // 网络错误 → 重试
}

自动 401 拦截 → 登录重定向：

typescript
if (r.status === 401) {
  setToken(null)
  if (!window.location.pathname.includes('/login')) {
    window.location.assign('/login')
  }
}

登录多策略兼容：

typescript
// 依次尝试 3 种 Content-Type，兼容不同后端配置
const attempts = [
  () => fetch(..., JSON),                               // application/json
  () => fetch(..., 'application/x-www-form-urlencoded'),// 表单
  () => fetch(..., '?username=...'),                     // Query 参数
]

CSV 流式导出 + 进度条：

typescript
async function downloadCsv(params, onProgress) {
  const reader = r.body?.getReader()
  // ReadableStream 分块读取
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    chunks.push(value)
    receivedLength += value.length
    onProgress(Math.round((receivedLength / total) * 100))
  }
}

第四层：Tauri 桌面应用（sensorApp）

json
// sensorApp/sensorapp/package.json
{
  "dependencies": {
    "react": "^19.1.0",
    "react-router-dom": "^7.13.1",
    "echarts": "^6.0.0",
    "@tauri-apps/api": "^2",
    "@tauri-apps/plugin-opener": "^2"
  },
  "scripts": {
    "tauri": "tauri"
  }
}

Tauri v2 将 Web 前端打包为原生桌面窗口，通过 Rust 后端获得系统级能力（文件系统、通知、进程管理），适合在 Windows/Linux 上长期运行作为监控面板。

根目录工具脚本

项目根目录还包含大量 Python/Node.js 数据处理脚本：

脚本用途
format_paper.py论文排版（自动生成 Word 文档）
create_tables.py数据表格生成
convert_md_tables_to_sanxian.pyMarkdown 表格 → 三线表（学术格式）
mdtable_to_sanxian.py三线表转换核心
create_en_paper.js英文论文生成（Node.js）
pack_doc.py / pack_docx.py文档打包
replace_tables.py表格替换工具
analyze_headings.py论文标题结构分析

完整数据流链路

code
BH1750/SHT35/MQ-2
    │ I2C0/I2C1/ADC
    ▼
ESP32 FreeRTOS 任务
    │ 队列通信（lux_queue / climate_queue）
    ▼
app_logic_task（合流 + JSON序列化）
    ├─ WiFi 正常 → MQTT Publish（esp32bishe/sensor/all）
    └─ 断网 → SD 环形缓存 → 恢复后 backfill 补发
    ▼
FastAPI 内嵌 amqtt Broker（:1883）
    │ paho-mqtt on_message 回调
    ▼
dispatch_mqtt_message() → SQLAlchemy → MySQL
    │ 同时更新 SensorStore 内存缓存
    ▼
REST API（/api/v1/*）
    │ JWT Bearer 鉴权
    ▼
┌─ Web 前端（my-app）           ┌─ 桌面应用（sensorApp）
│  ApiCache + fetchWithRetry    │  Tauri v2 原生窗口
│  ZoneMap SVG 拓扑             │  React + ECharts
│  ECharts 多参曲线             │  系统托盘常驻
│  CSV 流式导出 + 进度          │
│  JWT 自动续期/401跳转         │
└───────────────────────────────┴──────────────────────────

技术栈总汇

层技术要点
MCUC++ / ESP-IDF / FreeRTOS四任务并行，队列通信
传感器BH1750 + SHT35 + MQ-2I2C0/I2C1 双总线
WiFiesp-wifi-connectAP 热点配网
SD 缓存环形缓冲区 + backfilldelta-pack + Base64 协议
MQTTJSON (all/climate/lux/gas) + binary双向：发布 + control 订阅
Brokeramqtt (FastAPI 进程内)无需 Mosquitto
后端Python FastAPIlifespan 管理全部生命周期
ORMSQLAlchemy + PyMySQLMySQL 持久化
鉴权python-jose JWTBearer Token
定时APScheduler统计/告警/清理
Web 前端React 19 + Vite + ECharts 615 个模块，495 行 API 层
API 层ApiCache + fetchWithRetryTTL缓存 + 指数退避重试
桌面应用Tauri v2 (Rust + React)原生窗口监控面板
数据处理Python/Node.js 脚本集论文排版/三线表/文档打包

总结

这个项目的真正复杂度远远超过"传感器+MQTT+图表"：

  1. ESP32 端：不仅仅是采集——SD 离线缓存、backfill 补发协议（魔数+版本+delta-pack+Base64）、MQTT 收发确认（ACK 5s 超时）
  2. 后端：FastAPI 内嵌 amqtt Broker（零外部依赖）、SQLAlchemy 自动建表+种子数据、APScheduler 定时任务、9 组 API 端点（含告警模拟）
  3. Web 前端：ApiCache 客户端缓存层、fetchWithRetry 指数退避、自动 401 拦截登录重定向、流式 CSV 导出带进度、ZoneMap SVG 地图
  4. 桌面应用：Tauri v2（Rust+React）打包为原生应用
  5. 工程化：i18n 国际化、ErrorBoundary、LoadingSpinner/EmptyState 四状态覆盖、StatusBadge 含单元测试、隐私模块、主题系统

相关推荐

评论 (0)

加载评论中...