设备可以使用物模型功能,实现属性上报、事件上报和服务调用。
前提条件
已配置开发环境,下载Demo文件。具体请参见配置开发环境。
背景信息
- 物联网平台物模型功能的更多信息,请参见什么是物模型。
- 物模型数据格式的更多信息,请参见设备属性、事件、服务。
初始化
请参考代码的thing_alink.py
文件, 在__init__
函数中完成对象的初始化。
def __init__(self):
self.__linkkit = linkkit.LinkKit(
host_name="cn-shanghai",
product_key="a18wP******",
device_name="LightSwitch",
device_secret="uwMTmVAMnGGHaAkqmeDY6cHxxB******")
配置物模型文件
用户需要从云端控制台下载物模型文件,该文件需要集成到应用工程中,这样对应的topic才能正确接收和发送消息:
self.__linkkit.thing_setup("tsl.json")
- tsl.json默认为utf-8格式,如有修改请确保在保存时格式为utf-8。
- 该设置需要在连接云端之前调用。
物模型功能可通过on_thing_enable通知用户,然后用户可以进行属性上报、事件上报和服务响应:
self.__linkkit.on_thing_enable = self.on_thing_enable
def on_thing_enable(self, userdata):
print("on_thing_enable")
物模型功能不可用时通过 on_thing_disable 通知用户,属性上报、事件上报、服务响应不可用。
self.__linkkit.on_thing_disable = self.on_thing_disable
def on_thing_disable(self, userdata):
print("on_thing_disable")
属性上报
通过thing_post_property上报属性,传入参数为与属性名称及值:
prop_data = {
"abs_speed": 11,
"power_stage": 10
}
self.__linkkit.thing_post_property(prop_data)
服务端对上报的属性做出处理发出响应,SDK通过on_thing_prop_post 通知用户:
self.__linkkit.on_thing_prop_post = self.on_thing_prop_post
def on_thing_prop_post(self, request_id, code, data, message,userdata):
print("on_thing_prop_post request id:%s, code:%d, data:%s message:%s" %
(request_id, code, str(data), message))
thing_post_property返回值rc为0时表明请求写入发送缓冲区成功,rc为其它值为写入发送缓冲失败。当on_thing_prop_post 调用时,表明结果从云端返回了请求结果,code为200时表明解析成功,其它值为失败,失败信息可在message中查看。
事件上报
通过thing_trigger_event上报事件,传入参数为事件identifier和物模型中定义事件对应的对象,如下所示:
event_data = {
"power": 10,
"power_style": 1
}
self.__linkkit.thing_trigger_event(("power_state", event_data))
服务端对上报的事件处理后发出响应,SDK通过on_thing_event_post通知用户:
self.__linkkit.on_thing_event_post = self.on_thing_event_post
def on_thing_event_post(self, event, request_id, code, data, message, userdata):
print("on_thing_event_post event:%s,request id:%s, code:%d, data:%s, message:%s" %
(event, request_id, code, str(data), message))
pass
设置属性
服务端发送设置属性消息后,SDK通过设置的回调函数on_thing_prop_changed通知用户,回调函数中params为包含属性名与值的JSON对象,用户需要对接收到的新属性进行处理:
self.__linkkit.on_thing_prop_changed = self.on_thing_prop_changed
def on_thing_prop_changed(self, params, userdata):
print("on_thing_prop_changed params:" + str(params))
thing_post_property()
发送。
服务响应(异步)
服务端发送服务请求消息后,SDK通过设置的回调函数on_thing_call_service通知用户:
注:所有的服务均通过on_thing_call_service进行通知用户,通过identifier区分不同服务:
self.__linkkit.on_thing_call_service = self.on_thing_call_service
def on_thing_call_service(self, identifier, request_id, params, userdata):
print("on_thing_call_service identifier:%s, request id:%s, params:%s" %
(identifier, request_id, params))
self.__call_service_request_id = request_id
pass
identifier 对应物模型中服务对应的identifier, request_id为区分每次调用的id,params为服务调用参数。
设备端对服务做出返回响应:
self.__linkkit.thing_answer_service(identifier, request_id, code, params)
identifer为物模型中服务对应的identifier,request_id为on_thing_call_service中传递的reques_id,code为返回码,200为本地处理成功,params为字典类型的参数,和物模型中服务返回参数进行对应。
物模型-自定义协议
Python SDK支持您使用自定义格式(非物模型格式)传输数据。
请参考代码thing_custom.py
文件, 在__init__
函数中完成对象的初始化。
def __init__(self):
self.__linkkit = linkkit.LinkKit(
host_name="cn-shanghai",
product_key="a18wP******",
device_name="LightSwitch",
device_secret="uwMTmVAMnGGHaAkqmeDY6cHxxB******")
首先需要将物模型文件参数设置为空:
self.__linkkit.thing_setup()
设备上行数据,通过thing_raw_post_data向服务端发送自定义格式数据:
def protocolToRawData(self, params):
# command set
payload_bytes = b'\x00'
# id
payload_bytes += b'\x00'
payload_bytes += b'\x00'
payload_bytes += b'\x00'
payload_bytes += b'\x01'
# prop_int16 big en
prop_int16 = params["prop_int16"]
payload_bytes += bytes([prop_int16 // 256])
payload_bytes += bytes([prop_int16 % 256])
return payload_bytes
params = {
"prop_int16": 11
}
payload = self.protocolToRawData(params)
print("payload:%r" % payload.hex())
self.__linkkit.thing_raw_post_data(payload)
服务端收到自定义格式数据后给出回复,SDK通过on_thing_raw_data_post通知用户:
self.__linkkit.on_thing_raw_data_post = self.on_thing_raw_data_post
def on_thing_raw_data_post(self, payload, userdata):
print("on_thing_raw_data_post: %s" % str(payload))
服务端发送数据后,SDK通过on_thing_raw_data_arrived通知用户:
self.__linkkit.on_thing_raw_data_arrived = self.on_thing_raw_data_arrived
def on_thing_raw_data_arrived(self, payload, userdata):
print("on_thing_raw_data_arrived:%r" % payload)
print("prop data:%r" % self.rawDataToProtocol(payload))