基于ESP32+Python的蓝牙设备动态监测系统：实现云端可配置的设备通断检测

一、前言

在物联网项目中，经常需要监测蓝牙设备（如人员工牌、资产标签）的进出范围情况，实现“设备通断”检测。但实际场景中，监测区域内可能有许多蓝牙设备，我们只关心部分设备，且关注列表可能随时变化。如果每次收到数据都全部处理，不仅浪费资源，还可能干扰业务逻辑。

本文将介绍一个完整的解决方案：使用 ESP32 作为蓝牙扫描节点，通过串口将扫描到的设备信息（JSON格式）发送给上位机（例如树莓派或工控机）。上位机运行 Python 程序接收数据，并根据本地配置文件（由云端通过MQTT动态下发）决定是否处理该设备。当设备状态为“启用”时，记录或上报；为“禁用”时，直接忽略。整个过程无需重启程序，配置实时生效。

二、系统整体架构

┌─────────┐      ┌─────────┐      ┌─────────┐      ┌─────────┐
│ 蓝牙设备 │ ───> │  ESP32  │ ───> │ 上位机  │ ───> │ 云平台  │
└─────────┘      └─────────┘      └─────────┘      └─────────┘
   (标签)          (扫描+串口)      (过滤+处理)        (下发配置)

• 蓝牙设备：持续广播自身信息（设备ID、名称、RSSI等）。
• ESP32：进行低功耗蓝牙扫描，获取周围设备信息，打包成JSON通过串口发送。
• 上位机（Linux系统）：运行Python脚本监听串口，解析JSON，根据配置文件决定是否处理。
• 云平台：通过MQTT协议与上位机通信，下发需要监测的蓝牙设备列表，上位机实时更新本地配置。

三、代码实现详解

1. 蓝牙设备配置管理模块（ble_device_config.py）

该模块负责解析、保存、加载蓝牙设备的状态配置。核心数据结构是一个字典，键为设备ID，值为1（启用）或0（禁用）。配置以字符串形式通过云端下发，格式为：

7cba60,1|a1b2c3,0|d4e5f6,1

完整代码：

# ble_device_config.py
import json
import os

# 设备状态配置文件路径
DEVICE_CONFIG_FILE = "/ibox/ble_device_status.json"

def parse_device_config(config_str):
    """解析配置字符串为字典 {设备ID: 状态}"""
    device_status = {}
    if not config_str or config_str.strip() == "":
        return device_status
    items = config_str.split('|')
    for item in items:
        if ',' in item:
            device_id, status = item.split(',')
            device_id = device_id.strip()
            status = status.strip()
            if device_id and status in ['0', '1']:
                device_status[device_id] = int(status)
    return device_status

def save_device_config(config_str):
    """保存配置到JSON文件"""
    try:
        device_status = parse_device_config(config_str)
        # 确保目录存在
        os.makedirs(os.path.dirname(DEVICE_CONFIG_FILE), exist_ok=True)
        with open(DEVICE_CONFIG_FILE, 'w', encoding='utf-8') as f:
            json.dump(device_status, f, indent=4, ensure_ascii=False)
        return True
    except Exception as e:
        print(f"保存设备配置失败: {e}")
        return False

def load_device_config():
    """从JSON文件加载设备状态字典"""
    try:
        if not os.path.exists(DEVICE_CONFIG_FILE):
            return {}
        with open(DEVICE_CONFIG_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        print(f"加载设备配置失败: {e}")
        return {}

def is_device_enabled(device_id):
    """检查设备是否启用（实时读取配置文件）"""
    device_status = load_device_config()
    return device_status.get(device_id, 0) == 1

设计亮点：

• 每次调用 is_device_enabled 都会重新读取JSON文件，确保配置实时生效。
• 配置持久化到文件，避免程序重启丢失。
• 解析函数支持灵活的配置字符串格式。

2. 上位机串口接收程序（hdmi_blue.py）

该程序使用 pyserial 监听串口，读取ESP32发送的JSON数据，并根据设备是否启用来决定是否处理。

完整代码：

#!/usr/bin/env python3
# hdmi_blue.py
import serial
import json
import time
from ble_device_config import is_device_enabled

class MCUUpperComputer:
    def __init__(self, port, baudrate=115200):
        self.port = port
        self.baudrate = baudrate
        self.ser = None
        self.running = False

    def connect(self):
        try:
            self.ser = serial.Serial(
                port=self.port,
                baudrate=self.baudrate,
                bytesize=serial.EIGHTBITS,
                parity=serial.PARITY_NONE,
                stopbits=serial.STOPBITS_ONE,
                timeout=1
            )
            print(f"✅ 成功连接到串口 {self.port}")
            return True
        except Exception as e:
            print(f"❌ 连接串口失败：{e}")
            return False

    def process_json_data(self, json_data):
        device_id = json_data.get('device_id', '')
        # 检查设备是否启用
        if not is_device_enabled(device_id):
            print(f"⚠️  设备 {device_id} 未启用，跳过输出")
            return
        # 打印设备信息
        print("=" * 60)
        print(f"📱 设备名称：{json_data.get('name', 'N/A')}")
        print(f"📡 信号强度：{json_data.get('rssi', 'N/A')} dBm")
        print(f"🔗 设备地址：{json_data.get('address', 'N/A')}")
        print(f"🏷️  设备ID：{device_id}")
        print(f"⚡ GPIO电平：{'高(1)' if json_data.get('gpio_level', 0) else '低(0)'}")
        print("=" * 60)

    def start_receiving(self):
        if not self.ser or not self.ser.is_open:
            print("❌ 串口未连接")
            return
        self.running = True
        buffer = ""
        print("🔄 开始接收数据，按 Ctrl+C 停止")
        try:
            while self.running:
                if self.ser.in_waiting > 0:
                    data = self.ser.read(self.ser.in_waiting).decode('utf-8', errors='ignore')
                    buffer += data
                    # 按换行符分割完整的JSON行
                    while '\n' in buffer:
                        line, buffer = buffer.split('\n', 1)
                        line = line.strip()
                        if line:
                            try:
                                json_data = json.loads(line)
                                self.process_json_data(json_data)
                            except json.JSONDecodeError:
                                print(f"⚠️ JSON解析失败：{line}")
                time.sleep(0.01)
        except KeyboardInterrupt:
            print("\n🛑 用户中断")
        finally:
            self.stop()

    def stop(self):
        self.running = False
        if self.ser and self.ser.is_open:
            self.ser.close()
            print("✅ 串口已关闭")

def main():
    port = "/dev/ttyS1"   # 根据实际情况修改
    upper = MCUUpperComputer(port=port, baudrate=115200)
    if upper.connect():
        upper.start_receiving()

if __name__ == "__main__":
    main()

注意点：

• ESP32发送的JSON必须以换行符结尾，否则无法正确分割。
• 使用 errors='ignore' 避免解码错误导致程序崩溃。
• 每次处理都调用 is_device_enabled，保证配置动态生效。

3. MQTT客户端处理模块（mqtt_client_handler.py 节选）

该模块负责连接MQTT服务器，订阅主题，接收云端下发的设备配置，并调用 save_device_config 更新本地文件。

关键代码片段：

# mqtt_client_handler.py 中相关部分
from ble_device_config import save_device_config

class MqttClientHandler:
    # ... 初始化等省略

    def process_function(self, payload):
        try:
            data = json.loads(payload)
            for item in data:
                id_val = item["id"]
                value = item["value"]
                # ... 其他功能处理
                elif id_val == "ble_device_config":
                    self.log_handler.print_msg(f"蓝牙设备配置： {value}")
                    if save_device_config(value):
                        self.log_handler.print_msg("蓝牙设备配置保存成功")
                    else:
                        self.log_handler.print_msg("蓝牙设备配置保存失败")
                # ...
            self.publish_property(json.dumps(data))
        except Exception as e:
            self.log_handler.print_msg(f"处理功能数据时出错: {e}")

MQTT消息示例（发布到上位机订阅的主题）：

[{"id":"ble_device_config","value":"7cba60,1|a1b2c3,0|d4e5f6,1"}]

上位机收到后立即解析并保存，之后串口接收程序中的 is_device_enabled 就会读到新配置，实现动态切换。

4. 配置文件实时生效的原理

为什么不需要重启程序？因为 is_device_enabled 每次调用都会重新读取JSON文件，而不是在程序启动时一次性加载。这种设计虽然增加了磁盘I/O，但在设备数量不多、调用频率不高的情况下完全可以接受，且极大地简化了动态更新的实现。

def is_device_enabled(device_id):
    device_status = load_device_config()  # 每次都重新加载
    return device_status.get(device_id, 0) == 1

四、实验演示：动态切换监测设备

假设我们有两个蓝牙设备：A（ID: 7cba60）和B（ID: a1b2c3）。

1. 初始状态：配置文件为空或所有设备状态为0，运行 hdmi_blue.py，不输出任何信息。
2. 云端下发配置：通过MQTT发送启用A、禁用B的消息。

   [{"id":"ble_device_config","value":"7cba60,1|a1b2c3,0"}]
   

3. 上位机收到后更新JSON文件。此时当ESP32上报A的数据时，程序打印详细信息；上报B时跳过。

   ============================================================
   📱 设备名称：Tag-01
   📡 信号强度：-65 dBm
   🔗 设备地址：aa:bb:cc:dd:ee:ff
   🏷️  设备ID：7cba60
   ============================================================
   ⚠️  设备 a1b2c3 未启用，跳过输出
   

4. 再次下发配置：切换为禁用A、启用B。

   [{"id":"ble_device_config","value":"7cba60,0|a1b2c3,1"}]
   

   程序立即生效，只处理B，A被忽略。

五、总结与扩展

本文通过一个实际案例，展示了如何构建一个动态配置的蓝牙设备监测系统。核心思想是利用 配置文件+实时读取 实现动态生效，结合 MQTT远程下发 实现云端控制。这套架构可以灵活扩展：

• 多节点部署：增加多个ESP32，实现区域覆盖，数据汇总到上位机。
• 数据上报：将RSSI和启用设备信息上报云端，用于人员定位或资产盘点。
• 联动控制：根据特定设备出现或消失，触发本地继电器、报警等动作。

如果你正在开发类似的物联网项目，可以直接复用这里的代码，只需根据实际需求调整数据格式和处理逻辑。希望本文能给你带来启发！

---

注意事项：

• 确保上位机已安装依赖：pip install pyserial aiomqtt
• 串口设备名请根据实际情况修改（如 /dev/ttyUSB0、COM3）。
• ESP32端需实现蓝牙扫描并按行发送JSON，可参考以下简单格式：

  {"device_id":"7cba60","name":"Tag-01","rssi":-65,"address":"aa:bb:cc:dd:ee:ff","gpio_level":1}
  

欢迎留言交流，共同进步！