2025年P(guān)ython物聯(lián)網(wǎng)數(shù)據(jù)采集培訓(xùn)試卷：實(shí)戰(zhàn)操作訓(xùn)練考試時(shí)間：______分鐘總分：______分姓名：______一、環(huán)境與庫(kù)準(zhǔn)備1.請(qǐng)簡(jiǎn)述為完成本次數(shù)據(jù)采集任務(wù)，你需要配置哪些Python環(huán)境或安裝哪些關(guān)鍵庫(kù)（請(qǐng)列舉至少5個(gè)），并說(shuō)明其中一個(gè)庫(kù)的主要用途。2.假設(shè)你使用的是樹(shù)莓派，請(qǐng)寫出至少三行Python代碼，用于導(dǎo)入`paho-mqtt`庫(kù)并創(chuàng)建一個(gè)MQTT客戶端實(shí)例，該實(shí)例的客戶端ID為`my_raspberrypi_client`，連接地址為``，端口為1883。二、硬件數(shù)據(jù)采集3.假設(shè)你連接了一個(gè)DHT11溫濕度傳感器到樹(shù)莓派的GPIO4（BCM編碼）引腳，請(qǐng)寫出使用`gpiozero`庫(kù)讀取該傳感器數(shù)據(jù)的Python代碼片段。你需要包含導(dǎo)入庫(kù)、創(chuàng)建傳感器對(duì)象以及讀取一次溫度和濕度的代碼。4.假設(shè)你使用的是ESP32開(kāi)發(fā)板，通過(guò)I2C協(xié)議連接了一個(gè)MPX4115氣壓傳感器（I2C地址為0x28）。請(qǐng)寫出使用`machine`庫(kù)（樹(shù)莓派可用`smbus`庫(kù)）初始化I2C總線和創(chuàng)建MPX4115傳感器對(duì)象的Python代碼片段。5.在讀取傳感器數(shù)據(jù)時(shí)，可能會(huì)遇到串口超時(shí)或I2C通信錯(cuò)誤。請(qǐng)寫出一段Python代碼，嘗試讀取DHT11傳感器數(shù)據(jù)三次，如果讀取失?。僭O(shè)函數(shù)返回None或特定錯(cuò)誤碼），則打印錯(cuò)誤信息并等待1秒后重試。三、數(shù)據(jù)處理與存儲(chǔ)6.假設(shè)你從傳感器獲取到了一組溫度數(shù)據(jù)（單位為攝氏度）`[22.5,23.1,22.8,23.4,23.0,22.9,23.2]`。請(qǐng)寫出Python代碼，計(jì)算這組數(shù)據(jù)的平均值和標(biāo)準(zhǔn)差，并將結(jié)果分別存儲(chǔ)在變量`temp_mean`和`temp_std`中。7.請(qǐng)寫出Python代碼，將問(wèn)題6中計(jì)算得到的`temp_mean`和`temp_std`存儲(chǔ)到一個(gè)名為`sensor_data.csv`的CSV文件中。文件應(yīng)包含兩列標(biāo)題：`Temperature`和`Statistics`，第一行數(shù)據(jù)為平均值，第二行數(shù)據(jù)為標(biāo)準(zhǔn)差。四、數(shù)據(jù)傳輸8.請(qǐng)完善下面的Python代碼，使其能夠連接到MQTTBroker``，并向主題`python/io/temperature`發(fā)布問(wèn)題6計(jì)算得到的溫度平均值`temp_mean`（假設(shè)`temp_mean`變量已定義并包含計(jì)算結(jié)果，類型為float）。請(qǐng)確保包含錯(cuò)誤處理，捕獲連接或發(fā)布過(guò)程中可能發(fā)生的異常，并在異常發(fā)生時(shí)打印錯(cuò)誤信息。```pythonimportpaho.mqtt.clientasmqtt#MQTT回調(diào)函數(shù)defon_connect(client,userdata,flags,rc):print("Connectedwithresultcode"+str(rc))#創(chuàng)建MQTT客戶端實(shí)例client=mqtt.Client("mean_temp_client")#綁定回調(diào)函數(shù)client.on_connect=on_connect#連接MQTTBrokerclient.connect("",1883,60)#發(fā)布消息#...(請(qǐng)?jiān)诖颂幯a(bǔ)充發(fā)布代碼)#斷開(kāi)連接client.disconnect()```9.解釋MQTT協(xié)議中的`QoS`（QualityofService）級(jí)別0、1和2分別代表什么含義？在哪些場(chǎng)景下選擇使用QoS1或QoS2可能更合適？五、綜合應(yīng)用與調(diào)試10.假設(shè)你正在開(kāi)發(fā)一個(gè)簡(jiǎn)單的數(shù)據(jù)采集系統(tǒng)，需求如下：*使用樹(shù)莓派，通過(guò)GPIO讀取DHT11傳感器的溫濕度數(shù)據(jù)。*將讀取到的原始溫濕度數(shù)據(jù)和計(jì)算后的平均值、標(biāo)準(zhǔn)差一起發(fā)送到MQTT主題`python/raspberrypi/data`。*將處理后的數(shù)據(jù)（包含時(shí)間戳、原始數(shù)據(jù)、平均值、標(biāo)準(zhǔn)差）存儲(chǔ)到`processed_data.csv`文件中。*請(qǐng)簡(jiǎn)述你會(huì)如何設(shè)計(jì)這個(gè)系統(tǒng)的主要功能模塊（至少包含數(shù)據(jù)采集、數(shù)據(jù)處理、數(shù)據(jù)傳輸、數(shù)據(jù)存儲(chǔ)四個(gè)模塊），并描述每個(gè)模塊大致需要完成哪些任務(wù)。11.下面是一段嘗試讀取DHT11傳感器數(shù)據(jù)并計(jì)算平均溫度的Python代碼，但其中包含了幾處錯(cuò)誤。請(qǐng)指出至少三處錯(cuò)誤，并簡(jiǎn)要說(shuō)明如何修改。```pythonimportrandomimporttimedefread_dht11():#模擬讀取傳感器，返回None表示失敗returnNoneifrandom.random()<0.2elserandom.uniform(20,30)defmain():temps=[]foriinrange(5):temp=read_dht11()temps.append(temp)print(f"Readtemp:{temp}")time.sleep(1)#延時(shí)1秒mean_temp=sum(temps)/len(temps)#計(jì)算平均值print(f"AverageTemperature:{mean_temp}")if__name__=="__main__":main()```試卷答案一、環(huán)境與庫(kù)準(zhǔn)備1.需要配置Python解釋器及其環(huán)境（如虛擬環(huán)境venv），安裝`gpiozero`庫(kù)（用于樹(shù)莓派GPIO操作）、`smbus2`庫(kù)（用于樹(shù)莓派I2C通信）、`paho-mqtt`庫(kù)（用于MQTT通信）、`numpy`庫(kù)（用于數(shù)據(jù)處理）、`pandas`庫(kù)（用于數(shù)據(jù)存儲(chǔ)）等。例如：`gpiozero`,`smbus2`,`paho-mqtt`,`numpy`,`pandas`。`numpy`庫(kù)的主要用途是進(jìn)行高效的數(shù)值計(jì)算，如數(shù)學(xué)運(yùn)算、數(shù)組操作等。2.```pythonimportpaho.mqtt.clientasmqttclient=mqtt.Client("my_raspberrypi_client")client.connect("",1883,60)```二、硬件數(shù)據(jù)采集3.```pythonfromgpiozeroimportDHT11,DeviceDevice.pin_factory=None#根據(jù)實(shí)際情況配置引腳工廠sensor=DHT11(pin=4)humidity=sensor.humiditytemperature=sensor.temperature#print(f"Temp:{temperature},Humidity:{humidity}")```解析思路：首先導(dǎo)入`DHT11`類，然后創(chuàng)建一個(gè)`DHT11`對(duì)象，指定GPIO引腳。調(diào)用對(duì)象的`humidity`和`temperature`屬性獲取溫濕度值。4.```pythonimportmachinei2c=machine.I2C(scl=machine.Pin(22),sda=machine.Pin(21))#ESP32示例引腳frommpx4115importMPX4115sensor=MPX4115(i2c)```解析思路：使用`machine.I2C`初始化I2C接口，指定SCL和SDA引腳。然后從特定的庫(kù)（如`mpx4115`）中導(dǎo)入`MPX4115`類，創(chuàng)建傳感器對(duì)象，傳入I2C實(shí)例。5.```pythonfromgpiozeroimportDHT11,DeviceDevice.pin_factory=Nonesensor=DHT11(pin=4)attempts=0whileattempts<3:try:humidity=sensor.humiditytemperature=sensor.temperature#print(f"Temp:{temperature},Humidity:{humidity}")break#讀取成功，退出循環(huán)exceptExceptionase:print(f"ErrorreadingDHT11:{e}")attempts+=1time.sleep(1)#ifattempts==3:#print("FailedtoreadDHT11after3attempts")```解析思路：設(shè)置嘗試次數(shù)計(jì)數(shù)器。使用`try...except`結(jié)構(gòu)嘗試讀取數(shù)據(jù)，如果成功則跳出循環(huán)。如果捕獲到異常（讀取失?。瑒t打印錯(cuò)誤信息，等待1秒后增加計(jì)數(shù)器并重試，最多嘗試三次。三、數(shù)據(jù)處理與存儲(chǔ)6.```pythonimportnumpyasnptemps=[22.5,23.1,22.8,23.4,23.0,22.9,23.2]temps_array=np.array(temps)temp_mean=np.mean(temps_array)temp_std=np.std(temps_array)#print(f"Mean:{temp_mean},Std:{temp_std}")```解析思路：使用`numpy`庫(kù)的`mean`函數(shù)計(jì)算平均值，`std`函數(shù)計(jì)算標(biāo)準(zhǔn)差。先將列表轉(zhuǎn)換為`numpy`數(shù)組以方便計(jì)算。7.```pythonimportpandasaspddata={'Temperature':[temp_mean],'Statistics':[temp_std]}df=pd.DataFrame(data)df.to_csv('sensor_data.csv',index=False)```解析思路：創(chuàng)建一個(gè)包含平均值和標(biāo)準(zhǔn)差的字典，然后使用`pandas`庫(kù)創(chuàng)建一個(gè)`DataFrame`。最后調(diào)用`to_csv`方法將`DataFrame`保存為CSV文件，`index=False`表示不保存行索引。四、數(shù)據(jù)傳輸8.```pythonimportpaho.mqtt.clientasmqttimporttimedefon_connect(client,userdata,flags,rc):ifrc==0:print("Connectedsuccessfully.")#client.subscribe("python/io/temperature")#可選，如果需要訂閱else:print(f"Connectfailedwithcode{rc}.")client=mqtt.Client("mean_temp_client")client.on_connect=on_connecttry:client.connect("",1883,60)#等待連接確認(rèn)time.sleep(2)#發(fā)布消息result=client.publish("python/io/temperature",str(temp_mean))print(f"Publishresult:{result[0]}")#result[0]是MQTT返回的結(jié)果代碼exceptExceptionase:print(f"Errorduringconnectionorpublish:{e}")finally:client.disconnect()```解析思路：定義`on_connect`回調(diào)函數(shù)處理連接事件。創(chuàng)建MQTT客戶端實(shí)例并綁定回調(diào)。使用`try...except`結(jié)構(gòu)嘗試連接Broker。連接成功后使用`publish`方法發(fā)布消息，消息內(nèi)容為溫度平均值（轉(zhuǎn)換為字符串）。`finally`塊確保最終斷開(kāi)連接。捕獲并打印可能發(fā)生的任何異常。9.QoS0代表最多一次交付，消息可能會(huì)丟失。QoS1代表至少一次交付，消息可能重復(fù)。QoS2代表只交付一次，確保消息可靠到達(dá)。在要求高可靠性、低數(shù)據(jù)冗余的場(chǎng)景（如關(guān)鍵監(jiān)控?cái)?shù)據(jù)、配置更新）下，選擇QoS1或QoS2更合適。五、綜合應(yīng)用與調(diào)試10.設(shè)計(jì)思路：*數(shù)據(jù)采集模塊：負(fù)責(zé)初始化GPIO，周期性地讀取DHT11傳感器的溫濕度數(shù)據(jù)。需要處理傳感器讀取的異常和超時(shí)。*數(shù)據(jù)處理模塊：負(fù)責(zé)對(duì)采集到的原始溫濕度數(shù)據(jù)進(jìn)行必要的處理（如單位轉(zhuǎn)換、異常值判斷或平滑），并計(jì)算平均值和標(biāo)準(zhǔn)差。*數(shù)據(jù)傳輸模塊：負(fù)責(zé)將原始數(shù)據(jù)、處理后的數(shù)據(jù)（平均值、標(biāo)準(zhǔn)差）以及可能的時(shí)間戳封裝成消息，通過(guò)MQTT協(xié)議發(fā)送到指定的主題。*數(shù)據(jù)存儲(chǔ)模塊：負(fù)責(zé)將處理后的數(shù)據(jù)（包括時(shí)間戳、原始值、計(jì)算值）結(jié)構(gòu)化，并以一定的格式（如CSV）存儲(chǔ)到本地文件系統(tǒng)中，方便后續(xù)分析或備份。11.錯(cuò)誤及修改：