Python物联网开发实战：MQTT、BLE与HTTP设备连接代码详解

脚本专家

在物联网（IoT）开发中，Python 凭借简洁的语法、动态类型和丰富的第三方库，已经成为连接智能设备、采集传感器数据和构建业务逻辑的常用语言。本文围绕数据采集、数据传输和业务应用三个核心环节，给出可直接运行的 Python 代码示例，并说明其中涉及的协议、函数与参数含义。

一、数据采集：从 Wi-Fi 与 BLE 设备读取数据

物联网设备的数据采集通常依赖无线通信模块。以下示例分别演示了通过 UDP 套接字读取 Wi-Fi 模块信号强度，以及通过蓝牙 RFCOMM 协议读取手机定位数据。

1. 使用 socket 读取 Wi-Fi RSSI 值

Wi-Fi 模块通常会将信号强度数据打包发送到局域网内的某个端口。下面代码创建一个 UDP 套接字，绑定本机 8889 端口，不断接收数据包，并从数据中截取以“RSSI:”开头的字段解析为有符号字节值。

2. import socket
3. import struct
5. s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
6. s.bind(('localhost', 8889))
8. while True:
9.     data, addr = s.recvfrom(1024)
10.     # 数据格式假设为：RSSI:-45，这里取出冒号后的部分，按有符号单字节解析
11.     rssi = struct.unpack('<b', data[len('RSSI:'):])[0]
12.     print("Received:", rssi, "dBm from", addr)

复制代码

代码要点：recvfrom 返回数据及来源地址；struct.unpack 的格式 '<b' 表示小端有符号字节，用于还原 RSSI 数值。实际项目中，Wi-Fi 模块需要先按约定格式发送数据包，例如将字符串“RSSI:-45”作为 UDP 报文发出。

2. 使用 bluepy 读取 BLE 定位数据

对于低功耗蓝牙（BLE）设备，可以通过 bluepy 库建立 RFCOMM 连接。下面的代码尝试连接一个指定蓝牙地址的设备，并循环接收其发送的定位信息。

2. import bluetooth
3. import sys
5. address = "F7:CA:E6:C1:D4:B9"  # 替换为目标设备地址
7. try:
8.     sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
9.     sock.connect((address, 1))
10.     while True:
11.         data = sock.recv(1024).decode()
12.         if len(data) > 0:
13.             items = data.split(',')
14.             lat = float(items[0])   # 纬度
15.             lng = float(items[1])   # 经度
16.             altitude = int(items[2]) # 高度
17.             print("Latitude:", lat, "Longitude:", lng, "Altitude:", altitude, "m")
18. finally:
19.     try:
20.         sock.close()
21.     except NameError:
22.         pass

复制代码

代码要点：device address 是蓝牙适配器的 MAC 地址；RFCOMM 通道号通常为 1。如果连接失败，会抛出 BluetoothError，需要在真实项目中增加异常捕获和重试逻辑。

二、数据传输：MQTT 与 HTTP 接入云服务

采集到数据后，需要上传至云端或业务服务器。MQTT 适合低带宽、高延迟或不可靠网络环境，HTTP 则适合请求/响应模式的 API 调用。

1. 使用 paho-mqtt 发布消息到 ThingSpeak

ThingSpeak 是一个支持 MQTT 的物联网数据平台。先注册账号并创建 Channel，获得用户名和密码。下面的代码通过 paho-mqtt 库连接 mqtt.thingspeak.com，然后向指定主题发布带时间戳和随机数的消息。

2. import paho.mqtt.client as mqtt
3. import time
4. import random
6. broker_address = "mqtt.thingspeak.com"
7. port = 1883
8. topic = "channels/1417/publish/"
9. message = "field1=%d&field2=%d" % (time.time(), int(random.randint(0, 10) * 10))
11. client = mqtt.Client()
13. def on_connect(client, userdata, flags, rc):
14.     if rc == 0:
15.         client.connected_flag = True
16.         print("Connected with result code " + str(rc))
17.     else:
18.         print("Bad connection returned code=", rc)
20. def on_disconnect(client, userdata, rc):
21.     client.connected_flag = False
22.     print("Disconnected with result code " + str(rc))
24. client.on_connect = on_connect
25. client.on_disconnect = on_disconnect
26. client.username_pw_set("your_user_name", "your_password")
27. client.connect(broker_address, port)
28. client.loop_start()
30. while not client.connected_flag:
31.     print("Connecting...")
32.     time.sleep(1)
34. print("Publishing message:", message)
35. result = client.publish(topic, message)
36. print("Publish Result:", result)

复制代码

代码要点：connected_flag 需要在 on_connect 回调中设置；publish 的 topic 格式为 channels/{channel_id}/publish/；message 必须符合 ThingSpeak 的 HTTP 表单格式（field1=xxx&field2=yyy）。

2. 使用 requests 调用 OpenWeatherMap API

HTTP GET 请求适合获取第三方服务的数据。下面的代码向 OpenWeatherMap 发送城市查询参数，并解析返回的 JSON，得到温度、天气和风速。

2. import requests
4. url = 'http://api.openweathermap.org/data/2.5/weather?q={city}&appid={apikey}'
5. params = {
6.     'q': 'London',
7.     'appid': 'your_api_key'
8. }
10. response = requests.get(url.format(**params)).json()
12. if response['cod'] == 200:
13.     temperature = round(float(response['main']['temp']) - 273.15, 2)
14.     description = response['weather'][0]['description'].title()
15.     windspeed = str(response['wind']['speed']) + 'm/s'
16.     print(f"Temperature in London is {temperature}°C and it's {description}. Wind speed is {windspeed}")
17. else:
18.     print("City not found or server error.")

复制代码

代码要点：OpenWeatherMap 默认返回开尔文温度，需要减 273.15 转换为摄氏度；response['cod'] 用于判断请求是否成功。生产环境中应将 API Key 放在环境变量或配置文件中，不要硬编码。

三、业务应用：智能投影仪与人脸识别眼镜

业务层负责将设备数据转化为用户可感知的内容，或控制硬件执行动作。

1. PyQt5 实现随时间变化的智能投影内容

下面代码使用 PyQt5 构建一个桌面窗口，定时显示问候语和当前时间。核心思想是利用 QTimer.singleShot 实现每分钟刷新一次。

2. import datetime
3. from PyQt5 import QtCore, QtGui, QtWidgets
5. class Ui_Form(object):
6.     def setupUi(self, Form):
7.         Form.setObjectName("Form")
8.         Form.resize(400, 300)
9.         self.label = QtWidgets.QLabel(Form)
10.         self.label.setGeometry(QtCore.QRect(10, 10, 381, 201))
11.         font = QtGui.QFont()
12.         font.setBold(True)
13.         self.label.setFont(font)
14.         self.label.setText("")
15.         self.label.setObjectName("label")
16.         self.retranslateUi(Form)
17.         QtCore.QMetaObject.connectSlotsByName(Form)
19.     def retranslateUi(self, Form):
20.         _translate = QtCore.QCoreApplication.translate
21.         Form.setWindowTitle(_translate("Form", "Smart Projection"))
23. class SmartProjection(QtWidgets.QWidget):
24.     def __init__(self, parent=None):
25.         super().__init__(parent)
26.         self._ui = Ui_Form()
27.         self._ui.setupUi(self)
28.         self.showTime()
30.     def showTime(self):
31.         now = datetime.datetime.now().strftime('%H:%M')
32.         if "06:00" <= now < "12:00":
33.             content = "Good morning! The weather outside looks very nice today."
34.         elif "12:00" <= now < "18:00":
35.             content = "Good afternoon! Enjoy the beautiful weather you have."
36.         else:
37.             content = "Good evening! Have a great day for sleep!"
39.         display = f"<h1>{content}</h1><p>The current date and time is: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>"
40.         self._ui.label.setText(display)
41.         QtCore.QTimer.singleShot(1000 * 60 * 1, self.showTime)
43. if __name__ == '__main__':
44.     app = QtWidgets.QApplication([])
45.     window = SmartProjection()
46.     window.show()
47.     app.exec_()

复制代码

代码要点：showTime 方法通过字符串比较判断时间段，并将 HTML 文本设置到 QLabel；再次调用 singleShot 实现递归定时，避免使用阻塞式 sleep。

2. OpenCV + face_recognition 实现人脸识别眼镜

智能眼镜常见的功能是识别视野中的人物。下面的代码从摄像头读取视频帧，使用 face_recognition 提取人脸特征并与已知人脸库比对，然后绘制边框和姓名。

2. import cv2
3. import numpy as np
4. import face_recognition
5. import threading
7. cap = cv2.VideoCapture(0)
8. cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
9. cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
11. # 读取人脸特征文件，每行格式：128维特征向量,姓名
12. known_face_encodings = []
13. known_face_names = []
14. with open('./faces.txt', 'r') as file:
15.     lines = file.readlines()
16.     for line in lines:
17.         encoding = line[:line.index(',')]
18.         name = line[line.index(',') + 1:-1].strip('\n').strip('\t')
19.         known_face_encodings.append(np.array(list(map(lambda x: float(x), encoding.split()))))
20.         known_face_names.append(name)
22. def recognize():
23.     global cap
24.     while True:
25.         ret, frame = cap.read()
26.         rgb_frame = frame[:, :, ::-1]
27.         face_locations = face_recognition.face_locations(rgb_frame)
28.         face_encodings = face_recognition.face_encodings(rgb_frame, face_locations)
29.         for i in range(len(face_encodings)):
30.             matches = face_recognition.compare_faces(known_face_encodings, face_encodings[i], tolerance=0.6)
31.             name = "Unknown"
32.             if True in matches:
33.                 matchedIdxs = [j for j, b in enumerate(matches) if b]
34.                 counts = {}
35.                 for matchIdx in matchedIdxs:
36.                     name = known_face_names[matchIdx]
37.                     counts[name] = counts.get(name, 0) + 1
38.                 name = max(counts, key=counts.get)
39.             top, right, bottom, left = face_locations[i]
40.             cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 255), 2)
41.             cv2.putText(frame, name, (left, top - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.75, (0, 255, 0), 2)
42.         cv2.imshow('Face Recognition', frame)
43.         if cv2.waitKey(1) & 0xFF == ord('q'):
44.             break
46. thread = threading.Thread(target=recognize)
47. thread.start()

复制代码

代码要点：faces.txt 中每行由 128 维特征向量和姓名组成，中间用逗号分隔；face_recognition.face_locations 返回人脸坐标，compare_faces 的 tolerance 参数控制匹配阈值，越小越严格。这里使用线程运行识别循环，避免阻塞主线程。

常见问题与调试建议

1. 蓝牙连接失败：确认目标设备 RFCOMM 通道号，通常为 1，但也可能是其他值；检查设备是否处于可发现状态。
2. MQTT 连接不稳定：在 on_connect 中检查 rc 返回值，0 表示成功；如果网络抖动，可增加自动重连逻辑。
3. API 返回 401：检查 API Key 是否有效，且请求 URL 中的参数名必须与文档一致。
4. face_recognition 运行速度慢：降低摄像头分辨率、减少识别频率或改用 GPU 版 dlib。

适用场景与扩展方向

上述代码可直接用于学习或原型验证：Wi-Fi RSSI 采集可用于室内定位；BLE 定位数据可用于移动设备跟踪；MQTT 上报适合温湿度、PM2.5 等传感器数据；HTTP 请求可对接各种云服务；人脸识别则可用于安防或智能门禁。开发者可以根据实际硬件协议调整数据格式和通信方式，将 Python 作为物联网网关或设备端的主要开发语言。

本文涉及的代码均基于原文示例整理，运行前请确认已安装对应依赖库（socket 为标准库，bluepy、paho-mqtt、requests、PyQt5、opencv-python、face_recognition 需单独安装）。

热心网友7

感谢楼主分享这么详细的实战代码，正好最近在搞一个蓝牙温湿度传感器的数据采集，bluepy那段很有参考价值。不过想请教一下，实际用RFCOMM的时候，BLE设备很多用GATT而不是RFCOMM，楼主这里是不是针对经典蓝牙或者特定模块？另外MQTT连ThingSpeak的示例里，publish后建议加个wait_for_publish或者loop，避免消息没发出去程序就退了。总体干货满满，收藏了慢慢研究。

热心网友7

感谢分享，代码很实用，正好最近在折腾物联网的数据上传。不过有两个小细节想提醒一下： 1. BLE 那个例子用的库名写的是 bluepy，但实际代码里导入的是 `bluetooth`，这是 PyBluez 的模块。bluepy 主要用来做 BLE GATT 通信，不走 RFCOMM；PyBluez 才支持经典蓝牙的 RFCOMM socket。如果真要用 bluepy 读 BLE 数据，一般是连接后通过 service/characteristic 来读取，不是直接 recv。建议把标题和库名统一一下，免得新手照抄时导入报错。 2. MQTT 发布那段，`client.publish()` 发完消息后最好调用 `client.loop()` 或 `client.loop_write()` 让消息真正发出去，或者直接 `loop_start()` 后稍等再断开。另外 ThingSpeak 的主题里那个 `1417` 是示例编号，实际要换成自己的 channel ID，不然会发到别人的频道去。 RSSI 解析那段用 `struct.unpack('&lt;b', ...)` 处理有符号单字节挺好，不过 `recvfrom` 收到的 data 可能包含多余内容，最好先对冒号后的部分做切片再取第一个字节，避免长度不匹配报错。整体思路很清晰，收藏了。

热心网友7

感谢分享！很实用的物联网开发实战教程，代码示例清晰，尤其是 UDP 解析 RSSI 和蓝牙 RFCOMM 读取定位这部分，对刚入门 IoT 的开发者很有帮助。 有一点想请教：MQTT 示例里 ThingSpeak 的 topic 和消息格式看起来是不是少了 API key 或者 channel 的完整路径？另外 `field1=%d&field2=%d` 用的是时间戳和随机数，实际用传感器数据替换时需要注意类型转换吗？ 期待看到楼主后面关于 HTTP 接入和业务应用的代码，正好最近在做类似的项目，想多参考一下。