#include "pp.h"
#include "dy_utils/dy_pp.h"

extern int pp_data_exec_crc_info(pp_var_t *var, template_cfg_t *ptemplate, char *data, int len, unsigned int *crc);

static cJSON *pp_get_rglt_data_from_json_str(pp_var_t *var, pp_regulate_signal_t *rglt_data, char *json_str)
{
    cJSON *root = cJSON_Parse(json_str);
    if (root)
    {
        GET_JSON_VALUE_STRING(root, "sn", rglt_data->sn);
        GET_JSON_VALUE_STRING(root, "identifier", rglt_data->src_identifier);
        GET_JSON_VALUE_INT(root, "mi", rglt_data->mi);
        GET_JSON_VALUE_INT(root, "server_period", rglt_data->period);
        GET_JSON_VALUE_STRING(root, "requester", rglt_data->requester);
    }
    else
    {
        dbg_syslog(LOG_ERR, "json parse error, %s", json_str);
    }

    return root;
}

#define SCRIPT_LUA_BUFSIZ 0x4000
cJSON *exec_encode_operation_script_lua(pp_var_t *var, cJSON *tag_node, cJSON *jtag_new, service_param_t *param)
{
    size_t buflen;
    char * ptmp, * pstr;
    char * buf_p = calloc(2, SCRIPT_LUA_BUFSIZ);

    buflen = 0;
    ptmp = pstr = NULL;
    if (buf_p == NULL) {
        dbg_syslog(LOG_WARNING, "calloc failed");
        return 0;
    }
    ptmp = (char *) malloc(SCRIPT_LUA_BUFSIZ);
    pstr = (char *) malloc(SCRIPT_LUA_BUFSIZ);
    if (ptmp == NULL || pstr == NULL) {
        free(buf_p);
        if (ptmp != NULL)
            free(ptmp);
        if (pstr != NULL)
            free(pstr);
        dbg_syslog(LOG_WARNING, "malloc twice failed");
        return 0;
    }

    //strcpy(buf_p, LUA_DDEFAULT_ENV);
    cJSON *child = NULL;
    for (child = tag_node->child; child; child = child->next)
    {
        if (child->string && strcmp(child->string, "sn"))
        {
            int plen;
            double val = 0xFFFFFFFF;

            plen = 0;
            ptmp[0] = '\0';
            GET_JSON_VALUE_DOUBLE(tag_node, child->string, val);
            if (val == 0xFFFFFFFF || child->type == cJSON_String) {
                pstr[0] = '\0';
                GET_JSON_VALUE_STRING(tag_node, child->string, pstr);
                if (pstr[0] != '\0') {
                    plen = snprintf(ptmp, SCRIPT_LUA_BUFSIZ, "local %s = '%s' ", child->string, pstr);
                }
            } else {
                plen = snprintf(ptmp, SCRIPT_LUA_BUFSIZ, "local %s = %1.15g ", child->string, val);
            }

            if (plen < 0)
                plen = 0;
            else if (plen >= SCRIPT_LUA_BUFSIZ) {
                dbg_syslog(LOG_ERR, "Fatal warning, buffer overflow: %d", plen);
                plen = SCRIPT_LUA_BUFSIZ - 1;
                ptmp[plen] = '\0';
            }
            if (plen > 0) {
                size_t newlen = buflen;
                newlen += (size_t) plen;
                if (newlen >= (2 * SCRIPT_LUA_BUFSIZ)) {
                    dbg_syslog(LOG_ERR, "Fatal Error, buffer overflow: %u, %u",
                        (unsigned int) plen, (unsigned int) newlen);
                } else {
                    strcpy(buf_p + buflen, ptmp);
                    buflen = newlen;
                }
            }
        }
    }
    if (!strstr(param->desc, "return")) //如果物模型中表达式没有return 加return
    {
        int plen;
        plen = snprintf(ptmp, SCRIPT_LUA_BUFSIZ, "return %s ", param->desc);
        if (plen > 0) {
            strcpy(buf_p + buflen, ptmp);
            buflen += (size_t) plen;
        }
    } else {
        strcat(buf_p, param->desc);
    }
    if (luaL_dostring(var->lua, buf_p))
    {
        dbg_syslog(LOG_ERR, "luaL_loadstring %s error!!!", buf_p);
        goto _out;
    }
    if (param->raw_data_type != TAG_TYPE_VLENGTH)
    {
        double tagD = lua_tonumber(var->lua, -1);
        cJSON_DeleteItemFromObject(jtag_new, param->sign_mark);
        cJSON_AddNumberToObject(jtag_new, param->sign_mark, tagD);
        dbg_syslog(LOG_INFO, "operation %s sign_mark %s val %1.15g ", buf_p, param->sign_mark, tagD);
    }
    else
    {
        const char *tagStr = lua_tostring(var->lua, -1);
        if (tagStr)
        {
            cJSON_DeleteItemFromObject(jtag_new, param->sign_mark);
            cJSON_AddStringToObject(jtag_new, param->sign_mark, tagStr);
        }
        dbg_syslog(LOG_INFO, "operation %s identifier %s val %s ", buf_p, param->sign_mark, tagStr);
    }

_out:
    lua_pop(var->lua, 1);
    free(buf_p);
    if (ptmp != NULL)
        free(ptmp);
    if (pstr != NULL)
        free(pstr);
    return 0;
}

static void exec_encode_script_lua(pp_var_t *var, lua_State *lua, char *file_name, char *json_str, unsigned short len, char *bin, int *index, encode_info_t *encode_info)
{
    size_t jlen = 0;
    char *bin_str = NULL;
    long error_code = 0;
    unsigned short json_len = 0;
    const unsigned char * jdat = NULL;
    int tmp_index = *index;
    int i;
    int ret = 0;

    lua_getglobal(lua, FUN_PROTOCOL_ENCODE_NAME);
    lua_pushstring(lua, json_str);
    lua_pushinteger(lua, len);
    ret = lua_pcall(lua, 2, 3, 0);
    if (ret)
    {
        const char *err_msg = lua_tostring(lua, -1);
        dbg_syslog(LOG_ERR, "lua error:%s", err_msg);
        *index = -1;
        lua_pop(lua, 1);
        goto _out;
    }

    json_len = lua_tointeger(lua, -1);
    error_code = lua_tointeger(lua, -3);

    dbg_syslog(LOG_DEBUG, "encode file_name %s error_code %ld json_len %d",
        file_name, error_code, (int) json_len);
    if ((error_code >> 12) & 0x01)
    {
        encode_info->stream_direction = 1;
    }
    if ((error_code >> 13) & 0x01)
    {
        encode_info->no_base64 = 1;
    }
    if ((error_code >> 14) & 0x01)
    {
        encode_info->no_bin = 1;
    }

    error_code = error_code & 0xFFF;
    if (error_code != 0 || json_len < 0 || json_len > 24576)
    {
        dbg_syslog(LOG_ERR, "error_code: %ld json_len:%d", error_code, (int) json_len);
        *index = -1;
        lua_pop(lua, 3);
        goto _out;
    }

    bin_str = calloc(json_len + 1, 1);
    if (bin_str == NULL)
    {
        dbg_syslog(LOG_ERR, "calloc fail json_len %d !!!", json_len);
        *index = -1;
        lua_pop(lua, 3);
        goto _out;
    }

    /*
     * jdat might contain NUL bytes ('\0'),
     * cannot use `sprintf(bin_str, "%s", ...)
     */
    jdat = (const unsigned char *) lua_tolstring(lua, -2, &jlen);
    if (jdat != NULL && json_len > 0) {
        if (jlen >= (size_t) json_len) {
            memcpy(bin_str, jdat, json_len);
        } else {
            memcpy(bin_str, jdat, jlen);
        }
    }
    lua_pop(lua, 3);

    if (!error_code && json_len)
    {
        if (encode_info->no_base64 == 0 && encode_info->stream_direction == 0 && encode_info->no_bin == 0)
        {
            for (i = 0; i < json_len / 2; i++)
            {
                sscanf(bin_str + i * 2, "%02x", &bin[tmp_index + i]);
                //dbg_syslog(LOG_INFO, "bin[%d] %x", tmp_index + i, bin[tmp_index + i]);
            }
            *index = tmp_index + json_len / 2;
        }
        else
        {
            *index = json_len;
            memcpy(bin, bin_str, json_len);
            bin[json_len] = '\0';
        }
    }
_out:
    if (bin_str)
        free(bin_str);
    return;
}

static void exec_encode_script_python(pp_var_t *var, char *file_name, char *json_str, unsigned short len, char *bin, int *index, encode_info_t *encode_info)
{
    PyObject *pName = NULL;
    PyObject *pModule = NULL;
    PyObject *pDict = NULL;
    PyObject *pFunc = NULL;
    PyObject *pArgs = NULL;
    int is_string = 0;
    unsigned char tmp_index = *index;
    int i;
    char *bin_str;
    long json_len = 0, error_code = 0;

    //加载名为py_add的python脚本
#if _PP_USE_PYTHON3
    pName = PyUnicode_FromString(file_name);
#else
    pName = PyString_FromString(file_name);
#endif
    pModule = PyImport_Import(pName);
    if (!pModule)
    {
        dbg_syslog(LOG_WARNING, "Load %s failed!", file_name);
        return;
    }
    pDict = PyModule_GetDict(pModule);
    if (!pDict)
    {
        dbg_syslog(LOG_WARNING, "Can't find dict in %s!", file_name);
        return;
    }

    pFunc = PyDict_GetItemString(pDict, FUN_PROTOCOL_ENCODE_NAME);
    if (!pFunc || !PyCallable_Check(pFunc))
    {
        dbg_syslog(LOG_WARNING, "Can't find function!\n");
        return;
    }

    pArgs = PyTuple_New(2);
    PyTuple_SetItem(pArgs, 0, Py_BuildValue("s", json_str));
    PyTuple_SetItem(pArgs, 1, Py_BuildValue("i", len));

    PyObject *presult = PyObject_CallObject(pFunc, pArgs);
    PyArg_ParseTuple(presult, "isi", &error_code, &bin_str, &json_len);
    dbg_syslog(LOG_DEBUG, "encode file_name %s error_code %ld bin_str %s json_len %d",
        file_name, error_code, bin_str, (int) json_len);
    //清理python对象
    if (pName)
    {
        Py_DECREF(pName);
    }
    if (pArgs)
    {
        Py_DECREF(pArgs);
    }
    if (pModule)
    {
        Py_DECREF(pModule);
    }

    if ((error_code >> 12) & 0x01)
    {
        encode_info->stream_direction = 1;
    }
    if ((error_code >> 13) & 0x01)
    {
        encode_info->no_base64 = 1;
    }
    if ((error_code >> 14) & 0x01)
    {
        is_string = 1;
    }

    error_code = error_code & 0xFFF;
    if (!error_code && json_len)
    {
        if (is_string == 0 && encode_info->no_base64 == 0 && encode_info->stream_direction == 0)
        {
            for (i = 0; i < json_len / 2; i++)
            {
                sscanf(bin_str + i * 2, "%02x", &bin[tmp_index + i]);
                //dbg_syslog(LOG_INFO, "bin[%d] %x", tmp_index + i, bin[tmp_index + i]);
            }
            *index = tmp_index + json_len / 2;
        }
        else
        {
            *index = json_len;
            strncpy(bin, bin_str, json_len);
            bin[json_len] = '\0';
        }
    }

    return;
}

static void exec_encode_script_javascript(pp_var_t *var, char *file_name, char *json_str, unsigned short len, char *bin, int *index, encode_info_t *encode_info)
{
    char file[256] = {0};
    char param_json[512] = {0};
    char buf[1024] = {0};
    char bin_str[512] = {0}, json_len = 0, error_code = 0;
    unsigned char tmp_index = *index;
    int i, ret;

    sprintf(file, "%s%s %s ", PARSE_SCRIPT_DIR, file_name, FUN_PROTOCOL_ENCODE_NAME);
    strcat(buf, file);
    sprintf(param_json, "%s %d", json_str, len);
    strcat(buf, param_json);
    dbg_syslog(LOG_INFO, "js_send:%s", buf);

    ret = send(var->javascript.fd, buf, strlen(buf) + 1, 0);
    memset(buf, 0, sizeof(buf));
    ret = recv(var->javascript.fd, buf, sizeof(buf), 0);
    if (ret > 0)
    {
        dbg_syslog(LOG_INFO, "ret %d js_recv: %s", ret, buf);

        char *p1 = strstr(buf, " ");
        if (p1)
        {
            memcpy(bin_str, buf, p1 - buf);
            error_code = atoi(bin_str);
            memset(bin_str, 0, sizeof(bin_str));
        }

        char *p2 = strstr(p1 + 1, " ");
        if (p2)
        {
            memcpy(bin_str, p1 + 1, p2 - p1);
        }

        json_len = atoi(p2 + 1);

        dbg_syslog(LOG_INFO, "file %s error_code %d json_str %s json_len %d", file, error_code, bin_str, json_len);
        if (!error_code && json_len)
        {
            for (i = 0; i < json_len / 2; i++)
            {
                sscanf(bin_str + i * 2, "%02x", &bin[tmp_index + i]);
                dbg_syslog(LOG_INFO, "bin[%d] %x", tmp_index + i, bin[tmp_index + i]);
            }
            *index = tmp_index + json_len / 2;
        }
    }
    else if (var->javascript.connect_timer == 0)
    {
        var->javascript.connected = 0;
        var->javascript.connect_timer = my_timer_create();
        if (var->javascript.connect_timer > 0)
        {
            my_timer_set(var->javascript.connect_timer, 1, 1 * 1000);
        }

        if (var->javascript.fd)
        {
            close(var->javascript.fd);
            var->javascript.fd = 0;
        }
    }

    return;
}
void protocol_encode_script_service(pp_var_t *var, template_cfg_t *ptemplate, cJSON *node, char *cmd, int *index, encode_info_t *encode_info)
{
    //char *file = GetFileName(file_name);
    char *json_str = cJSON_PrintUnformatted(node);
    char *file_name = ptemplate->parser_script;

    if (file_name == NULL)
    {
        goto out;
    }

    dbg_syslog(LOG_DEBUG, "file_name %s, json_str %s", file_name, json_str);
    if (strstr(file_name, LUA_EXTEND))
    {
        exec_encode_script_lua(var, ptemplate->script.lua, file_name, json_str, strlen(json_str), cmd, index, encode_info);
    }
    else if (strstr(file_name, PYTHON_EXTEND))
    {
        char file[128] = {0};
        char *file_t = GetFileName(file_name);
        char *p = strstr(file_t, PYTHON_EXTEND);

        strncpy(file, file_t, p - file_t);
        dbg_syslog(LOG_INFO, "file %s", file);
        exec_encode_script_python(var, file, json_str, strlen(json_str), cmd, index, encode_info);
    }
    else if (strstr(file_name, JAVASCRIPT_EXTEND))
    {
        exec_encode_script_javascript(var, file_name, json_str, strlen(json_str), cmd, index, encode_info);
    }
    else
    {
        dbg_syslog(LOG_WARNING, "file_name %s extend error!!!", file_name);
    }
out:
    free(json_str);
}

void protocol_encode_template_service(pp_var_t *var, template_cfg_t *ptemplate, service_cfg_t *pservice, node_cfg_t *pnode, cJSON *node, gw_protocol_e protocol, gw_port_e port, char *cmd, int *index)
{
    int k;
    char * pbuf = NULL;
    int tmp_index = *index;
    int data_index = *index;
    char data_type[16] = {0, 2, 2, 4, 4, 4, 2, 4, 2, 4, 4, 1, 1, 8, 8, 0};

    pbuf = (char *) malloc(SCRIPT_LUA_BUFSIZ);
    if (pbuf == NULL) {
        dbg_syslog(LOG_ERR, "System out of memory!\n");
        return;
    }

    //遍历服务功能码、组织下发数据包
    for (k = 0; k < pservice->param_tab->paramCnt; k++)
    {
        //根据数据类型获取k v,按位组织数据
        unsigned int val = 0xFFFFFFFF;
        //dbg_syslog(LOG_INFO, "protocol %d,sign_mark %s", ptemplate->protocol, pservice->param_tab->param[k].sign_mark);
        if (ptemplate->protocol != GW_PROTC_MODBUS && ptemplate->protocol != GW_PROTC_DYWL_LORA_V1 && !strcmp(pservice->param_tab->param[k].sign_mark, CRC_STRING))
        {
            pp_data_exec_crc_info(var, ptemplate, cmd + data_index, tmp_index - data_index, &val);
        }
        else
        {
            if (!strcmp(pservice->param_tab->param[k].sign_mark, CRC_STRING))
            {
                continue;
            }
            if (pservice->param_tab->param[k].desc != NULL && pservice->param_tab->param[k].desc[0])
            {
                exec_encode_operation_script_lua(var, node, node, &pservice->param_tab->param[k]);
            }

            GET_JSON_VALUE_INT(node, pservice->param_tab->param[k].sign_mark, val);
            //dbg_syslog(LOG_INFO, "paramCnt %d k %d index %d sign_mark %s val 0x%x data_type %d", pservice->param_tab->paramCnt, k, tmp_index, pservice->param_tab->param[k].sign_mark, val, pservice->param_tab->param[k].raw_data_type);
            if (val == 0xFFFFFFFF)
            {
                if (!strcmp(pservice->param_tab->param[k].sign_mark, "term_addr"))
                {
                    memcpy(cmd + tmp_index, pnode->term_addr, data_type[pservice->param_tab->param[k].raw_data_type]);
                    if (data_type[pservice->param_tab->param[k].raw_data_type] > 1)
                    {
                        //dbg_syslog(LOG_INFO, "sign_mark %s val 0x%x.%x data_type %d", pservice->param_tab->param[k].sign_mark, cmd[tmp_index], cmd[tmp_index + 1]);
                    }
                    else
                    {
                        //dbg_syslog(LOG_INFO, "sign_mark %s val 0x%x data_type %d", pservice->param_tab->param[k].sign_mark, cmd[tmp_index]);
                    }
                    tmp_index += data_type[pservice->param_tab->param[k].raw_data_type];
                    continue;
                }
                else
                {
                    val = pservice->param_tab->param[k].sign_val;
                }
                //GET_JSON_VALUE_INT(service_cmd,var->dev_tb.pnode[i]->pservice[j]->pparam[k]->sign_mark,val);
                //dbg_syslog(LOG_INFO, "val %d service_cmd paramCnt %d k %d index %d sign_mark %s val 0x%x data_type %d", val, pservice->param_tab->paramCnt, k, tmp_index, pservice->param_tab->param[k].sign_mark, val, pservice->param_tab->param[k].raw_data_type);
            }

            if (!strcmp(pservice->param_tab->param[k].sign_mark, CRC_PADDING))
            {
                data_index += data_type[pservice->param_tab->param[k].raw_data_type];
            }
        }
        //dbg_syslog(LOG_INFO, "sign_mark %s raw_data_type %d val 0x%x", pservice->param_tab->param[k].sign_mark, pservice->param_tab->param[k].raw_data_type, val);
        switch (pservice->param_tab->param[k].raw_data_type)
        {
        case TAG_TYPE_UINT16I:
        case TAG_TYPE_INT16I:
        case TAG_TYPE_INT32I:
        case TAG_TYPE_UINT32I:
        case TAG_TYPE_FLOAT:
        case TAG_TYPE_FLOATI:
        {
            int f;
            for (f = 0; f < data_type[pservice->param_tab->param[k].raw_data_type]; f++)
            {
                cmd[tmp_index + f] = val >> ((data_type[pservice->param_tab->param[k].raw_data_type] - f - 1) * 8) & 0xFF;
            }
        }
        break;
        case TAG_TYPE_VLENGTH:
        {
            size_t flen;
            pbuf[0] = '\0';
            GET_JSON_VALUE_STRING(node, pservice->param_tab->param[k].sign_mark, pbuf);
            flen = pbuf[0] ? strlen(pbuf) : 0;
            if (flen > 0) {
                memcpy(cmd + tmp_index, pbuf, flen);
                tmp_index += flen;
            }
        }
        break;
        default:
            memcpy(&cmd[tmp_index], (char *)(&val), data_type[pservice->param_tab->param[k].raw_data_type]);
            break;
        }
        tmp_index += data_type[pservice->param_tab->param[k].raw_data_type];
    }

    if (protocol == GW_PROTC_MODBUS)
    {
        if (port == LORA_1 || port == LORA_DTU_RS485)
        {
            tmp_index = calculate_crc((unsigned char *)cmd + 3, tmp_index - 3);
            tmp_index += 3;
        }
        else
        {
            tmp_index = calculate_crc((unsigned char *)cmd, tmp_index);
        }
    }

    *index = tmp_index;
    free(pbuf);
}

unsigned int pp_get_template_term_addr(pp_var_t *var, template_cfg_t *ptemplate_id, unsigned char *term_addr, unsigned char term_addr_len, char *buf)
{
    int j;
    /* char data_type[16] = {0, 2, 2, 4, 4, 4, 2, 4, 2, 4, 4, 1, 1, 8, 8, 0}; */

    for (j = 0; j < term_addr_len; j++)
    {
        char tmp[4] = {0};
        sprintf(tmp, "%02X", term_addr[j]);
        strcat(buf, tmp);
    }
    return 0;
}

int pp_send_regulate_signal(pp_var_t *var, pp_regulate_signal_t *regulate_data, int no_base64)
{
    char *data_tmp = NULL;
    char *buf = NULL;
    size_t maxlen;
    cJSON *rs_data = cJSON_CreateObject();
    char topic[TOPIC_MAX_LEN] = {0};

    maxlen = MAXBUF;
    if (maxlen < (size_t) regulate_data->len) {
        maxlen = (size_t) regulate_data->len;
        maxlen += 1;
        /* maximum length: 32 KB: */
        if (maxlen > 0x8000) {
            /* TODO: emit a warning, buffer overflow! */
            maxlen = 0x8000;
        }
    }
    buf = (char *)calloc(1, maxlen);
    if (buf == NULL)
    {
        dbg_syslog(LOG_ERR, "malloc fail");
        return -1;
    }

    if (no_base64 == 0)
        b64_encode(regulate_data->data, regulate_data->len, buf, MAXBUF);
    else
        memcpy(buf, regulate_data->data, regulate_data->len);
    cJSON_AddStringToObject(rs_data, "data_b64", buf);
    cJSON_AddNumberToObject(rs_data, "len", regulate_data->len);
    cJSON_AddNumberToObject(rs_data, "period", regulate_data->period);
    cJSON_AddStringToObject(rs_data, "port", port_enum2char(regulate_data->port));
    cJSON_AddNumberToObject(rs_data, "mi", regulate_data->mi);
    cJSON_AddStringToObject(rs_data, "src_identifier", regulate_data->src_identifier);
    cJSON_AddStringToObject(rs_data, "sn", regulate_data->sn);
    cJSON_AddStringToObject(rs_data, "dtu_sn", regulate_data->dtu_sn);
    cJSON_AddNumberToObject(rs_data, "protocol", regulate_data->protocol);
    cJSON_AddNumberToObject(rs_data, "communication_timeout", regulate_data->communication_timeout);
    cJSON_AddStringToObject(rs_data, "term_addr", (const char *) regulate_data->term_addr);
    cJSON_AddStringToObject(rs_data, "tcp_ip_addr", regulate_data->tcp_ip_addr);
    cJSON_AddNumberToObject(rs_data, "tcp_port", regulate_data->tcp_port);
    if (regulate_data->requester[0] != '\0')
        cJSON_AddStringToObject(rs_data, "requester", regulate_data->requester);

    data_tmp = cJSON_Print(rs_data);
    if (regulate_data->app_key[0] != '\0')
    {
        snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/%s", var->sn_str, regulate_data->app_key,
                 regulate_data->sn, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
    }
    else
    {
        snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/%s", var->sn_str, port_enum2char(regulate_data->port),
                 regulate_data->sn, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
    }

    ipc_session_publish(var->session, topic, (unsigned char *) data_tmp, (int) strlen(data_tmp));

    // 回收资源
    free(data_tmp);
    free(buf);
    cJSON_Delete(rs_data);

    return 0;
}

int pp_regulate_signal_processor(pp_var_t *var, template_cfg_t *ptemplate, cJSON *node, node_cfg_t *node_cfg, pp_regulate_signal_t *rglt_data)
{
    encode_info_t encode_info = {0};
    int find = 0;
    char *cmd = calloc(16360, 1);
    int index = 0;
    int j = 0;
    int ret = 0;

    if (cmd == NULL)
    {
        dbg_syslog(LOG_ERR, "calloc fail len 12288 !!!");
        ret = -1;
        goto out;
    }

    //是否为LORA DTU
    if (node_cfg->depth && node_cfg->dtu_sn && node_cfg->port == LORA_DTU_RS485)
    {
        for (j = 0; j < var->nodes_cfg_table->node_cnt; j++)
        {
            if (strcmp(node_cfg->dtu_sn, var->nodes_cfg_table->node[j].sn) == 0 && var->nodes_cfg_table->node[j].port == LORA_1)
            {
                dbg_syslog(LOG_INFO, "dtu_sn %s", node_cfg->dtu_sn);
                memcpy(&cmd[index], var->nodes_cfg_table->node[j].term_addr, 2);
                index += 2;
                //DTU透传命令码
                cmd[index] = DTU_TT_DOWN;
                index += 1;
                break;
            }
        }
        if (index <= 0)
        {
            dbg_syslog(LOG_ERR, "Can't find DTU or PORT ERROR !!!");
            ret = -1;
            goto out;
        }
    }

    if (ptemplate->use_parser_url == SCRIPT_PARSER)
    {
        char term_addr[64] = {0};
        pp_get_template_term_addr(var, ptemplate, node_cfg->term_addr, node_cfg->term_addr_len, term_addr);
        if (node_cfg->tcp_ip_addr[0] && node_cfg->tcp_port)
        {
            cJSON_AddStringToObject(node, "ip_addr", node_cfg->tcp_ip_addr);
            cJSON_AddNumberToObject(node, "port", node_cfg->tcp_port);
        }
        if (node_cfg->ext_data && node_cfg->ext_data[0])
        {
            cJSON_AddStringToObject(node, "ext_data", node_cfg->ext_data);
        }
        cJSON_AddStringToObject(node, "term_addr", term_addr);
        protocol_encode_script_service(var, ptemplate, node, cmd, &index, &encode_info);
        find = 1;
    }
    else if (ptemplate->use_parser_url == SCRIPT_NONE)
    {
        int j = 0;
        for (j = 0; j < ptemplate->service_tab->serviceCnt; j++)
        {
            if (ptemplate->service_tab->service[j].direction == 0 && !strcmp(ptemplate->service_tab->service[j].identifier, rglt_data->src_identifier))
            {
                protocol_encode_template_service(var, ptemplate, &ptemplate->service_tab->service[j], node_cfg, node, ptemplate->protocol, node_cfg->port, cmd, &index);
                find = 1;
                break;
            }
        }
        if (j >= ptemplate->service_tab->serviceCnt)
        {
            dbg_syslog(LOG_WARNING, "identifier %s error !!!!", rglt_data->src_identifier);
            ret = -1;
            goto out;
        }
    }

    if ((index) < 0)
    {
        dbg_syslog(LOG_WARNING, "data encode error len %d !!!", index);
        ret = -1;
        goto out;
    }

    if (encode_info.stream_direction == 0)
    {
        rglt_data->data = (unsigned char *)cmd;
        rglt_data->len = index;
        rglt_data->port = node_cfg->port;
        rglt_data->communication_timeout = ptemplate->communication_timeout;
        rglt_data->protocol = ptemplate->protocol;
        strcpy(rglt_data->app_key, node_cfg->app_key);
        strcpy(rglt_data->sn, node_cfg->sn);
        strcpy(rglt_data->tcp_ip_addr, node_cfg->tcp_ip_addr);
        rglt_data->tcp_port = node_cfg->tcp_port;
        if (node_cfg->depth)
        {
            strcpy(rglt_data->dtu_sn, node_cfg->dtu_sn);
        }
        i82hexstr(node_cfg->term_addr, (char *) rglt_data->term_addr, 8);
        if (index < 512)
            dbg_syslog_hex(LOG_INFO, rglt_data->data, index, "regulate signal parser len %d port %d period %d mi %d src_identifier %s", index, rglt_data->port, rglt_data->period, rglt_data->mi, rglt_data->src_identifier);
        else
            dbg_syslog(LOG_INFO, "regulate signal parser len %d port %d period %d mi %d src_identifier %s", index, rglt_data->port, rglt_data->period, rglt_data->mi, rglt_data->src_identifier);
        pp_send_regulate_signal(var, rglt_data, encode_info.no_base64);
    }
    else if (index > 0)
    {
        tag_table_t ptag = {0};
        tag_table_t ptag_post = {0};

        strcpy(ptag.sn, node_cfg->sn);
        ptag.mi = rglt_data->mi;
        strncpy(ptag.identifier, rglt_data->src_identifier, sizeof(ptag.identifier));
        ptag.report_period = ptemplate->report_period;
        ptag.data_type = DATA_TYPE_SERVICE;
        ptag.tag_node = calloc(index + 1, 1);
        memcpy(ptag.tag_node, cmd, index);
        ptag.time = time(NULL);

        memcpy(&ptag_post, &ptag, sizeof(tag_table_t));
        strcpy(ptag_post.identifier, "post");
        ptag_post.data_type = DATA_TYPE_PROPERTY;
        ptag_post.tag_node = calloc(index + 1, 1);
        memcpy(ptag_post.tag_node, cmd, index);

        dbg_syslog(LOG_DEBUG, "parse result:%s post:%s", ptag.tag_node, ptag_post.tag_node);
        if (ptag.tag_node != NULL)
        {
            ptag.port = node_cfg->port;
            strncpy(ptag.app_key, node_cfg->app_key, MAX_APP_KEY_LEN);
            pp_send_tag_data(var->session, &ptag);
            free(ptag.tag_node);
        }
        if (ptag_post.tag_node != NULL)
        {
            ptag_post.port = node_cfg->port;
            strncpy(ptag_post.app_key, node_cfg->app_key, MAX_APP_KEY_LEN);
            pp_send_tag_data(var->session, &ptag_post);
            free(ptag_post.tag_node);
        }
    }
    ret = find;

out:
    if (cmd)
        free(cmd);
    free(rglt_data);
    cJSON_Delete(node);
    return ret;
}

static void pp_transparent_processor(pp_var_t *var, pp_regulate_signal_t *regulate_data, node_cfg_t *node_cfg, char *cmd, int len)
{
    //返回要发送的数据,暂时直接发送给LORA模块
    regulate_data->data = (unsigned char *)cmd;
    regulate_data->len = len;
    regulate_data->port = node_cfg->port;
    strcpy(regulate_data->sn, node_cfg->sn);
    strcpy(regulate_data->tcp_ip_addr, node_cfg->tcp_ip_addr);
    strcpy(regulate_data->app_key, node_cfg->app_key);
    regulate_data->tcp_port = node_cfg->tcp_port;

    i82hexstr(node_cfg->term_addr, (char *) regulate_data->term_addr, 8);
    pp_send_regulate_signal(var, regulate_data, 0);
}

static int pp_regulate_signal_parser(pp_var_t *var, char *rsData)
{
    int i = 0;
    int ret = 1;
    template_cfg_t *ptemplate = NULL;

    pp_regulate_signal_t *rglt_data = NULL;
    cJSON *data_node = NULL;

    rglt_data = calloc(1, sizeof(pp_regulate_signal_t));
    if (rglt_data == NULL)
    {
        dbg_syslog(LOG_ERR, "malloc failed");
        return -1;
    }

    data_node = pp_get_rglt_data_from_json_str(var, rglt_data, rsData);
    if (!data_node)
    {
        dbg_syslog(LOG_WARNING, "json rsData %s error!!!", rsData);
        free(rglt_data);
        return -1;
    }

    if (rglt_data->src_identifier[0] == '\0')
    {
        dbg_syslog(LOG_WARNING, "identifier %s error!!!", rglt_data->src_identifier);
        free(rglt_data);
        cJSON_Delete(data_node);
        return -1;
    }

    if (strcmp(rglt_data->src_identifier, "__Upgrade") == 0)
    {
        char topic[TOPIC_MAX_LEN] = {0};

        snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/%s", var->sn_str, var->proc_name, TOPIC_UPGRADE_NODE);

        ipc_session_publish(var->session, topic, (unsigned char *) rsData, (int) strlen(rsData));
        free(rglt_data);
        cJSON_Delete(data_node);
        return 1;
    }
    //遍历设备列表
    for (i = 0; i < var->nodes_cfg_table->node_cnt; i++)
    {
        node_cfg_t *node_cfg = &var->nodes_cfg_table->node[i];
        if (strcmp(rglt_data->sn, node_cfg->sn) == 0)
        {
            //添加透传通道
            if (strcmp(rglt_data->src_identifier, "transparent") == 0 || strcmp(rglt_data->src_identifier, "__transparent") == 0)
            {
                int len = 0;
                char *tmp = (char *)malloc(MAXBUF);
                char *raw_data = (char *)malloc(MAXBUF);

                if (tmp == NULL || raw_data == NULL)
                {
                    dbg_syslog(LOG_ERR, "malloc fail");
                    ret = -1;
                    free(tmp);
                    free(raw_data);
                    goto out;
                }

                GET_JSON_VALUE_STRING(data_node, "raw_data", tmp);
                len = b64_decode(tmp, raw_data, MAXBUF);

                pp_transparent_processor(var, rglt_data, node_cfg, raw_data, len);
                free(tmp);
                free(raw_data);
                ret = 0;
                goto out;
            }

            ptemplate = pp_find_template(var->template_table, node_cfg->template_id);
            if (ptemplate)
            {
                if (ptemplate->use_parser_url == SCRIPT_PARSER)
                {
                    script_data_t script_data;
                    script_data.data_encode.data = data_node;
                    script_data.data_encode.rglt_data = rglt_data;
                    if (pp_send_data_to_script(var, ptemplate, &script_data, node_cfg, SCRIPT_ENCODE) == 0) {
                        rglt_data = NULL;
                        data_node = NULL;
                    }
                    ret = 1;
                }
                else if (ptemplate->use_parser_url == SCRIPT_NONE)
                {
                    ret = pp_regulate_signal_processor(var, ptemplate, data_node, node_cfg, rglt_data);
                    rglt_data = NULL;
                    data_node = NULL;
                }
            }
            break;
        }
    }

    if (rglt_data != NULL)
        free(rglt_data);
    if (data_node != NULL)
        cJSON_Delete(data_node);
    return ret;

out:
    free(rglt_data);
    cJSON_Delete(data_node);
    return ret;
}

int msg_set_regulate_signal_val(pp_var_t *var, ipc_msg_t *msg)
{
    int ret;

    char topic[TOPIC_MAX_LEN] = {0};

    //控制命令下发解析
    ret = pp_regulate_signal_parser(var, msg->payload);
    cJSON *node = cJSON_CreateObject();
    if (node)
    {
        unsigned int UTC = (unsigned int)time(0);
        cJSON *root = cJSON_Parse(msg->payload);
        if (root)
        {
            char sn[SN_MAX_LEN] = {0};
            GET_JSON_VALUE_STRING(root, "sn", sn);
            cJSON_AddStringToObject(node, "sn", sn);
            cJSON_Delete(root);
        }

        cJSON_AddNumberToObject(node, "mi", json_pkt_json_mi(msg->payload));
        cJSON_AddNumberToObject(node, "UTC", UTC);
        cJSON_AddNumberToObject(node, "status", ret);
    }
    char *rt = cJSON_Print(node);
    //回复上位机消息
    if (strstr(msg->topic, TOPIC_EVT_SET_RGLT))
    {
        replace_sub_str(msg->topic, TOPIC_EVT_SET_RGLT, TOPIC_RSP_RGLT, topic);
    }
    else
    {
        replace_sub_str(msg->topic, "/service/", "/service_rsp/", topic);
        topic[0] = 'G';
    }
    ipc_session_publish(var->session, topic, (unsigned char *) rt, (int) strlen(rt));
    free(rt);
    cJSON_Delete(node);

    return 0;
}