物聯網iot、mqtt協議與華為云平臺的綜合實踐（萬字0基礎保姆級教程）

????????本學期的物聯網技術與應用課程，其結課設計內容包含：mqtt、華為云、PyQT5和MySQL等結合使用，完成了從華為云配置產品信息以及轉發規則，到mqtt命令轉發，再到python編寫邏輯代碼實現相關功能，最后用PyQT5實現面向用戶的可視化操作界面，博主將其中的詳細流程進行了一個整理和分享，希望可以幫助到大家。

一、華為云平臺配置

引言

接下來的內容是基于已經注冊華為云賬號

首先進入華為云官網，選擇產品 -> IoT物聯網 -> 設備接入IoTDA

進入該界面，選擇 免費試用 創建實例，其步驟較簡單，大家可以參考其他資料

創建好實例后點擊進入

認證后，華為云獲取ak/sk的步驟文檔：

獲取AK/SK_AK/SK簽名認證操作指導_API簽名指南_API簽名指南-華為云 (huaweicloud.com)https://support.huaweicloud.com/devg-apisign/api-sign-provide-aksk.html

測試ak/sk的python代碼，命名為：ak_sk_data

# -*- coding: utf-8 -*-def get_ak():ak = ' 'return akdef get_sk():sk = ' 'return skdef print_hello(name):print('{}歡迎您！'.format(name))def print_hello2(name='智能溫度系統'):print('{}歡迎您！'.format(name))if __name__=='__main__':print(get_ak())print(get_sk())print_hello('智能溫度系統')print_hello2()

1.1 創建產品

創建好實例后，點擊左側導航欄，點擊 創建產品

創建產品的具體參數，選擇MQTT協議，JSON格式

1.2 添加服務

創建產品后，點擊?詳情

進入產品詳情界面，點擊 自定義產品（此處定義產品后，界面改變，故產品不一致）

點擊后為此界面，根據需求定義服務的各參數

根據產品功能設置其對應的?屬性（MQTT傳輸的參數），此處為空氣溫度，數據類型設置為 可讀可寫 ，后續需要通過MQTT寫入

根據以上步驟配置后，如圖所示

1.3 添加設備

產品創建后，創建設備，點擊左側導航欄設備 ->? 所以設備 -> 注冊設備

注冊設備參數如下所示，設備會繼承所屬產品的屬性，自定義密鑰

點擊確定后，顯示設備創建成功

1.4 制定規則

接下來我們制定規則，以此實現MQTT數據轉發

點擊左側導航欄規則 -> 數據轉發 -> 創建規則

首先設置轉發數據基本信息，數據來源選擇 設備屬性 ，觸發事件選擇 設備屬性上報

設置添加轉發目標，轉發目標選擇 MQTT推送消息隊列 ，自定義推送Topic名稱，一般是與產品設備相關

最好創建啟動規則即可

二、mqtt的使用?

2.1 mqtt三元組

通過mqtt三元組與華為云平臺設備連接，以下是獲取網址：

Huaweicloud IoTDA Mqtt ClientId GeneratorSource code generated using layoutit.comhttps://iot-tool.obs-website.cn-north-4.myhuaweicloud.com/Deviced是設備id，DeviceSecret是創建設備時設置的密碼

此外還需獲取接入信息的 MQTTS 的接入地址，如下圖所示：

左側導航欄：總覽 -> 接入信息 ->?設備接入 -> MQTT接入地址

2.2 mqtt配置

接下來我們以MQTT.fx為例進行配置

自定義Proflie Name名稱，Profile Type選擇 MQTT Broker

Broker Address粘貼 MQTTS接入地址，Broker Port選擇 1883

后面三個依次填入三元組生成的信息

在配置好后保存，回到首頁，點擊Connect進行連接

若連接成功，右側會解鎖亮綠燈，且華為云平臺也會亮燈顯示設備在線，若失敗則紅燈

2.3 命令發送

在通過MQTT.fx成功與華為云平臺連接后，我們查看參數，準備進行命令下發

如下圖所示，返回產品詳情，服務為 Bulb，其屬性參數為 brightness 和 color_temp

接下來，我們配置數據上報

$oc/devices/{設備id}/sys/properties/report

{"services": [{"service_id": "XXX","properties": {"屬性名稱1": XXX,"屬性名稱2": XXX}}]
}

如圖填入后，點擊Publish進行上報

左側導航欄選擇設備，接入詳情刷新即可查看上報的數據信息

如上圖所示，已經成功上報數據

三、MySQL的創建

在數據庫內創建數據表并設置相關參數

-- 創建數據庫
CREATE DATABASE IF NOT EXISTS XXX;-- 使用數據庫
USE XXX;-- 創建 XXX 表（帶自動更新時間）
CREATE TABLE IF NOT EXISTS air (XXX INT,XXX INT,student_id INT DEFAULT 123,student_name VARCHAR(50) DEFAULT '小明',-- 自動更新時間record_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

四、PyQT5的配置與簡單使用

4.1 創建虛擬環境

可以先創建一個虛擬環境，專門用于此項目庫和包的下載安裝

詳情可以參考此博客，主頁也包含內核切換和pycharm配置版本conda等常見問題anaconda虛擬環境的安裝_vmware安裝anaconda-CSDN博客文章瀏覽閱讀428次。本文指導如何在Windows系統中通過AnacondaPrompt設置PyTorch的CPU或GPU版本環境，遇到問題時推薦觀看土堆教程的詳細視頻鏈接。https://blog.csdn.net/m0_73544543/article/details/137649088?spm=1001.2014.3001.5502

4.2 下載pyqt5相關包

下載pyqt5

pip install PyQt5-i https://pypi.tuna.tsinghua.edu.cn/simple

下載PyQt5 Tools

pip install PyQt5-tools-i https://pypi.tuna.tsinghua.edu.cn/simple

下載PyQtChart

pip install PyQtChart

下載PyMySQL

pip install PyMySQL

下載pyserial

pip install pyserial

4.3 Pycharm配置PyQT5相關工具

在pycharm中點擊 File，選擇 Settings 進入設置

選擇工具 Tools，選擇 External Tools

新建工具，點擊 +號，選擇program啟動程序的路徑

找到安裝路徑，例如

其他PyUIC以及PyRcc插件同理，注意后兩框的填入

$FileName$-o $FileNameWithoutExtension$.py
$FileDir$

4.4 使用QT Designer

菜單欄點擊 Tools -> External Tools -> Qt Designer

啟動后創建項目即可，博客在此選擇的是 Widget 進行項目創建，左側是控件可直接進行拖拽，使用方法類似于Java GUI，對應著學習即可

4.5 使用PyUIC

在保存QT Designer的項目后，會生成一個ui文件，此時需要通過PyUIC將ui文件轉化為py文件

右擊ui文件，選擇External Tools，點擊PyUIC等待轉換即可，會生成一個同名的py文件

4.6 使用PyRcc

在項目文件夾下創建一個image文件夾，將圖片文件放入，并創建一個qrc文件

格式如下：

<RCC><qresource prefix="xxx"><file>XXX.png</file></qresource>
</RCC>

創建完整后通過PyRcc工具進行轉化，步驟同PyUIC

將qrc文件轉化為可識別的py文件，將圖片轉為十六進制格式，如圖所示

然后在QT中使用，在資源瀏覽器中選擇畫筆

進入編輯資源中，點擊紅框按鈕，在文件夾中選擇相應的qrc文件

再按照下圖對應順序配置圖片路徑

在配置完成后進行使用，例如我們在此使用label控件，右擊選擇 改變多信息文本

選擇圖片資源，選擇配置的圖片即可

4.7 測試程序

我們通過以下代碼進行測試：

 from PyQt5 import QtWidgets, QtGuiimport sysapp = QtWidgets.QApplication(sys.argv)window = QtWidgets.QWidget();window.show()sys.exit(app.exec_())

測試成功

五、測試

5.1 影子設備調試

此外我們可以通過影子設備進行調試，我們在搜索框搜索：API Explorer，選擇 設備接入IOTDA

進入后，在設備接入選擇設備影子下的查詢設備影子數據，選擇只看必選項

instance_id為實例id，project_id為“我的憑證”中的項目id，device_id為設備id

在配置成功后再次進行數據上報可得，右側為詳細信息

5.2 代碼測試python與華為云連接

在測試前，想要安裝相應的庫實現python連接華為云

安裝核心庫

pip install huaweicloudsdkcore

安裝IoTDA服務庫

pip install huaweicloudsdkiotda

華為云提供的官方使用指導指南：

應用側Python SDK使用指南_應用側SDK_SDK參考_設備接入 IoTDA-華為云物聯網平臺提供Python語言的應用側SDK供開發者使用。本文介紹應用側 Python SDK的安裝和配置，及使用Python SDK調用應用側API的示例。訪問Python官網，下載并按說明安裝Python開發環境。華為云應用側 Python SDK 支持 Python3 及以上版本。訪問pip官網，下載并按說明安裝pip工具。執行如下https://support.huaweicloud.com/sdkreference-iothub/iot_10_10003.html根據官網提供的模板：

各參數填入的具體步驟在上方 # 號中給出

# -*- coding: utf-8 -*-
import ak_sk_data
import json
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcore.region.region import Region
from huaweicloudsdkiotda.v5 import *
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkcore.auth.credentials import DerivedCredentialsif __name__ == "__main__":# 認證用的ak和sk直接寫到代碼中有很大的安全風險，建議在配置文件或者環境變量中密文存放，使用時解密，確保安全；# 本示例以ak和sk保存在環境變量中為例，運行本示例前請先在本地環境中設置環境變量HUAWEICLOUD_SDK_AK和HUAWEICLOUD_SDK_SK。ak = ak_sk_data.get_ak()sk = ak_sk_data.get_sk()#右上角用戶，選擇 我的憑證 - API憑證 - 項目ID（所屬區域對應）project_id = " "# region_id：如果是上海一，請填寫"cn-east-3"；如果是北京四，請填寫"cn-north-4"；如果是華南廣州，請填寫"cn-south-1"region_id = "cn-north-4"# endpoint：請在控制臺的"總覽"界面的"平臺接入地址"中查看"應用側"的https接入地址endpoint = " "# 標準版/企業版：需自行創建Region對象REGION = Region(region_id, endpoint)# 創建認證# 創建BasicCredentials實例并初始化credentials = BasicCredentials(ak, sk, project_id)# 標準版/企業版需要使用衍生算法，基礎版請刪除配置"with_derived_predicate"credentials.with_derived_predicate(DerivedCredentials.get_default_derived_predicate())# 基礎版：請選擇IoTDAClient中的Region對象 如： .with_region(IoTDARegion.CN_NORTH_4)# 標準版/企業版：需要使用自行創建的Region對象# 配置是否忽略SSL證書校驗， 默認不忽略： .with_http_config(HttpConfig(ignore_ssl_verification=True)) \client = IoTDAClient.new_builder() \.with_credentials(credentials) \.with_region(REGION) \.build()try:'''# 實例化請求對象request = ListDevicesRequest()# 調用查詢設備列表接口response = client.list_devices(request)'''request = ShowDeviceShadowRequest()#設備idrequest.device_id = " "response = client.show_device_shadow(request)# 將響應對象轉換為字典response_dict = json.loads(str(response))# 提取shadow數據shadow_data = response_dict["shadow"][0]# 填入產品屬性參數# 例如博主是提取brightness,  color_tempbrightness = shadow_data["reported"]["properties"]["brightness"]color_temp = shadow_data["reported"]["properties"]["color_temp"]# 打印提取的值print("完整的響應數據:", response_dict)print("亮度值(brightness):", brightness)print("色溫值(color_temp):", color_temp)except exceptions.ClientRequestException as e:print(e.status_code)print(e.request_id)print(e.error_code)print(e.error_msg)

運行結果為：

5.3 代碼測試mqtt與華為云連接

安裝mqtt的依賴

pip install paho-mqtt==2.0.0

根據華為云文檔構建 ClientConf、MqttClient、MqttDemo三個py文件

Python Demo使用說明_使用MQTT轉發_數據轉發至第三方應用_規則引擎_用戶指南_設備接入 IoTDA-華為云本文以Python語言為例，介紹應用通過MQTTS協議接入平臺，接收服務端訂閱消息的示例。熟悉Python語言開發環境配置，熟悉Python語言基本語法。本示例使用了Python 3.8.8版本。本示例使用的Python語言的Mqtt依賴為paho-mqtt(本示例使用版本為2.0.0)，可以通過以下命令下載依賴。ClientConf代碼https://support.huaweicloud.com/usermanual-iothub/iot_01_00115.html

其中MQTTS8883接入憑證獲取方式為：

MQTT客戶端接入說明_使用MQTT轉發_數據轉發至第三方應用_規則引擎_用戶指南_設備接入 IoTDA-華為云在調用創建規則觸發條件、創建規則動作和修改規則觸發條件配置并激活規則后，您需要參考本文將MQTT客戶端接入物聯網平臺，成功接入后，在您的服務端運行MQTT客戶端，即可接收訂閱的消息。MQTT客戶端接入物聯網平臺的連接地址和連接認證參數說明如下：MQTT接入域名每個賬號會自動生成，請前往控制臺-接入信息頁面獲取。接入信息-應用側MQTT接入https://support.huaweicloud.com/usermanual-iothub/iot_01_00113.html其中MqttDemo的具體配置為：

from ClientConf import ClientConf
from MqttClient import MqttClient
import os
from typing import Optionaldef main():client_conf = ClientConf()# host-id:總覽 -> 接入信息 -> 應用接入 -> MQTTS(8883)client_conf.host = " "client_conf.port = 8883# 規則 -> 數據轉發 -> 詳情 -> 2設置轉發目標 -> 配置:推送Topicclient_conf.topic = " "# MQTTS8883接入憑證的key和codeclient_conf.access_key = " "client_conf.access_code = " "#instance_id：總覽 -> 詳情 -> 左上角IDclient_conf.instance_id = " "mqtt_client = MqttClient(client_conf)if mqtt_client.connect() != 0:print("init failed")returnif __name__ == "__main__":main()

運行結果如下圖所示，成功連接

連接成功后，我們再次提供MQTT.fx進行數據上報（命令在2.3 命令發送中有模板）

發送后，成功接收，實時更新數據

5.4 代碼測試mysql連接

在上述代碼基礎上完善

在 MqttClient.py 中修改 on_message 方法

    def _on_message(self, client, userdata, message: mqtt.MQTTMessage):msg = message.payload.decode()print("topic " + self.__topic + " Received message: " + msg)# 解析 JSON 數據try:data = json.loads(msg)properties = data["notify_data"]["body"]["services"][0]["properties"]# 提取需要的參數smoke_value = properties.get("smoke_value")# 如果有回調函數，則調用（檢查是否存在）if hasattr(self, "on_message_callback") and self.on_message_callback:self.on_message_callback({"smoke_value": smoke_value})except Exception as e:print("Failed to parse message:", e)

創建 mysqlset.py 建立數據庫連接

import pymysqldef get_connection(database_name):try:connection = pymysql.connect(host='127.0.0.1',user=' ',password=' ',database=' ',  # 使用傳入的數據庫名charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)return connectionexcept pymysql.MySQLError as e:print("數據庫連接失敗:", e)return None

在 MqttDemo.py 文件中添加 mysql 語句，實現mqtt發送數據插入數據庫

from ClientConf import ClientConf
from MqttClient import MqttClient
import os
from typing import Optional
from mysqlset import get_connectiondef handle_message(data):print("Received data:")print("smoke_value:", data["smoke_value"])# 插入數據庫insert_smoke_data(data)def insert_smoke_data(data):"""插入數據到 smoke 表"""conn = Nonetry:conn = get_connection("smoke")  # 替換為你的數據庫名cursor = conn.cursor()# SQL 插入語句sql = """INSERT INTO smoke (smoke_value)VALUES (%s)"""# 執行 SQLcursor.execute(sql, (data["smoke_value"]))conn.commit()print("數據插入成功！")except Exception as e:print("數據庫插入失敗:", e)finally:if conn:conn.close()
def main():# 同上client_conf = ClientConf()client_conf.host = " "client_conf.port = 8883client_conf.topic = " "client_conf.access_key = " "client_conf.access_code = " "client_conf.instance_id = " "mqtt_client = MqttClient(client_conf)mqtt_client.on_message_callback = handle_message  # 設置回調函數if mqtt_client.connect() != 0:print("init failed")returnif __name__ == "__main__":main()

運行結果為

六、相關功能實現

自此所有平臺的連接已經實現，接下來給大家分享一下博主在此基礎上構建的結課設計項目

6.1 概述

6.1.1?項目內容

????????根據對物聯網架構的理解，依據物聯網開發流程，基于華為云平臺，實現設備側、平臺側及應用側的開發。根據既定的物聯網方案設計，自行選用適合的開發技術，搭建實驗環境，設計并實現一個物聯網應用系統。

????????提升要求：能夠根據所選定應用場景，調研實際需求，設計符合需要的邏輯合理的功能業務，能夠利用人工智能技術智能化處理數據或者大模型的智能功能，具有一定創新性。

6.1.2 項目要求

????????本次項目根據所選用場景進行設備選型，采用 Mqtt.fx 工具模擬終端設備向華為云平臺實現數據上報并進行數據訂閱（設備側采用真實硬件更佳）。采用華為云物聯網平臺應用側的接口獲取設備側上報到平臺的設備影子數據，利用 mqtt 協議實現數據轉發，并能夠進行設備管理等功能。

????????可選用Python、Web、Android等技術實現圖形化交互界面設計。

6.1.3 項目方案（實現功能）

????????基于Python/web/Android 的環境安全監控管理應用系統本系統需實現兩個及以上的設備（煙感模塊及自定義模塊）的管理：參考功能如下：

????????1 實現華為云平臺煙感模塊及自定義模塊的產品創建和設備注冊（含自己的學號）。 2 通過Mqtt.fx模擬煙感設備及自定義設備進行數據上報。 3 設計基于 pyqt5 的應用界面，展示云平臺設備影子數據獲取，實現設備管理，展示設備列表，利用數據轉發功能實現數據實時獲取等。 4 煙感及自定義模塊異常數據處理：應用界面可以展示設備情況，如正常與否，亦可通過應用界面報警。 5 通過應用界面手動下發報警命令，設備側能夠接收到報警命令。

????????6 煙感及自定義設備異常數據存儲及查看。（選做） 7 利用人工智能技術使應用具有智能化處理功能。（選做） 8.自定義。

6.2 項目介紹

????????博主在基于上述項目內容和要求，實現除訂閱下發報警命令外所有功能

????????本系統采用模塊化設計，主要包含四大功能模塊：環境監測模塊、數據分析模塊、異常預警模塊和視覺識別模塊。

????????環境監測模塊負責實時采集空氣溫濕度、光照強度以及土壤溫濕度、PH值、EC值等多維度數據，通過傳感器網絡實現全天候監控；數據分析模塊提供數據可視化功能，支持折線圖、柱狀圖、餅圖等多種圖表展示，并具備歷史數據查詢和條件篩選能力，幫助用戶全面掌握農業生產環境變化趨勢。

????????異常預警模塊基于預設閾值和智能算法，自動檢測環境參數異常，當溫度、濕度等關鍵指標超出安全范圍時，系統會觸發聲光報警機制，并通過可視化界面突出顯示異常數據。視覺識別模塊采用YOLOv8目標檢測算法，可實時識別監控畫面中的特定物體（如農藥瓶、玻璃容器等），當檢測到目標且置信度超過60%時，系統會立即啟動蜂鳴器報警，有效預防潛在風險。各模塊通過統一的數據總線進行通信，確保系統各部分協同工作，共同構成完整的智慧農業管理解決方案。

6.3 項目展示

項目首頁：

數據監測界面：

異常數據監測界面：

在線識別界面：

關于我們：

6.4 部分功能實現代碼

異常數據閾值判定

def filter_abnormal_data(table_type):"""通用異常數據過濾函數:param table_type: 'air' 或 'soil':return: 插入的記錄數"""# 定義各表的異常閾值thresholds = {'air': {'temp': 70, 'humi': 70, 'lux': 70},'soil': {'temp': 70, 'humi': 70, 'ph': 70, 'ec': 70, 'co2': 70}}try:connection = mysqlset.get_connection('agro')if not connection:raise Exception("數據庫連接失敗")with connection.cursor() as cursor:# 1. 檢查目標表是否存在target_table = f'error_{table_type}'cursor.execute(f"SHOW TABLES LIKE '{target_table}';")if not cursor.fetchone():cursor.execute(f"SHOW CREATE TABLE {table_type};")create_sql = cursor.fetchone()['Create Table'].replace(f'CREATE TABLE `{table_type}`',f'CREATE TABLE `{target_table}`')cursor.execute(create_sql)# 2. 構建動態SQL條件conditions = " OR ".join([f"{field} > {threshold}"for field, threshold in thresholds[table_type].items()])# 3. 執行數據過濾插入sql = f"""INSERT INTO {target_table}SELECT * FROM {table_type} WHERE ({conditions})AND id NOT IN (SELECT id FROM {target_table})"""cursor.execute(sql)connection.commit()return cursor.rowcountexcept Exception as e:print(f"過濾{table_type}數據失敗:", e)if connection:connection.rollback()return 0finally:if connection:connection.close()

部分可視化圖表

  def create_pie_chart_view(self, parent=None):"""創建餅狀圖，顯示各參數占比"""soil_data = self.get_all_soil_data()if not soil_data:return QtWidgets.QLabel("無數據可用",parent)latest_data = soil_data[0]series = QPieSeries()# 使用原始值series.append(f"溫度\n{latest_data['temp']}°C", float(latest_data['temp']))series.append(f"濕度\n{latest_data['humi']}%", float(latest_data['humi']))series.append(f"PH值\n{latest_data['ph']}", float(latest_data['ph']))series.append(f"EC值\n{latest_data['ec']}", float(latest_data['ec']))series.append(f"CO2\n{latest_data['co2']}ppm", float(latest_data['co2']))# 設置切片顏色colors = [QtGui.QColor(255, 0, 0),  # 紅-溫度QtGui.QColor(0, 0, 255),  # 藍-濕度QtGui.QColor(0, 180, 0),  # 綠-PH值QtGui.QColor(180, 0, 180),  # 紫-EC值QtGui.QColor(255, 165, 0)  # 橙-CO2]for i, slice in enumerate(series.slices()):slice.setColor(colors[i])slice.setLabelVisible(True)slice.setLabel(f"{slice.label()} ({slice.percentage():.1f}%)")slice.setLabelArmLengthFactor(0.2)chart = QChart()chart.addSeries(series)chart.setTitle("土壤參數占比")chart.legend().setVisible(True)chart.legend().setAlignment(Qt.AlignBottom)chart_view = QChartView(chart,parent)chart_view.setRenderHint(QtGui.QPainter.Antialiasing)return chart_view

接收設備參數

    def _on_message(self, client, userdata, message: mqtt.MQTTMessage):msg = message.payload.decode()print("topic " + self.__topic + " Received message: " + msg)# 解析 JSON 數據try:data = json.loads(msg)properties = data["notify_data"]["body"]["services"][0]["properties"]# 提取需要的參數air_data = {"temp": properties.get("air_temp"),"humi": properties.get("air_humi"),"lux": properties.get("lux")}soil_data = {"temp": properties.get("soil_temp"),"humi": properties.get("soil_humi"),"ph": properties.get("ph"),"ec": properties.get("ec"),"co2": properties.get("co2")}# 如果有回調函數，則調用if hasattr(self, "on_message_callback") and self.on_message_callback:self.on_message_callback({"air_data": air_data,"soil_data": soil_data})except Exception as e:print("Failed to parse message:", e)traceback.print_exc()

構建一個過濾器，檢查接收的消息是否符合表中格式，實現air和soil相關參數的區分

    def insert_soil_data(data):"""插入土壤數據到 soil 表"""conn = Nonetry:# 檢查是否有有效的土壤數據if data["temp"] is None and data["humi"] is None and data["ph"] is None and data["ec"] is None and data["co2"] is None:print("土壤數據全為空，跳過插入")return Falseconn = get_connection("agro")cursor = conn.cursor()sql = """INSERT INTO soil (temp, humi, ph, ec, co2, record_time)VALUES (%s, %s, %s, %s, %s, %s)"""cursor.execute(sql, (data["temp"],data["humi"],data["ph"],data["ec"],data["co2"],datetime.now()))conn.commit()print("土壤數據插入成功！")return Trueexcept Exception as e:if conn:conn.rollback()print(f"土壤數據插入失敗: {e}\n數據內容: {data}")return Falsefinally:if conn:conn.close()

搜索功能

 def search_air_data(self, field, keyword):if field not in self.field_map:return []try:with self.connection.cursor() as cursor:sql = f"""SELECT id, temp, humi, lux, student_id, student_name, record_time FROM air WHERE {self.field_map[field]} LIKE %s"""cursor.execute(sql, (f"%{keyword}%",))return cursor.fetchall()except Exception as e:print("搜索數據失敗:", e)return []

異常數據觸發蜂鳴器報警

    def play_beep_sound():"""播放蜂鳴聲（兼容性改進）"""try:current_time = time.time()# 只有當有新異常數據且超過間隔時間時才播放if (current_time - ErrorDataService.last_alert_time < ErrorDataService.ALERT_INTERVALor not ErrorDataService.has_new_errors):returnErrorDataService.last_alert_time = current_timeErrorDataService.has_new_errors = False  # 重置標志位system = platform.system()if system == "Windows":try:import winsoundwinsound.Beep(1000, 3000)  # 嘗試硬件蜂鳴器except:# 改用系統提示音import winsoundwinsound.PlaySound("SystemExclamation", winsound.SND_ALIAS)elif system == "Darwin":os.system('afplay /System/Library/Sounds/Ping.aiff')else:os.system('play -nq -t alsa synth 3 sine 1000')except Exception as e:print("播放蜂鳴聲失敗:", e)

目標檢測與實時報警

    def _update_frame(self):"""內部方法：處理每一幀"""if self.cap is None or not self.cap.isOpened():returnret, frame = self.cap.read()if not ret:self.result_signal.emit("無法從攝像頭獲取幀")return# YOLO 檢測try:results = self.model(frame, conf=0.6)annotated_frame = results[0].plot()  # 帶檢測框的圖像# 處理檢測結果should_alarm = False  # 標記是否需要報警for r in results:for box in r.boxes:class_name = self.model.names[int(box.cls)]conf = box.conf.item()result_text = f"檢測到: {class_name}, 置信度: {conf:.2f}"self.result_signal.emit(result_text)# 檢查是否需要報警if (class_name in ['bottle', 'wine glass']) and conf > 0.6:should_alarm = True# 觸發或停止蜂鳴器if should_alarm and not self.alarm_triggered:winsound.Beep(1000, 500)  # 頻率1000Hz，持續500msself.alarm_triggered = Trueself.result_signal.emit("檢測到目標對象，觸發報警！")elif not should_alarm and self.alarm_triggered:self.alarm_triggered = False

格式轉化

 # 轉換圖像格式并發送信號rgb_image = cv2.cvtColor(annotated_frame, cv2.COLOR_BGR2RGB)h, w, ch = rgb_image.shapebytes_per_line = ch * wqt_image = QImage(rgb_image.data, w, h, bytes_per_line, QImage.Format_RGB888)pixmap = QPixmap.fromImage(qt_image)self.update_signal.emit(pixmap)except Exception as e:self.result_signal.emit(f"處理幀時出錯: {str(e)}")

????????部分代碼給大家提供一種思路，詳細的教程以及步驟提供給大家，希望本篇博客能夠幫助到大家，如果對大家有所幫助，希望可以給博主來個三連