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

脚本专家 发表于 7 天前

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

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

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

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

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

import socket
import struct

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(('localhost', 8889))

while True:
data, addr = s.recvfrom(1024)
# 数据格式假设为：RSSI:-45，这里取出冒号后的部分，按有符号单字节解析
rssi = struct.unpack('<b', data)
print("Received:", rssi, "dBm from", addr)

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

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

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

import bluetooth
import sys

address = "F7:CA:E6:C1:D4:B9"# 替换为目标设备地址

try:
sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
sock.connect((address, 1))
while True:
   data = sock.recv(1024).decode()
   if len(data) > 0:
         items = data.split(',')
         lat = float(items) # 纬度
         lng = float(items) # 经度
         altitude = int(items) # 高度
         print("Latitude:", lat, "Longitude:", lng, "Altitude:", altitude, "m")
finally:
try:
   sock.close()
except NameError:
   pass

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

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

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

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

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

import paho.mqtt.client as mqtt
import time
import random

broker_address = "mqtt.thingspeak.com"
port = 1883
topic = "channels/1417/publish/"
message = "field1=%d&field2=%d" % (time.time(), int(random.randint(0, 10) * 10))

client = mqtt.Client()

def on_connect(client, userdata, flags, rc):
if rc == 0:
   client.connected_flag = True
   print("Connected with result code " + str(rc))
else:
   print("Bad connection returned code=", rc)

def on_disconnect(client, userdata, rc):
client.connected_flag = False
print("Disconnected with result code " + str(rc))

client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.username_pw_set("your_user_name", "your_password")
client.connect(broker_address, port)
client.loop_start()

while not client.connected_flag:
print("Connecting...")
time.sleep(1)

print("Publishing message:", message)
result = client.publish(topic, message)
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，得到温度、天气和风速。

import requests

url = 'http://api.openweathermap.org/data/2.5/weather?q={city}&appid={apikey}'
params = {
'q': 'London',
'appid': 'your_api_key'
}

response = requests.get(url.format(**params)).json()

if response['cod'] == 200:
temperature = round(float(response['main']['temp']) - 273.15, 2)
description = response['weather']['description'].title()
windspeed = str(response['wind']['speed']) + 'm/s'
print(f"Temperature in London is {temperature}°C and it's {description}. Wind speed is {windspeed}")
else:
print("City not found or server error.")

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

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

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

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

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

import datetime
from PyQt5 import QtCore, QtGui, QtWidgets

class Ui_Form(object):
def setupUi(self, Form):
   Form.setObjectName("Form")
   Form.resize(400, 300)
   self.label = QtWidgets.QLabel(Form)
   self.label.setGeometry(QtCore.QRect(10, 10, 381, 201))
   font = QtGui.QFont()
   font.setBold(True)
   self.label.setFont(font)
   self.label.setText("")
   self.label.setObjectName("label")
   self.retranslateUi(Form)
   QtCore.QMetaObject.connectSlotsByName(Form)

def retranslateUi(self, Form):
   _translate = QtCore.QCoreApplication.translate
   Form.setWindowTitle(_translate("Form", "Smart Projection"))

class SmartProjection(QtWidgets.QWidget):
def __init__(self, parent=None):
   super().__init__(parent)
   self._ui = Ui_Form()
   self._ui.setupUi(self)
   self.showTime()

def showTime(self):
   now = datetime.datetime.now().strftime('%H:%M')
   if "06:00" <= now < "12:00":
         content = "Good morning! The weather outside looks very nice today."
   elif "12:00" <= now < "18:00":
         content = "Good afternoon! Enjoy the beautiful weather you have."
   else:
         content = "Good evening! Have a great day for sleep!"

   display = f"<h1>{content}</h1><p>The current date and time is: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>"
   self._ui.label.setText(display)
   QtCore.QTimer.singleShot(1000 * 60 * 1, self.showTime)

if __name__ == '__main__':
app = QtWidgets.QApplication([])
window = SmartProjection()
window.show()
app.exec_()

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

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

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

import cv2
import numpy as np
import face_recognition
import threading

cap = cv2.VideoCapture(0)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)

# 读取人脸特征文件，每行格式：128维特征向量,姓名
known_face_encodings = []
known_face_names = []
with open('./faces.txt', 'r') as file:
lines = file.readlines()
for line in lines:
   encoding = line[:line.index(',')]
   name = line.strip('\n').strip('\t')
   known_face_encodings.append(np.array(list(map(lambda x: float(x), encoding.split()))))
   known_face_names.append(name)

def recognize():
global cap
while True:
   ret, frame = cap.read()
   rgb_frame = frame[:, :, ::-1]
   face_locations = face_recognition.face_locations(rgb_frame)
   face_encodings = face_recognition.face_encodings(rgb_frame, face_locations)
   for i in range(len(face_encodings)):
         matches = face_recognition.compare_faces(known_face_encodings, face_encodings, tolerance=0.6)
         name = "Unknown"
         if True in matches:
            matchedIdxs =
            counts = {}
            for matchIdx in matchedIdxs:
               name = known_face_names
               counts = counts.get(name, 0) + 1
            name = max(counts, key=counts.get)
         top, right, bottom, left = face_locations
         cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 255), 2)
         cv2.putText(frame, name, (left, top - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.75, (0, 255, 0), 2)
   cv2.imshow('Face Recognition', frame)
   if cv2.waitKey(1) & 0xFF == ord('q'):
         break

thread = threading.Thread(target=recognize)
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 发表于 7 天前

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

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

热心网友7 发表于 7 天前

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

感谢分享，代码很实用，正好最近在折腾物联网的数据上传。不过有两个小细节想提醒一下： 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('<b', ...)` 处理有符号单字节挺好，不过 `recvfrom` 收到的 data 可能包含多余内容，最好先对冒号后的部分做切片再取第一个字节，避免长度不匹配报错。整体思路很清晰，收藏了。

热心网友7 发表于 7 天前

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

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

页: [1]

红客联盟 - 由08小组运营's Archiver