Este tema proporciona una plantilla de script en Python y un script de ejemplo que puede usar para analizar sintácticamente los mensajes de Thing Specification Language (TSL).
Plantilla del script
Puede escribir un script en Python con base en la siguiente plantilla para analizar sintácticamente los mensajes TSL.
La plantilla del script solo aplica a los productos cuyo parámetro
Data Type
se establece como
Custom
.
# Convert custom data of a device to Alink JSON data when the device submits data to IoT Platform.
# rawData: the input parameter. The value is a list of integers that cannot be empty.
# jsonObj: the output parameter. The value is a JSON object that indicates a dictionary and cannot be empty.
def raw_data_to_protocol(rawData):
jsonObj = {}
return jsonObj
# Convert Alink JSON data to data that can be recognized by a device when IoT Platform sends data to the device.
# jsonData: the input parameter. The value is a JSON object that indicates a dictionary and cannot be empty.
# rawData: the output parameter. The value is a list of integers within a value range of [0,255] and cannot be empty.
def protocol_to_raw_data(jsonData):
rawData = []
return rawDataNotas para la escritura del script
No use variables globales. De lo contrario, los resultados podrían ser inconsistentes.
El script procesa los datos mediante la operación de complemento a dos. El rango de complemento para los valores entre [-128,127] es [0,255]. Por ejemplo, el complemento de -1 es 255 en formato decimal.
La función raw_data_to_protocol analiza sintácticamente los datos que envían los dispositivos. El parámetro de entrada de esta función es un arreglo de enteros. Use
0xFFen la operación AND bit a bit para obtener el complemento.La función protocol_to_raw_data analiza sintácticamente los datos que envía IoT Platform y devuelve el resultado en un arreglo. Cada elemento del arreglo debe ser un entero dentro del rango de valores [0,255].
Script de ejemplo
A continuación se muestra un script de ejemplo escrito con base en las propiedades y el protocolo definidos en el tema Submit a TSL model message parsing script:
Para más información sobre los tipos de datos compatibles con los modelos TSL, consulte Supported data types. Cuando un dispositivo envía datos de propiedades o eventos TSL, IoT Platform genera una respuesta con datos en formato Alink JSON. Antes de enviar dicha respuesta al dispositivo, IoT Platform usa el script para convertir el formato de los datos a uno que el dispositivo reconozca. Para más información sobre el formato Alink JSON, consulte Device properties, events, and services.
# coding=UTF-8
import struct
import common_util
COMMAND_REPORT = 0x00 # Submit properties.
COMMAND_SET = 0x01 # Configure properties.
COMMAND_REPORT_REPLY = 0x02 # Return results of data submission.
COMMAND_SET_REPLY = 0x03 # Return results of property setting.
COMMAD_UNKOWN = 0xff # Unknown commands.
ALINK_PROP_REPORT_METHOD = 'thing.event.property.post' # The topic for devices to submit property data to IoT Platform.
ALINK_PROP_SET_METHOD = 'thing.service.property.set' # The topic for IoT Platform to send a property setting command to devices.
SELF_DEFINE_TOPIC_UPDATE_FLAG = '/user/update' # Define the following custom topic: /user/update.
SELF_DEFINE_TOPIC_ERROR_FLAG = '/user/update/error' # Define the following custom topic: /user/update/error.
# Sample data:
# Submit property data
# Input:
# 0x000000000100320100000000
# Output:
# {"method":"thing.event.property.post","id":"1","params":{"prop_float":0,"prop_int16":50,"prop_bool":1},"version":"1.0"}
# Returned result of property setting
# Input:
# 0x0300223344c8
# Output:
# {"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
# Sample data:
# Sends downstream commands to configure device properties
# Input:
# {"method":"thing.service.property.set","id":"12345","version":"1.0",
# "params":{"prop_float":123.452, "prop_int16":333, "prop_bool":1}}
# Output:
# 0x0100003039014d0142f6e76d
# Returned result after the device submits data
# Input:
# {"method":"thing.event.property.post","id":"12345","version":"1.0","code":200,"data":{}}
# Output:
# 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
# Sample data:
# Use the following custom topic to submit data:
# /user/update.
# Input:
# topic:/{productKey}/{deviceName}/user/update
# bytes: 0x000000000100320100000000
# Output:
# {
# "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
# Convert a byte array to an integer.
def bytes_to_int(bytes):
data = ['%02X' % i for i in bytes]
return int(''.join(data), 16)
# Convert an 8-bit integer to a byte array.
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)
# Convert a 32-bit integer to a byte array.
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)
# Convert a 16-bit integer to a byte array.
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)
# Convert a float to an integer array.
def float_to_byte(param):
return hex_string_to_byte_array(struct.pack(">f", param).encode('hex'))
# Convert a hexadecimal string to a byte array.
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