基于python的MQTT協(xié)議物聯(lián)網云平臺開發(fā)實踐（第2版）01屬性上報目錄CONTENTS02屬性設置03事件上報04服務調用屬性上報(Upload-para)PART0101【實驗目的】

（1）掌握使用MQTT協(xié)議基于物模型進行屬性上報的Python移植方法。

（2）掌握Python的常用命令及代碼解讀。

【實驗設備】

（1）一臺PC，可連接Internet。

（2）Python軟件、Pycharm編程環(huán)境。

【實驗要求】

在任務一實驗一創(chuàng)建的數(shù)據(jù)格式為OneJson的設備下，基于物模型中溫度和濕度兩個屬性功能點，采用Python編程實現(xiàn)設備屬性上報。

屬性上報屬性上報二、屬性上報

1．選擇topic屬性上報需要找到對應的系統(tǒng)topic，在模擬器進行調試的過程中，已經查閱過相關topic，發(fā)布數(shù)據(jù)采用如下topic：

$sys/{pid}/{device-name}/thing/property/post

2．構建上報的屬性數(shù)據(jù)流

屬性上報的采用JSON數(shù)據(jù)流格式，以上報屬性值num至功能點temp、上報屬性值num1至功能點Hum

為例，構建如下數(shù)據(jù)流：一、建立連接

按照上述步驟，建立客戶端與服務器的連接，并通過回調函數(shù)打印連接結果。{"id":"124","version":"1.0","params":{"temp":{"value":25.5},"Hum":{"value":60}}}屬性上報3．屬性上報

調用如下函數(shù)，實現(xiàn)屬性上報：publish(topic,payload,qos,retain)該函數(shù)包含4個參數(shù)，需要重點設置前3項參數(shù)。

topic：表示發(fā)布信息的主題。

payload：表示要發(fā)布的具體信息。

qos：表示服務質量，分別為0、1、2三種級別。

示例代碼如下：

client.publish(pub_topic,json.dumps(body),qos=0)屬性上報import…

#定義token計算函數(shù)

deftoken(product_id,access_key,auth_info=None)…

#定義回調函數(shù)

defon_connect(client,userdata,flags,reason_code,properties):

print(f"Connected!Result:{mqtt.connack_string(reason_code)}(Code:{reason_code})")

if__name__=='__main__':

host='6'

port=8883

access_key=‘xxxxx'#devicekey

auth_info=‘xxxxxxxx'#client_id

product_id=‘xxxxxxxx'#username

#計算token

api_token=token(product_id,access_key,auth_info)

print(api_token)

#client=mqtt.Client(client_id=auth_info,protocol=mqtt.MQTTv311)

client=mqtt.Client(

client_id=auth_info,

protocol=mqtt.MQTTv311,

callback_api_version=mqtt.CallbackAPIVersion.VERSION2

)

client.on_connect=on_connect

client.tls_set(ca_certs='serverCert.pem',cert_reqs=CERT_REQUIRED)

client.tls_insecure_set(True)

client.username_pw_set(username=product_id,password=api_token)

client.connect(host=host,port=port,keepalive=30)

client.loop_start()whileTrue:

num=22

num1=56

body={

"id":"124",

"version":"1.0",

"params":{

"temperature":{

"value":num

},

"humidity":{

"value":num1

}

}

}

pub_topic='$sys/%s/%s/thing/property/post'%(product_id,auth_info)

print(pub_topic)

print('fp:temperature->%s'%num)

print('fp:humidity->%s'%num1)

client.publish(pub_topic,json.dumps(body),qos=0)

time.sleep(5)onejson—屬性上報import…

#定義token計算函數(shù)

deftoken(product_id,access_key,auth_info=None):

…

defon_connect(client,userdata,flags,reason_code,properties):

print(mqtt.connack_string(reason_code)+'rc:%s'%reason_code)

if__name__=='__main__':

host='6'

port=8883

access_key=‘xxxx'#devicekey

auth_info=‘xxxx'#設備名稱

product_id=‘xxxx'#產品id

#計算token

api_token=token(product_id,access_key,auth_info)

print(api_token)

client=mqtt.Client(

client_id=auth_info,

protocol=mqtt.MQTTv311,

callback_api_version=mqtt.CallbackAPIVersion.VERSION2

)

client.on_connect=on_connect

client.on_message=on_message

client.tls_set(ca_certs='serverCert.pem',cert_reqs=CERT_REQUIRED)

client.tls_insecure_set(True)

client.username_pw_set(username=product_id,password=api_token)

client.connect(host=host,port=port,keepalive=30)

client.loop_start()whileTrue:

time.sleep(1)

num=20

body={

"id":123,

"dp":{

"temperature":[{

"v":num,

}]

}

}

pub_topic='$sys/%s/%s/dp/post/json'%(product_id,auth_info)

print(pub_topic)

print('dp:temperature->%s'%num)

client.publish(pub_topic,json.dumps(body),qos=0)

time.sleep(30)

body={

"id":"124",

"version":"1.0",

"params":{

"temperature":{

"value":num

},

"humidity":{

"value":num1

}

}

}數(shù)據(jù)流--發(fā)布信息pub_topic='$sys/%s/%s/thing/property/post'%(product_id,auth_info)第一條信息表示token，第二條信息表示發(fā)布數(shù)據(jù)的系統(tǒng)topic，第三、四條信息表示功能點標識符和屬性值，第五條信息表示連接成功。進入OneNET平臺，在“設備接入管理”頁面中，選擇設備管理，并單擊該設備對應的“屬性”，即可看到上報的屬性。

屬性上報res=products/O1wlcBxQXO/devices/HT&sign=bv5pA9xhmwmz9kUgpnKyFpgIsBY%3D&et=1712736146&method=sha1&version=2018-10-31$sys/O1wlcBxQXO/HT/thing/property/postfp:temp->22fp:Hum->56ConnectionAccepted.rc:0屬性設置get_setget_settingPART0202【實驗目的】

（1）掌握使用MQTT協(xié)議基于物模型接收云平臺屬性設置的Python移植方法。

（2）掌握Python的常用命令及代碼解讀。

【實驗設備】

（1）一臺PC，可連接Internet。

（2）Python軟件、Pycharm編程環(huán)境。

【實驗要求】

在任務一實驗一創(chuàng)建的數(shù)據(jù)格式為OneJson的設備下，采用Python編程接收云平臺對物模型中“溫度”功能點的屬性設置指令，并打印“TemperatureSetting：”和接收到的設置值；當收到的“濕度”功能點的屬性設置指令，打印“HumiditySetting：”和接收到的設置值；當收到其他信息，打印“Error”。

屬性設置一、建立連接

按照上述步驟，建立客戶端與服務器的連接，并通過回調函數(shù)打印連接結果。

二、查詢物模型topic在“設備開發(fā)”頁面，單擊“Topic管理”，再單擊“物模型topic”，其中，與直連設備屬性設置相關的topic。

三、訂閱主題

1．選擇topic為了能在底層程序端接收云平臺屬性設置的指令，可訂閱如下topic：

$sys/{pid}/{device-name}/thing/property/set

上述topic表示接收系統(tǒng)“屬性設置”指令。

2．訂閱topic

調用如下函數(shù)，實現(xiàn)訂閱主題：

subscribe(topic,qos)該函數(shù)包含兩個參數(shù)，如下所述。

topic：表示訂閱的主題。

qos：表示服務質量，分別為0、1、2三種級別。

示例代碼如下：

client.subscribe('$sys/%s/%s/thing/property/set'%(product_id,auth_info),qos=0)屬性設置四、定義消息回調函數(shù)

在收到客戶端訂閱主題的消息后，通常會調用on_message()函數(shù)來打印接收到的反饋信息。

client.on_message=on_message

在調用前，會對該函數(shù)的具體執(zhí)行內容進行定義，部分代碼如下：defon_message(client,userdata,msg):print('ON_MESSAGE:'+msg.topic+"",msg.payload)

上述代碼表示接收到訂閱主題推送的消息后，打印消息的主題和內容。屬性設置get_setif__name__=='__main__':

host='6'

port=8883

access_key='xxxxxx'#devicekey

auth_info=‘xxxxx'#client_id

product_id='xxxxx'#username

#計算token

api_token=token(product_id,access_key,auth_info)

print(api_token)

#client=mqtt.Client(client_id=auth_info,protocol=mqtt.MQTTv311)

client=mqtt.Client(

client_id=auth_info,

protocol=mqtt.MQTTv311,

callback_api_version=mqtt.CallbackAPIVersion.VERSION2

)

client.on_connect=on_connect

client.on_message=on_message

client.tls_set(ca_certs='serverCert.pem',cert_reqs=CERT_REQUIRED)

client.tls_insecure_set(True)

client.username_pw_set(username=product_id,password=api_token)

client.connect(host=host,port=port,keepalive=30)

client.loop_start()

whileTrue:

time.sleep(1)import…

#定義token計算函數(shù)

deftoken(product_id,access_key,auth_info=None)…#定義回調函數(shù)

defon_connect(client,userdata,flags,reason_code,properties):…defon_message(client,userdata,msg):

print('ON_MESSAGE:'+msg.topic+"",msg.payload)

if'$sys/%s/%s/thing/property/set'%(product_id,auth_info)inmsg.topic:

msg1=str(msg.payload)

msg2=json.loads(msg1[2:-1])

params=msg2.get("params")

key=list(params.keys())

value=params.get(key[0])

ifkey[0]=="temperature":

print("temperatureSetting:",value)

elifkey[0]=="humidity":

print("humiditySetting:",value)

else:

print("Error")

time.sleep(5)

運行上述代碼，從OneNET云平臺應用模擬器中選取功能點“溫度”，屬性設置值設為30，單擊屬性設置，得到如下運行結果：

第一條信息表示token，第二條信息表示連接成功，第三條信息表示收到的平臺下發(fā)的內容，具體包括topic和具體平臺下發(fā)的內容，表示屬性名稱為temp、具體的屬性設置值為30，第四條表示屬性設置響應的topic，第五條是屬性設置的響應消息，第六條表示根據(jù)接收到的屬性設置值，軟件端按照功能要求執(zhí)行的操作具體內容，此處可以根據(jù)打開燈、打開風扇等具體功能要求進行個性化定義。

屬性設置get_setting運行結果res=products/O1wlcBxQXO/devices/HT&sign=MD2adLGB102WbLLY11fO%2BKkeXw0%3D&et=1713002009&method=sha1&version=2018-10-31ConnectionAccepted.rc:0ON_MESSAGE:$sys/O1wlcBxQXO/HT/thing/property/setb'{"id":"1","version":"1.0","params":{"temp":30}}'TemperatureSetting:30從OneNET云平臺應用模擬器中選取功能點“濕度”，屬性設置值設為25，單擊屬性設置，得到如下運行結果：

第一條信息表示收到的平臺下發(fā)的內容，具體包括topic和具體平臺下發(fā)的內容，表示屬性名稱為Hum、具體的屬性設置值為25，第二條表示屬性設置響應的topic，第三條表示屬性設置的響應消息，第四條表示根據(jù)接收到的屬性設置值，軟件端按照功能要求執(zhí)行的操作具體內容，此處可以根據(jù)打開燈、打開風扇等具體功能要求進行個性化定義。屬性設置get_setting運行結果ON_MESSAGE:$sys/O1wlcBxQXO/HT/thing/property/setb'{"id":"2","version":"1.0","params":{"Hum":25}}'HumiditySetting:25事件上報upload_thPART0303【實驗目的】

（1）掌握使用MQTT協(xié)議基于物模型進行事件上報的Python移植方法。

（2）掌握Python的常用命令及代碼解讀。

【實驗設備】

（1）一臺PC，可連接Internet。

（2）Python軟件、Pycharm編程環(huán)境。

【實驗要求】

在任務一實驗一創(chuàng)建的數(shù)據(jù)格式為OneJson的設備下，進行高溫事件的上報。

事件上報一、建立連接

按照上述步驟，建立客戶端與服務器的連接，并通過回調函數(shù)打印連接結果。

事件上報二、事件上報

1．選擇topic

事件上報需要找到對應的系統(tǒng)topic，在模擬器進行調試的過程中，已經查閱過相關topic，發(fā)布數(shù)據(jù)采用如下topic：

$sys/{pid}/{device-name}/thing/event/post2．構建上報的事件數(shù)據(jù)流

事件上報的采用JSON數(shù)據(jù)流格式，以上報高溫事件為例，按照上傳事件至功能點“heat”，參數(shù)“Temperature”的值為90，構建如下數(shù)據(jù)流：

事件上報3．事件上報

調用如下函數(shù)，實現(xiàn)屬性上報：

4．訂閱反饋

訂閱上報事件響應主題，當云平臺收到上報的事件后，向客戶端發(fā)送反饋，客戶端訂閱相關主題后，可以收到反饋信息。示例代碼如下：

client.subscribe('$sys/%s/%s/thing/event/post/reply'%(product_id,auth_info),qos=0)publish(topic,payload,qos,retain)import…

#定義token計算函數(shù)

deftoken(product_id,access_key,auth_info=None)…

#定義回調函數(shù)

defon_connect(client,userdata,flags,rc):

#打印連接狀態(tài)

print(mqtt.connack_string(rc)+'rc:%s'%rc)

ifrc==0:

client.subscribe('$sys/%s/%s/thing/event/post/reply'%(product_id,auth_info),qos=0)

defon_message(client,userdata,msg):

print('ON_MESSAGE:'+msg.topic+"",msg.payload)

if__name__=='__main__':

host='6'

port=8883

access_key='RFFOamJDYTY2eEYwWWRyejY5R1ZFTU5YQWVwY01jUDI='#devicekey

auth_info='HT'#client_id

product_id='O1wlcBxQXO'#username

#計算token

api_token=token(product_id,access_key,auth_info)

print(api_token)

client=mqtt.Client(client_id=auth_info,protocol=mqtt.MQTTv311)

client.on_connect=on_connect

client.on_message=on_message

client.tls_set(ca_certs='serverCert.pem',cert_reqs=CERT_REQUIRED)

client.tls_insecure_set(True)

client.username_pw_set(username=product_id,password=api_token)

client.connect(host=host,port=port,keepalive=30)

client.loop_start()

whileTrue:

body={

"id":"123",

"params":{

"heat":{

"value":{

“temperature":90

}

}

}

}

pub_topic='$sys/%s/%s/thing/event/post'%(product_id,auth_info)

print(pub_topic)

print('Heat:alarm')

client.publish(pub_topic,json.dumps(body),qos=0)

time.sleep(50)運行上述代碼，得到如下運行結果：第一條信息表示token，第二條信息表示上報事件的系統(tǒng)topic，第三條信息打印了事件標識符和參數(shù)，第四條信息表示連接成功,第五條信息表示接收到云平臺上報事件的響應結果。

進入OneNET平臺，在“設備接入管理”頁面中，選擇設備管理，并單擊該設備對應的“事件記錄”，即可看到上報的事件。事件上報res=products/O1wlcBxQXO/devices/HT&sign=70VtxcVslQUCHH1UxEFYicWFaHc%3D&et=1713163520&method=sha1&version=2018-10-31ConnectionAccepted.rc:0$sys/O1wlcBxQXO/HT/thing/event/postHeat:alarmON_MESSAGE:$sys/O1wlcBxQXO/HT/thing/event/post/replyb'{"id":"123","code":200,"msg":"success"}'服務調用invokePART0404【實驗目的】

（1）掌握使用MQTT協(xié)議基于物模型接收云平臺服務調用指令的Python移植方法。（2）掌握Python的常用命令及代碼解讀。

【實驗設備】

（1）一臺PC，可連接Internet。

（2）Python軟件、Pycharm編程環(huán)境。

【實驗要求】

在任務一實驗一創(chuàng)建的數(shù)據(jù)格式為OneJson的設備下，采用Python編程接收云平臺對物模型中“風扇控制”功能點的服務調用指令，當接收到指令“1”時，打印“Turnon”；當接收到指令“0”時，打印“Turnoff”。

服務調用服務調用一、建立連接

按照上述步驟，建立客戶端與服務器的連接，并通過回調函數(shù)打印連接結果。

二、訂閱主題

1．選擇topic服務調用需要找到對應的系統(tǒng)topic，在模擬器進行調試的過程中，已經查閱過相關topic，為了能在底層程序端接收云平臺服務調用的指令，可訂閱如下topic：

$sys/{pid}/{device-name}/thing/service/{identifier}/invoke上述topic表示接收系統(tǒng)“服務調用”指令。

2．訂閱topic調用如下函數(shù)，實現(xiàn)訂閱主題：

subscribe(topic,qos)該函數(shù)包含兩個參數(shù)，如下所述。topic：表示訂閱的主題。

qos：表示服務質量，分別為0、1、2三種級別。

示例代碼如下：

client.subscribe('$sys/%s/%s/thing/service/%s/invoke'%(product_id,auth_info,identifier),qos=0)三、定義消息回調函數(shù)

在調用消息回調函數(shù)前，會對該函數(shù)的具體執(zhí)行內容進行定義，部分代碼如下：

服務調用上述代碼表示接收到訂閱主題推送的消息后，打印消息的主題和內容。然后在回調函數(shù)內對消息進行解析，提取云平臺調用服務的具體參數(shù)值。此處解析的方法與實驗二類似，并在解析消息體的同時，解析了消息主題，并向云平臺發(fā)送反饋信息。

defon_message(client,userdata,msg):print('ON_MESSAGE:'+msg.topic+"",msg.payload)三、定義消息回調函數(shù)

msg1=str(msg.payload)#將接收到的消息內容轉成字符串msg2=json.loads(msg1[2:-1])#將字符串轉成字典格式cmd_id=msg2.get('id')#讀取消息中鍵“id”對應的值params=msg2.get("params")#讀取消息中鍵“params”對應的值，值的格式為字典key=list(params.keys())#將上述值轉成列表value=params.get(key[0])#讀取第一個鍵對應的值cmd_resp_topic='$sys/%s/%s/thing/service/%s/invoke_reply'%(product_id,auth_info,identifier)#設備回復平臺調用服務reply_body={"id":cmd_id,"code":200,"msg":"success"}#設備回復成功收到調用指令body=json.dumps(reply_body)body=body.replace("'","\"")client.publish(cmd_resp_topic,json.dumps(body))#發(fā)布設備反饋信息服務調用以接收到如下消息內容為例：b'{"id":"3","version":"1.0","params":{"sw":1}}'cmd_id：“3”params：{'sw':1}key：['sw']value：1cmd_resp_topic:$sys/O1wlcBxQXO/HT/thing/service/FAN/invoke_replybody:{"id":"3","code":200,"msg":"success"}解析消息得到云平臺下發(fā)的屬性設置功能點和具體值后，根據(jù)本實驗目標功能，設計軟件代碼：三、定義消息回調函數(shù)

服務調用ifvalue==1:#判斷服務調用：開關sw值為1print("Turnon")elifvalue==0:#判斷服務調用：開關sw值為0print("Turnoff")else:print("Error")運行上述代碼，各參數(shù)計算結果如下：invokedefon_connect(client,userdata,flags,rc):print(mqtt.connack_string(rc)+'rc:%s'%rc)ifrc==0:client.subscribe('$sys/%s/%s/thing/service/%s/invoke'%(product_id,auth_info,identifier),qos=0)

defon_message(client,userdata,msg):print('ON_MESSAGE:'+msg.topic+"",msg.payload)#當接收到的主題滿足一定條件時

if'$sys/%s/%s/thing/service/%s/invoke'%(product_id,auth_info,identifier)inmsg.topic:#定義上傳客戶端反饋信息的主題cmd_resp_topic='$sys/%s/%s/thing/service/%s/invoke_reply'%(product_id,auth_info,identifier)#對接收到的信息進行格式轉換msg1=str(msg.payload)msg2=json.loads(msg1[2:-1])cmd_id=msg2.get('id')

#設備回復成功收到調用指令

reply_body={"id":cmd_id,"code":200,"msg":"success"}body=json.dumps(reply_body)body=body.replace("'","\"")#發(fā)布設備反饋信息client.publish(cmd_resp_topic,json.dumps(body))print(cmd_resp_topic)print(body)

#解析消息體，獲取云平臺調用的具體服務以及參數(shù)值

params=msg2.get("params")key=list(params.keys())value=params.get(key[0])

#根據(jù)調用服務內容，執(zhí)行相應操作

ifvalue==1:print("Turnon")elifvalue==0:print("Turnoff")el