腳本模板

您可以基于以下Python腳本模板編寫消息解析腳本。

# 將設(shè)備的自定義格式數(shù)據(jù)轉(zhuǎn)換為Alink協(xié)議的數(shù)據(jù)，設(shè)備上報(bào)數(shù)據(jù)到物聯(lián)網(wǎng)平臺(tái)時(shí)調(diào)用。
# 入?yún)? rawData，列表，列表元素取值為int類型，不能為空。
# 出參: jsonObj，字典，不能為空。
def raw_data_to_protocol(rawData):
    jsonObj = {}
    return jsonObj

# 將Alink協(xié)議的數(shù)據(jù)轉(zhuǎn)換為設(shè)備能識(shí)別的格式數(shù)據(jù)，物聯(lián)網(wǎng)平臺(tái)給設(shè)備下發(fā)數(shù)據(jù)時(shí)調(diào)用。
# 入?yún)? jsonData，字典，不能為空。
# 出參: rawdata，列表，列表元素取值為int類型且大小為[0, 255]之間，不能為空。
def protocol_to_raw_data(jsonData):
    rawData = []
    return rawData

腳本編寫注意事項(xiàng)

• 請(qǐng)避免使用全局變量，否則會(huì)造成執(zhí)行結(jié)果不一致。
• 腳本中，處理數(shù)據(jù)采用補(bǔ)碼的方式， [-128, 127]補(bǔ)碼范圍為[0, 255]。例如，-1對(duì)應(yīng)的補(bǔ)碼為255（10進(jìn)制表示）。
• 解析設(shè)備上報(bào)數(shù)據(jù)的函數(shù)（raw_data_to_protocol）的入?yún)檎蛿?shù)組。需要通過(guò)0xFF進(jìn)行與操作，獲取其對(duì)應(yīng)的補(bǔ)碼。
• 解析物聯(lián)網(wǎng)平臺(tái)下發(fā)數(shù)據(jù)的函數(shù)（protocol_to_raw_data）的返回結(jié)果為數(shù)組。數(shù)組元素為整型，取值為[0, 255]。

腳本示例

以下是基于提交物模型消息解析腳本中定義的屬性和通信協(xié)議編寫的腳本。

物模型消息中數(shù)據(jù)類型說(shuō)明，請(qǐng)參見(jiàn)物模型支持的數(shù)據(jù)類型。物模型屬性、事件上報(bào)數(shù)據(jù)后，物聯(lián)網(wǎng)平臺(tái)返回的Alink格式響應(yīng)結(jié)果，會(huì)通過(guò)腳本解析轉(zhuǎn)換后返回給設(shè)備。Alink數(shù)據(jù)格式說(shuō)明，請(qǐng)參見(jiàn)設(shè)備屬性、事件、服務(wù)。

# coding=UTF-8
import struct
import common_util

COMMAND_REPORT = 0x00  # 屬性上報(bào)。
COMMAND_SET = 0x01  # 屬性設(shè)置。
COMMAND_REPORT_REPLY = 0x02  # 上報(bào)數(shù)據(jù)返回結(jié)果。
COMMAND_SET_REPLY = 0x03  # 屬性設(shè)置設(shè)備返回結(jié)果。
COMMAD_UNKOWN = 0xff  # 未知的命令。
ALINK_PROP_REPORT_METHOD = 'thing.event.property.post'  # 物聯(lián)網(wǎng)平臺(tái)Topic，設(shè)備上傳屬性數(shù)據(jù)到云端。
ALINK_PROP_SET_METHOD = 'thing.service.property.set'  # 物聯(lián)網(wǎng)平臺(tái)Topic，云端下發(fā)屬性控制指令到設(shè)備端。
SELF_DEFINE_TOPIC_UPDATE_FLAG = '/user/update'  # 自定義Topic：/user/update。
SELF_DEFINE_TOPIC_ERROR_FLAG = '/user/update/error' # 自定義Topic：/user/update/error。


# 示例數(shù)據(jù)：
# 設(shè)備上報(bào)屬性數(shù)據(jù)：
# 傳入?yún)?shù)：
#    0x000000000100320100000000
# 輸出結(jié)果：
#    {"method":"thing.event.property.post","id":"1","params":{"prop_float":0,"prop_int16":50,"prop_bool":1},"version":"1.0"}
# 屬性設(shè)置的返回結(jié)果：
# 傳入?yún)?shù)：
#    0x0300223344c8
# 輸出結(jié)果：
#    {"code":"200","data":{},"id":"2241348","version":"1.0"}

def raw_data_to_protocol(bytes):
    uint8Array = []
    for byteValue in bytes:
        uint8Array.append(byteValue & 0xff)

    fHead = uint8Array[0]
    jsonMap = {}
    if fHead == COMMAND_REPORT:
        jsonMap['method'] = ALINK_PROP_REPORT_METHOD
        jsonMap['version'] = '1.0'
        jsonMap['id'] = str(bytes_to_int(uint8Array[1:5]))
        params = {}
        params['prop_int16'] = bytes_to_int(uint8Array[5:7])
        params['prop_bool'] = bytes_to_int(uint8Array[7: 8])
        params['prop_float'] = bytes_to_int(uint8Array[8:])
        jsonMap['params'] = params
    elif fHead == COMMAND_SET_REPLY:
        jsonMap['version'] = '1.0'
        jsonMap['id'] = str(bytes_to_int(uint8Array[1:5]))
        jsonMap['code'] = str(bytes_to_int(uint8Array[5:]))
        jsonMap['data'] = {}

    return jsonMap


# 示例數(shù)據(jù)：
# 云端下發(fā)屬性設(shè)置指令：
# 傳入?yún)?shù)：
#    {"method":"thing.service.property.set","id":"12345","version":"1.0",
#    "params":{"prop_float":123.452, "prop_int16":333, "prop_bool":1}}
# 輸出結(jié)果：
#    0x0100003039014d0142f6e76d
# 設(shè)備上報(bào)的返回結(jié)果：
# 傳入數(shù)據(jù)：
#    {"method":"thing.event.property.post","id":"12345","version":"1.0","code":200,"data":{}}
# 輸出結(jié)果 ->
#    0x0200003039c8
def protocol_to_raw_data(json):
    method = json.get('method', None)
    id = json.get('id', None)
    version = json.get('version', None)
    payload_array = []

    if method == ALINK_PROP_SET_METHOD:
        params = json.get('params')
        prop_float = params.get('prop_float', None)
        prop_int16 = params.get('prop_int16', None)
        prop_bool = params.get('prop_bool', None)

        payload_array = payload_array + int_8_to_byte(COMMAND_SET)
        payload_array = payload_array + int_32_to_byte(int(id))
        payload_array = payload_array + int_16_to_byte(prop_int16)
        payload_array = payload_array + int_8_to_byte(prop_bool)
        payload_array = payload_array + float_to_byte(prop_float)

    elif method == ALINK_PROP_REPORT_METHOD:
        code = json.get('code', None)
        payload_array = payload_array + int_8_to_byte(COMMAND_REPORT_REPLY)
        payload_array = payload_array + int_32_to_byte(int(id))
        payload_array = payload_array + int_8_to_byte(code)
    else:
        code = json.get('code')
        payload_array = payload_array + int_8_to_byte(COMMAD_UNKOWN)
        payload_array = payload_array + int_32_to_byte(int(id))
        payload_array = payload_array + int_8_to_byte(code)

    return payload_array



#  示例數(shù)據(jù)：
#  自定義Topic：
#      /user/update，上報(bào)數(shù)據(jù)。
#  輸入?yún)?shù)：
#      topic:/{productKey}/{deviceName}/user/update
#      bytes: 0x000000000100320100000000
#  輸出參數(shù)：
#  {
#     "prop_float": 0,
#     "prop_int16": 50,
#     "prop_bool": 1,
#     "topic": "/{productKey}/{deviceName}/user/update"
#   }
def transform_payload(topic, bytes):
    uint8Array = []
    for byteValue in bytes:
        uint8Array.append(byteValue & 0xff)

    jsonMap = {}
    if SELF_DEFINE_TOPIC_ERROR_FLAG in topic:
        jsonMap['topic'] = topic
        jsonMap['errorCode'] = bytes_to_int(uint8Array[0:1])

    elif SELF_DEFINE_TOPIC_UPDATE_FLAG in topic:
        jsonMap['topic'] = topic
        jsonMap['prop_int16'] = bytes_to_int(uint8Array[5:7])
        jsonMap['prop_bool'] = bytes_to_int(uint8Array[7: 8])
        jsonMap['prop_float'] = bytes_to_int(uint8Array[8:])

    return jsonMap


# byte轉(zhuǎn)成int。
def bytes_to_int(bytes):
    data = ['%02X' % i for i in bytes]
    return int(''.join(data), 16)



# 8位整型轉(zhuǎn)成byte數(shù)組。
def int_8_to_byte(value):
    t_value = '%02X' % value
    if len(t_value) % 2 != 0:
        t_value += '0'

    return hex_string_to_byte_array(t_value)


# 32位整型轉(zhuǎn)成byte數(shù)組。
def int_32_to_byte(value):
    t_value = '%08X' % value
    if len(t_value) % 2 != 0:
        t_value += '0'

    return hex_string_to_byte_array(t_value)


# 16位整型轉(zhuǎn)成byte數(shù)組。
def int_16_to_byte(value):
    t_value = '%04X' % value
    if len(t_value) % 2 != 0:
        t_value += '0'

    return hex_string_to_byte_array(t_value)


# float轉(zhuǎn)成整型數(shù)組。
def float_to_byte(param):
    return hex_string_to_byte_array(struct.pack(">f", param).encode('hex'))


# 16進(jìn)制字符串轉(zhuǎn)成byte數(shù)組。
def hex_string_to_byte_array(str_value):
    if len(str_value) % 2 != 0:
        return None

    cycle = len(str_value) / 2

    pos = 0
    result = []
    for i in range(0, cycle, 1):
        temp_str_value = str_value[pos:pos + 2]
        temp_int_value = int(temp_str_value, base=16)

        result.append(temp_int_value)
        pos += 2
    return result