一、所选主题和项目介绍

1.1 项目背景

在FastBond4多设备物联网系统中，多个M5Stack终端设备需要统一的用户身份认证和数据汇聚中心。传统的方案依赖云端服务器，存在延迟高、依赖网络、成本高等问题。树莓派作为低功耗、低成本的边缘计算设备，可以在本地完成数据汇聚、用户认证和信息存储，大大降低系统延迟和运营成本。

本项目以树莓派3B+为核心，运行Mosquitto MQTT Broker作为消息中间件，部署Python数据存储服务，配合M5Stack终端设备的RFID刷卡模块，构建了一套完整的本地化用户登录追踪与数据采集系统。当用户在M5Stack终端刷卡时，用户ID通过MQTT实时上报到树莓派，树莓派将登录信息存入SQLite数据库并同时在终端屏幕输出，实现本地化、实时化的用户管理。

1.2 项目目标

1. RFID用户登录：M5Stack终端通过RFID刷卡识别用户身份
2. MQTT实时传输：用户登录信息通过MQTT实时上报到树莓派
3. SQLite持久化存储：树莓派接收并存储所有用户登录记录
4. 终端实时输出：树莓派终端实时显示用户登录信息
5. 多设备统一管理：支持多台M5Stack终端同时接入
6. Mosquitto MQTT Broker：树莓派本地的消息中间件

1.3 创新点

1. 边缘计算架构：树莓派作为本地服务器，实现低延迟的数据处理
2. RFID+MQTT融合：将RFID身份识别与MQTT物联网协议结合
3. 通配符订阅：fastbond4/+/+/+ 自动适配任意新设备接入
4. 零配置接入：新设备上线即自动注册，无需手动配置
5. 终端实时输出：用户登录信息在树莓派终端彩色高亮显示
6. 离线模式支持：M5Stack终端在网络异常时仍可本地运行
7. 即插即用：部署脚本一键完成安装配置

二、硬件介绍

2.1 主控设备 - 树莓派3B+

树莓派3B+ (Raspberry Pi 3 Model B+)

树莓派3B+是Raspberry Pi基金会的第三代升级型号，采用BCM2837B0处理器，四核Cortex-A53架构，主频1.4GHz。板载1GB LPDDR2内存，支持2.4/5GHz双频WiFi和千兆以太网（通过USB 2.0实现）。作为边缘计算服务器，树莓派3B+能够同时处理多台M5Stack设备的数据上报和用户登录请求。

核心参数

软件环境

2.2 终端设备 - M5Stack CoreS3-SE

M5Stack CoreS3-SE (SKU: K128-SE)

CoreS3-SE是M5Stack开发套件系列的第三代主机CoreS3的轻量级版本，其核心主控采用ESP32-S3方案，双核Xtensa LX7处理器，主频240MHz，自带2.4G Wi-Fi功能，板载16MB Flash和8MB PSRAM。正面搭载2.0寸IPS屏，用于显示用户登录状态和操作提示。

2.3 RFID模块

Unit RFID2 (SKU: U031-B)

Unit RFID2是一款基于13.56MHz频率的射频识别读写单元。它内置WS1850S芯片，遵循ISO/IEC 14443 Type A/B标准，可支持MIFARE、NTAG等系列RFID卡的数据读写。单元通过I2C接口通信，读写距离小于20mm。

2.4 硬件连接

Unit RFID2连接

2.5 硬件框图

三、方案框图和项目设计思路

3.1 系统架构

3.2 程序流程图

3.2.1 M5Stack终端程序流程

3.2.2 树莓派数据服务流程

3.3 用户登录算法

采用RFID刷卡触发 + MQTT实时上报 + SQLite持久化存储的完整流程：

1. RFID检测：M5Stack通过I2C读取RFID卡片UID
2. 用户查找：根据RFID UID在本地用户数据库查找用户
3. MQTT上报：通过 fastbond4/{device_id}/login/user 主题上报JSON格式的登录信息
4. 树莓派接收：paho-mqtt回调函数接收MQTT消息
5. 主题解析：根据主题路径解析device_id和事件类型
6. 数据存储：将登录事件写入SQLite user_login_log 表
7. 终端输出：格式化输出用户登录信息到树莓派终端

为了测试使用了随机验证方式

四、原理图和PCB介绍

4.1 原理图展示

原理图

4.2 PCB布局展示

PCB

4.3 设计过程

通过KiCad完成了电路的原理图绘制和PCB版图设计。关于设计方案，方案选择了最简单的直通连接方式，将I2C总线的SCL和SDA信号直接分到各个接口，通过不同的I2C地址来区分设备。

4.4 系统网络连接示意图

本项目设计了专用的接口转接板（PCB），用于连接 M5Stack 主控与 RFID 读卡器等外设模块。转接板通过自定义 PCB 走线将 I2C 等信号可靠引出，实现了稳定的硬件连接与数据通信。

4.5 MQTT通信协议

登录事件 (M5Stack → 树莓派)

主题: fastbond4/{device_id}/login/user

{
    "event": "login",
    "rfid_uid": "A1B2C3D4",
    "user_name": "张三",
    "device_id": "login_terminal_01",
    "timestamp": 1747715400
}

传感器数据 (M5Stack → 树莓派)

主题: fastbond4/{device_id}/sensor/{type}

client.publish("temperature", {"value": 25.6, "unit": "celsius"})

五、软件流程和关键代码介绍

5.1 主程序结构 - 树莓派数据服务

"""
树莓派MQTT数据存储服务 - 支持用户登录追踪与传感器数据存储

功能说明:
    1. 订阅 fastbond4/+/+/+ 通配符主题
    2. 接收所有M5Stack设备上报的数据
    3. 用户登录信息存储到SQLite并在终端输出
    4. 传感器数据存储到SQLite

部署:
    树莓派3B/B+/4, Raspberry Pi OS, Python 3.7+
    依赖: paho-mqtt (pip3 install paho-mqtt), sqlite3 (内置)
"""
import sqlite3
import json
import time
import os
import signal
import sys
import argparse
from datetime import datetime
import paho.mqtt.client as mqtt

5.2 硬件初始化 - M5Stack终端

"""
M5Stack RFID用户登录终端 - RFID刷卡后通过MQTT上报登录信息到树莓派

功能说明:
    1. 连接WiFi和树莓派MQTT服务器
    2. RFID读卡验证用户身份
    3. 通过MQTT将用户登录信息上报到树莓派
    4. LCD屏幕显示用户登录状态

硬件:
    - CoreS3-SE (ESP32-S3主控, SKU: K128-SE)
    - Unit RFID2 (RFID读写单元, I2C地址: 0x28, SKU: U031-B)
"""

def setup():
    """系统初始化"""
    M5.begin()
    Lcd.init()
    Lcd.clear(0x000000)

    # 初始化I2C总线
    i2c = I2C(0, scl=Pin(11), sda=Pin(12), freq=100000)

    # 初始化RFID
    rfid = RFIDUnit(i2c, 0x28)

    # 连接WiFi和MQTT
    mqtt_client = M5MqttClient(WIFI_SSID, WIFI_PASS, MQTT_BROKER, DEVICE_ID)
    mqtt_client.connect()

    # 绘制登录界面
    draw_login_screen()

5.3 RFID刷卡检测

def check_rfid_card():
    """检查RFID卡片，读取UID并上报登录信息"""
    global rfid_authorized, current_user_name, current_rfid_uid

    if rfid is None:
        return False

    try:
        if rfid.is_card_present():
            card_uid = rfid.read_uid()
            if card_uid:
                uid_str = str(card_uid)
                print("[RFID] 检测到卡片: {}".format(uid_str))

                user_info = lookup_user(uid_str)
                if user_info:
                    user_name = user_info["name"]
                else:
                    user_name = "未知用户"

                # 通过MQTT上报登录信息
                if mqtt_client and mqtt_client.is_connected:
                    mqtt_client.publish_login(uid_str, user_name)

                # 更新显示
                draw_auth_status(user_name, uid_str, is_login=True)
                return True

    except Exception as e:
        print("[Error] RFID读取失败: {}".format(e))

    return False

5.4 MQTT登录消息发布

def publish_login(self, rfid_uid, user_name):
    """
    发布用户登录事件到MQTT

    树莓派终端会实时输出:
    ============================================
    [LOGIN]  设备: login_terminal_01 | 用户: 张三 | UID: A1B2C3D4
    ============================================
    """
    topic = "fastbond4/{}/login/user".format(self._device_id)
    payload = json.dumps({
        "event": "login",
        "rfid_uid": rfid_uid,
        "user_name": user_name,
        "device_id": self._device_id,
        "timestamp": int(time.time())
    })

    try:
        self._client.publish(topic, payload)
        return True
    except Exception as e:
        print("[M5Mqtt] 登录上报失败: {}".format(e))
        return False

5.5 树莓派数据存储

def store_user_login(device_id, event, rfid_uid, user_name):
    """
    存储用户登录/登出事件到数据库

    数据库表: user_login_log
    字段: timestamp, device_id, event, rfid_uid, user_name
    """
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    cursor.execute(
        "INSERT INTO user_login_log "
        "(timestamp, device_id, event, rfid_uid, user_name) "
        "VALUES (?, ?, ?, ?, ?)",
        (now_str, device_id, event, rfid_uid, user_name)
    )

    conn.commit()
    conn.close()

    # 终端彩色输出
    print("\n============================================")
    print("{} [LOGIN {}".format(event.upper(), _login_count))
    print("  设备: {}".format(device_id))
    print("  用户: {}".format(user_name))
    print("  RFID: {}".format(rfid_uid))
    print("  时间: {}".format(now_str))
    print("============================================\n")

5.6 数据库初始化

def init_database():
    """初始化数据库，创建数据表和用户登录日志表"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()

    # 传感器数据日志
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS data_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            device_id TEXT NOT NULL,
            sensor_type TEXT NOT NULL,
            value TEXT NOT NULL,
            raw_payload TEXT
        )
    """)

    # 用户登录日志
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS user_login_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            device_id TEXT NOT NULL,
            event TEXT NOT NULL,
            rfid_uid TEXT NOT NULL,
            user_name TEXT NOT NULL
        )
    """)

    conn.commit()
    conn.close()

5.7 MQTT消息回调

def on_message(client, userdata, msg):
    """MQTT消息回调 - 处理用户登录事件和传感器数据"""
    try:
        topic_parts = msg.topic.split("/")
        device_id = topic_parts[1]
        event_type = topic_parts[2]

        # 判断是否为登录/登出事件
        if event_type in ("login", "logout"):
            data = json.loads(msg.payload)
            rfid_uid = data.get("rfid_uid", "unknown")
            user_name = data.get("user_name", "unknown")
            store_user_login(device_id, event_type, rfid_uid, user_name)
            return

        # 传感器数据存储
        sensor_type = topic_parts[3]
        value_str = format_value(json.loads(msg.payload))
        store_data(device_id, sensor_type, value_str, msg.payload)

    except Exception as e:
        print("[Error] 消息处理异常: {}".format(e))

5.8 主循环

def loop():
    """M5Stack主循环 - 持续检测RFID卡片和按键操作"""
    global rfid_authorized

    M5.update()

    # 未授权状态下检测RFID
    if not rfid_authorized:
        if check_rfid_card():
            time.sleep_ms(500)

    # 按键A: 用户登出
    if BtnA.wasPressed():
        if mqtt_client and mqtt_client.is_connected:
            mqtt_client.publish_logout(current_rfid_uid, current_user_name)

        rfid_authorized = False
        current_user_name = ""
        current_rfid_uid = ""

        draw_login_screen()

    time.sleep_ms(100)

5.9 服务启动

def main():
    """树莓派主入口 - 启动MQTT数据存储服务"""
    parser = argparse.ArgumentParser(description="FastBond4 MQTT数据存储服务")
    parser.add_argument("--broker", default="localhost", help="MQTT服务器地址")
    parser.add_argument("--port", type=int, default=1883, help="MQTT服务器端口")
    parser.add_argument("--stats", action="store_true", help="仅显示统计信息后退出")
    parser.add_argument("--db", default="fastbond4_data.db", help="数据库文件路径")
    args = parser.parse_args()

    init_database()

    mqtt_client = mqtt.Client("rpi_service_{}".format(int(time.time())))
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = on_message
    mqtt_client.connect(args.broker, args.port, keepalive=60)
    mqtt_client.loop_start()

    print("FastBond4 MQTT数据存储服务")
    print("用户登录信息将通过终端实时输出...")

    while _running:
        time.sleep(1)

六、硬件功能展示图及说明

6.1 M5Stack终端显示界面

6.1.1 刷卡界面

6.1.2 登出界面

6.2 树莓派终端输出

==================================================
FastBond4 MQTT数据存储服务
  Broker: localhost:1883
  数据库: fastbond4_data.db
  订阅主题: fastbond4/+/+/+
==================================================
  用户登录信息将通过终端实时输出...
==================================================

[MQTT] 已连接到服务器
[MQTT] 已订阅主题: fastbond4/+/+/+

============================================
LOGIN  [001]
  设备: login_terminal_01
  用户: 张三
  RFID: A1B2C3D4
  时间: 2026-05-14 10:30:00
============================================

============================================
LOGIN  [002]
  设备: login_terminal_01
  用户: 李四
  RFID: E5F6G7H8
  时间: 2026-05-14 10:31:00
============================================

============================================
LOGOUT [003]
  设备: login_terminal_01
  用户: 张三
  RFID: A1B2C3D4
  时间: 2026-05-14 10:35:00
============================================

6.3 统计信息

# 显示数据库统计
python3 rpi_mqtt_data_service.py --stats

=======================================================
  FastBond4 数据服务统计
=======================================================
传感器总记录数: 150
用户登录总记录数: 25

各设备/传感器数据量:
  [sensor_device_01]
    temperature: 50 条
    humidity: 50 条
    light: 50 条

已注册设备:
  login_terminal_01 : 25 条, 最后活跃: 2026-05-14 10:35:00
  sensor_device_01 : 150 条, 最后活跃: 2026-05-14 10:30:00

用户登录统计:
  张三 : 10 次
  李四 : 8 次
  王五 : 7 次
=======================================================

七、设计中遇到的难题和解决方法

7.1 RFID读卡稳定性

问题描述：RFID读卡可能存在读取不稳定或距离不足的情况。

解决方案：

1. 在主循环中持续检测RFID卡片
2. 使用 is_card_present() 方法先检测卡片存在
3. 再使用 read_uid() 读取卡片UID
4. 添加异常处理机制

def check_rfid_card():
    """检查RFID卡片"""
    if rfid is None:
        return False

    try:
        if rfid.is_card_present():
            card_uid = rfid.read_uid()
            if card_uid:
                print("[RFID] 检测到卡片: {}".format(card_uid))
                return True
    except Exception as e:
        print("[Error] RFID读取失败: {}".format(e))

    return False

7.2 MQTT连接稳定性

问题描述：MQTT连接可能因WiFi不稳定或树莓派重启而断开。

解决方案：

1. 在M5MqttClient中提供 reconnect_mqtt() 方法
2. 在主循环中检查MQTT连接状态
3. 断开时自动重连
4. 支持离线模式（无MQTT时本地运行）

# MQTT重连
def reconnect_mqtt(self):
    """重新连接MQTT"""
    self._mqtt_connected = False
    return self.connect_mqtt()

# 离线模式
if mqtt_client and mqtt_client.is_connected:
    mqtt_client.publish_login(uid_str, user_name)
else:
    print("[System] 用户 {} 登录 (离线模式)".format(user_name))

7.3 SQLite并发访问

问题描述：MQTT消息可能同时到达，导致SQLite并发写入问题。

解决方案：

1. SQLite3默认支持串行化写入
2. 每个数据库操作使用独立的connection
3. 操作完成后立即close连接
4. 使用WAL模式提升并发性能

def store_user_login(device_id, event, rfid_uid, user_name):
    """存储用户登录事件 - 每次使用独立连接"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    conn.execute("PRAGMA journal_mode=WAL")

    cursor.execute(
        "INSERT INTO user_login_log (timestamp, device_id, event, rfid_uid, user_name) "
        "VALUES (?, ?, ?, ?, ?)",
        (now_str, device_id, event, rfid_uid, user_name)
    )

    conn.commit()
    conn.close()

7.4 终端输出格式化

问题描述：需要清晰、美观地格式化用户登录信息到终端。

解决方案：

1. 使用边框分隔不同登录事件
2. 使用ANSI颜色码区分登录/登出
3. 添加序号方便追踪
4. 对齐字段显示

# 终端彩色输出
COLOR_GREEN = "\033[92m"
COLOR_YELLOW = "\033[93m"
COLOR_CYAN = "\033[96m"
COLOR_RESET = "\033[0m"

event_color = COLOR_GREEN if event == "login" else COLOR_YELLOW
print("{}============================================{}".format(COLOR_CYAN, COLOR_RESET))
print("{}{}{}  [{:03d}]".format(event_color, event_label, COLOR_RESET, _login_count))
print("  设备: {}".format(device_id))
print("  用户: {}".format(user_name))
print("  RFID: {}".format(rfid_uid))
print("  时间: {}".format(now_str))
print("{}============================================{}".format(COLOR_CYAN, COLOR_RESET))

7.5 用户数据库扩展

问题描述：需要在M5Stack设备上维护用户RFID映射关系。

解决方案：

1. 在代码中定义 USER_DATABASE 字典
2. RFID UID作为key
3. 用户信息（姓名、角色）作为value
4. 可通过MQTT主题远程更新

# 本地用户数据库 (可扩展)
USER_DATABASE = {
    "A1B2C3D4": {"name": "张三", "role": "管理员"},
    "E5F6G7H8": {"name": "李四", "role": "普通用户"},
    "I9J0K1L2": {"name": "王五", "role": "普通用户"},
}

def lookup_user(rfid_uid):
    """根据RFID UID查找用户信息"""
    uid_str = str(rfid_uid).strip().upper()
    return USER_DATABASE.get(uid_str)

7.6 系统自启与服务化

问题描述：树莓派数据服务需要在系统启动后自动运行。

解决方案：

1. 使用systemd创建系统服务
2. 设置mosquitto为开机自启
3. 数据服务依赖mosquitto启动
4. 使用Restart=always确保意外退出后自动重启

# systemd服务配置
[Unit]
Description=FastBond4 MQTT Data Storage Service
After=network.target mosquitto.service
Wants=mosquitto.service

[Service]
Type=simple
ExecStart=/usr/bin/python3 /opt/rpi-mqtt-service/rpi_mqtt_data_service.py --broker localhost
WorkingDirectory=/opt/rpi-mqtt-service
Restart=always
RestartSec=10
User=pi

[Install]
WantedBy=multi-user.target

八、心得体会

8.1 项目收获

1. 树莓派边缘计算：掌握了树莓派作为本地MQTT服务器的完整部署流程
2. MQTT物联网通信：深入理解了MQTT协议的主题订阅/发布机制
3. RFID身份识别：学习了RFID射频识别技术在用户身份认证中的应用
4. SQLite数据存储：掌握了SQLite轻量级数据库在多用户场景下的使用
5. M5Stack + 树莓派协同：实践了M5Stack终端与树莓派服务器的协同工作
6. 多设备管理：学习了通过MQTT通配符主题管理多台设备的方法
7. 边缘计算架构：体验了本地服务器 vs 云端服务器的优势和权衡

8.2 技术亮点

1. 本地化部署：树莓派作为本地边缘计算服务器，降低延迟
2. MQTT通配符订阅：使用 fastbond4/+/+/+ 自动适配任意新设备
3. 用户登录实时输出：树莓派终端彩色高亮输出用户登录信息
4. 零配置接入：新设备按 fastbond4/{device_id} 发布即自动注册
5. SQLite轻量存储：无需安装数据库服务，即开即用
6. systemd服务化：支持7×24小时无人值守运行
7. 离线模式支持：M5Stack终端在网络异常时仍可本地记录

8.3 改进建议

1. Web管理界面：添加Flask/FastAPI Web界面，方便远程查看登录记录
2. 用户管理API：添加REST API，支持远程添加/删除/修改用户
3. 登录日志导出：支持CSV/JSON格式导出登录日志
4. 多级权限：根据用户角色分配不同的设备操作权限
5. 短信/邮件通知：检测到异常登录时发送通知
6. 人脸识别扩展：结合摄像头实现人脸+RFID双重认证
7. 数据备份：定期将SQLite数据备份到云端
8. MQTT TLS加密：配置Mosquitto TLS证书，加密MQTT通信
9. WebSocket支持：配置Mosquitto WebSocket支持，方便Web端接入
10. Docker部署：将服务打包为Docker镜像，一键部署
11. 统计分析：添加用户登录频率、时段等统计分析
12. 多语言支持：RFID卡片支持多语言用户信息

创意方向关联

基于本项目的技术成果，我们可以延伸到以下创意方向：

1. 人工智能在嵌入式系统中的应用

本项目MQTT + RFID身份识别为AI应用提供了基础：

• 异常检测：分析用户登录行为模式，AI识别异常登录
• 图像识别：MQTT数据管道可扩展为基于计算机视觉的身份识别
• 行为分析：结合传感器数据，AI分析用户行为模式

2. 机械臂 / 工业自动化

RFID和MQTT技术在自动化领域有广泛应用：

• 机械臂：RFID可用于机械臂工具识别和自动换刀系统
• 人员追踪：MQTT + RFID实现工厂人员位置追踪
• 工时统计：RFID刷卡自动统计工人工时

3. 楼宇自动化

RFID身份识别技术 + MQTT通信是楼宇自动化的关键：

• 智能门禁：RFID刷卡门禁 + MQTT统一管理
• 访客管理：RFID身份验证 + 树莓派数据库存储
• 考勤系统：RFID + MQTT + SQLite = 完整考勤系统

技术迁移价值

本项目开发的RFID + MQTT + SQLite + RPi技术方案可直接迁移到：

• AI应用：用户行为监测、异常检测
• 工业自动化：人员追踪、工时统计
• 楼宇自动化：门禁管理、访客系统
• 教育场景：实验室设备借还管理、学生考勤

致谢

感谢 DigiKey 和 电子森林 提供的FastBond4活动支持，本次活动链接：https://www.eetree.cn/page/digikey-fastbond

感谢树莓派（Raspberry Pi）基金会的优秀硬件平台，使得边缘计算项目变得简单易行。

感谢Mosquitto MQTT Broker开源项目和paho-mqtt Python库提供的强大通信能力。

感谢所有参与测试和提供反馈的老师和同学们，你们的建议让这个项目更加完善。