AlarmModule/core/handle.upload.c

/******************************************************************
** Copyright (c) 2009  Wuhan xiyun Technology Co., Ltd .
** FileName:	handle.upload.c
** Author:	 	chw
** Mail:		
** Editor:		chw
** Date:		2020-12-03 13:34
** Version: 	2020-12-03
** File Description:  处理mqtt的设备数据上传 包含:1.设备连接相关指令  2.设备电器或传感器数据上传
** Param explain: char* mac : 网关MAC地址
   char* subMac             ：子节点MAC地址
  MQTTAsync_message *pubmsg : 消息结构体
  MYSQL *_db                ：数据库连接
  char *topicName           : 消息主题
******************************************************************/

//查询指令发送表是否有数据
int is_have_data_inDB(char* mac,MYSQL *_db)
{
    MYSQL_RES* res = NULL;
    int row_count;
    char query_sql[512] = {0};
    sprintf(query_sql,"select id from dev_cmd_send where device_mac = '%s'",mac);
    excuteSql(_db,query_sql);
    res = mysql_store_result(_db);
    if (NULL == res)
    {
        return 1;
    }
    //获取行数
    row_count = mysql_num_rows(res);
    if(row_count<1)
    {
        mysql_free_result(res);
        return 1;
    }
    mysql_free_result(res);

    return 0;
}

//在数组中查询mac是否在二维数组中
int isexist_mac(char* subMac,char string_subMac[][13],int count)
{
    int i = 0;
    for(i = 0;i<count;i++)
    {
        if(strcmp(subMac,string_subMac[i]) == 0)
        {
            log("%s 子节点重复,跳过! \n",subMac);
            return 1;
        } 
    }
    return 0;
}

//------  连接指令  -------------------------------------------------------

//设备上电
 int  do_CN_FTC(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
 {
    //debug("topicName=%s payloadlen=%d\n", topicName, pubmsg->payloadlen);
    UNUSED(topicName);
    
    //校验长度
    if(pubmsg->payloadlen != 18 && pubmsg->payloadlen != 19)
        return 1;
    
    char update_sql[512] = {0};
    char update_table_cmd_send[100];
    char mac[25] = {0};
    char gateway_version[10] = {0};

    unsigned char* data = pubmsg->payload;

    Hex2Str((const char*)data,mac,12);
    memcpy(gateway_version,data+12,pubmsg->payloadlen-12);
    log("↑ %s 上电,网关版本号:%s \n",mac,gateway_version);

    
    sprintf(update_sql,"update dev_info_gateway a ,dev_status b set a.device_version = '%s',a.last_poweron_time = now(),a.version_update_time = now(),b.current_online = 1,b.update_time = now() "
                        "where a.device_mac = '%s' and a.device_mac = b.device_mac",gateway_version,mac);
    excuteSql(_db,update_sql);

    //更新指令发送表设备在线状态
    if(is_have_data_inDB(mac,_db) == 0)
    {
        sprintf(update_table_cmd_send,"update dev_cmd_send set current_online = 1 where device_mac = '%s'",mac);
        excuteSql(_db,update_table_cmd_send);
        log("%s 数据库有指令,更新指令发送表设备在线状态为1 \n",mac);
    }

    return 0;
 }


 //设备重连
 int  do_CN_RCN(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
 {
    //debug("topicName=%s payloadlen=%d\n", topicName, pubmsg->payloadlen);
    UNUSED(topicName);

    //校验长度
    if(pubmsg->payloadlen != 18 && pubmsg->payloadlen != 19)
        return 1;

    char update_sql[512] = {0};
    char update_table_cmd_send[100];
    char mac[25] = {0};
    char gateway_version[10] = {0};

    unsigned char* data = pubmsg->payload;
    Hex2Str((const char*)data,mac,12);
    memcpy(gateway_version,data+12,pubmsg->payloadlen-12);
    log("↑ %s 重连,网关版本号:%s \n",mac,gateway_version);

    sprintf(update_sql,"update dev_info_gateway a ,dev_status b set a.device_version = '%s',a.last_reconnection_time = now(),a.version_update_time = now(),b.current_online = 1,b.update_time = now() "
                        "where a.device_mac = '%s' and a.device_mac = b.device_mac",gateway_version,mac);
    excuteSql(_db,update_sql);

    //更新指令发送表设备在线状态
    if(is_have_data_inDB(mac,_db) == 0)
    {
        sprintf(update_table_cmd_send,"update dev_cmd_send set current_online = 1 where device_mac = '%s'",mac);
        excuteSql(_db,update_table_cmd_send);
        log("%s 数据库有指令,更新指令发送表设备在线状态为1 \n",mac);
    }

    return 0;
 }


//------  设备数据上传指令  ----------------------------------------------------

//猪称重数据
int do_DT_PIG(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
{
    UNUSED(topicName);
    UNUSED(_db);

    //校验长度,变长只比较最小
    if(pubmsg->payloadlen<15)
        return 1;

    char insert_sql[512] = {0};
    char update_sql[512] = {0};
    char log_str[512] = {0};
    char mac[25] = {0};
    char ear_id[15];    //耳标
    float f_weight = 0;//体重
    char2int ciWeight;
    char stime[64] = {0};//时间
    int affectedRows = 0;

    //mac
    unsigned char* data = pubmsg->payload;
    //Hex2Str((const char*)data,mac,12);
    memcpy(mac,data,12);
    

    //时间
    sprintf(stime,"%d/%d/%d %d:%d:%d",data[12],data[13],data[14],data[15],data[16],data[17]);

    //耳标
    memcpy(ear_id,data+18,15);

    //体重
    memcpy(ciWeight.c_char,data+33,4);
    EndianSwap((unsigned char*)ciWeight.c_char,0,4);
    f_weight = ((float)ciWeight.i_int)/10.0;

    sprintf(insert_sql,"insert into dev_pig_history_202205(device_mac,acqusition_time,ear_id,weight,update_time)values('%s',str_to_date('%s', '%%Y-%%m-%%d %%h:%%i:%%s'),'%s',%f,now())",
                        mac,stime,ear_id,f_weight);
    excuteSql(_db,insert_sql);

    sprintf(update_sql,"update dev_pig_status set weight = %f, acqusition_time = str_to_date('%s', '%%Y-%%m-%%d %%h:%%i:%%s'),update_time = now() where device_mac = '%s' and ear_id = '%s'",
                        f_weight,stime,mac,ear_id);
    excuteSql(_db,update_sql);
    affectedRows = mysql_affected_rows(_db);
    if (affectedRows < 1)
    {
        sprintf(insert_sql,"insert into dev_pig_status(device_mac,ear_id,acqusition_time,update_time,weight)values('%s','%s',str_to_date('%s', '%%Y-%%m-%%d %%h:%%i:%%s'),now(),%f)",
            mac,ear_id,stime,f_weight);
        excuteSql(_db,insert_sql);
    }

    sprintf(log_str,"↑ 收到猪称重数据,网关:%s,采集时间:%s ,耳标:%s ,体重:%f",mac,stime,ear_id,f_weight);
                    
    
    
    log("%s \n",log_str);

    return 0;
}

//流量计数据
int do_DT_COP(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
{
    UNUSED(topicName);
    UNUSED(_db);

    //校验长度,变长只比较最小
    if(pubmsg->payloadlen<15)
        return 1;

    char update_sql[512] = {0};
    char insert_sql[512] = {0};
    char log_str[512] = {0};
    char mac[25] = {0};
    char temp[64] = {0};
    memset(temp,0x0,64);
    char2short water_temp1;                   //水温1
    float f_water_temp1 = 0.0;
    char2short water_temp2;                   //水温2
    float f_water_temp2 = 0.0;
    char2int flow_rate;                     //瞬时流量
    float f_flow_rate = 0.0;
    char2int forward_integer_part;          //正向累计整数部分
    char2short forward_decimal_part;          //正向累计小数部分
    float forward_accumulative_flow = 0.0;  //正向累计流量
    char2int reverse_integer_part;          //反向累计整数部分
    char2short reverse_decimal_part;          //反向累计小数部分
    float reverse_accumulative_flow = 0.0;  //反向累计流量
    char2short active_power_a;                //a相有功功率
    float f_active_power_a = 0.0;
    char2short active_power_b;                //b相有功功率
    float f_active_power_b = 0.0;
    char2short active_power_c;                //c相有功功率
    float f_active_power_c = 0.0;
    char2short total_active_power;            //总共有功功率
    float f_total_active_power = 0.0;

    //mac
    unsigned char* data = pubmsg->payload;
    Hex2Str((const char*)data,mac,12);
    

    //水温1
    memcpy(water_temp1.c_char,data+12,2);
    EndianSwap((unsigned char*)water_temp1.c_char,0,2);
    f_water_temp1 = ((float)water_temp1.i_short)/10.0;
    

    //水温2
    memcpy(water_temp2.c_char,data+14,2);
    EndianSwap((unsigned char*)water_temp2.c_char,0,2);
    f_water_temp2 = ((float)water_temp2.i_short)/10.0;

    //瞬时流量
    memcpy(flow_rate.c_char,data+16,4);
    EndianSwap((unsigned char*)flow_rate.c_char,0,4);
    f_flow_rate = ((float)flow_rate.i_int)/1000.0;

    //正向累计流量
    memcpy(forward_integer_part.c_char,data+20,4);
    EndianSwap((unsigned char*)forward_integer_part.c_char,0,4);
    memcpy(forward_decimal_part.c_char,data+24,2);
    EndianSwap((unsigned char*)forward_decimal_part.c_char,0,2);
    sprintf(temp,"%d.%u",forward_integer_part.i_int,forward_decimal_part.i_short);
    forward_accumulative_flow = atof(temp);
    memset(temp,0x0,64);

    //反向累计流量
    memcpy(reverse_integer_part.c_char,data+26,4);
    EndianSwap((unsigned char*)reverse_integer_part.c_char,0,4);
    memcpy(reverse_decimal_part.c_char,data+30,2);
    EndianSwap((unsigned char*)reverse_decimal_part.c_char,0,2);
    sprintf(temp,"%d.%d",reverse_integer_part.i_int,reverse_decimal_part.i_short);
    reverse_accumulative_flow = atof(temp);
    
    //a相有功功率
    memcpy(active_power_a.c_char,data+32,2);
    EndianSwap((unsigned char*)active_power_a.c_char,0,2);
    f_active_power_a = ((float)active_power_a.i_short)/1000.0;
    f_active_power_a = abs(f_active_power_a);

    //b相有功功率
    memcpy(active_power_b.c_char,data+34,2);
    EndianSwap((unsigned char*)active_power_b.c_char,0,2);
    f_active_power_b = ((float)active_power_b.i_short)/1000.0;
    f_active_power_b = abs(f_active_power_b);

    //c相有功功率
    memcpy(active_power_c.c_char,data+36,2);
    EndianSwap((unsigned char*)active_power_c.c_char,0,2);
    f_active_power_c = ((float)active_power_c.i_short)/1000.0;
    f_active_power_c = abs(f_active_power_c);

    //总共有功功率
    memcpy(total_active_power.c_char,data+38,2);
    EndianSwap((unsigned char*)total_active_power.c_char,0,2);
    f_total_active_power = ((float)total_active_power.i_short)/1000.0;
    f_total_active_power = abs(f_total_active_power);

    sprintf(update_sql,"update dev_status_flowmeter set current_online = 1,update_time = now(),water_temp1 = %f,water_temp2 = %f,flow_rate = %f,"
                        "forward_accumulative_flow = %f,reverse_accumulative_flow = %f,active_power_a = %f,active_power_b = %f,"
                       " active_power_c = %f,total_active_power = %f where gateway_mac = '%s'",f_water_temp1,f_water_temp2,f_flow_rate,
                        forward_accumulative_flow,reverse_accumulative_flow,f_active_power_a,f_active_power_b,f_active_power_c,f_total_active_power,mac);

    sprintf(insert_sql,"insert into dev_flowmeter_history_202203(gateway_mac,update_time,water_temp1,water_temp2,flow_rate,forward_accumulative_flow,"
                        "reverse_accumulative_flow,active_power_a,active_power_b,active_power_c,total_active_power)values('%s',now(),%f,%f,%f,%f,%f,%f,%f,%f,%f)",
                        mac,f_water_temp1,f_water_temp2,f_flow_rate,forward_accumulative_flow,reverse_accumulative_flow,f_active_power_a,f_active_power_b,f_active_power_c,f_total_active_power);

    sprintf(log_str,"↑ 收到电流计数据,网关:%s,水温1:%f ,水温2:%f ,瞬时流量:%f ,正向累计流量:%f ,反向累计流量:%f ,a相有功功率:%f ,b相有功功率:%f ,c相有功功率:%f ,总共有功功率:%f",
                    mac,f_water_temp1,f_water_temp2,f_flow_rate,forward_accumulative_flow,reverse_accumulative_flow,f_active_power_a,f_active_power_b,f_active_power_c,f_total_active_power);
                    
    excuteSql(_db,update_sql);
    excuteSql(_db,insert_sql);
    log("%s \n",log_str);

    return 0;
}

//设备传感器数据
int  do_DT_SDT(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
{
    UNUSED(topicName);
    UNUSED(_db);

    //校验长度,变长只比较最小
    if(pubmsg->payloadlen<15)
        return 1;

    char update_sql[512] = {0};
    char log_str[512] = {0};
    char mac[25] = {0};
    int count; //传感器数量
    int line_count = 0;//线温计数
    int step = 13;
    int i= 0;

    unsigned char* data = pubmsg->payload;
    Hex2Str((const char*)data,mac,12);
    count = data[12]; 
    if(count<1) return 1;

    sprintf(update_sql,"update dev_status set current_online = 1,");
    sprintf(log_str,"↑ 收到传感器数据,网关:%s",mac);
    for(i = 0;i<count;i++)
    {
        if(step == pubmsg->payloadlen)
            return 1;

        int subType = data[step]; //节点类型
        step += 1;

        if(subType == 1) //温湿度传感器
        {
            //20210408 温湿度非255才更新
            if(data[step] != 255 && data[step+1] != 255)
            {
                sprintf(update_sql,"%s current_temperature=%d,current_humidity=%d,",update_sql,(char)data[step],(char)data[step+1]);
                sprintf(log_str,"%s,温度:%d,湿度:%d",log_str,(char)data[step],(char)data[step+1]);
            }
            step += 2;
        }
        else if(subType == 2) //陀螺仪传感器
        {
            char2short x_offset,y_offset,z_offset;
            x_offset.c_char[0] = data[step+1];
            x_offset.c_char[1] = data[step];
            y_offset.c_char[0] = data[step+3];
            y_offset.c_char[1] = data[step+2];
            z_offset.c_char[0] = data[step+5];
            z_offset.c_char[1] = data[step+4];
            sprintf(update_sql,"%s x_angle_offset=%f,y_angle_offset=%f,z_angle_offset=%f,",
                    update_sql,((double)x_offset.i_short)/10.0,((double)y_offset.i_short)/10.0,((double)z_offset.i_short)/10.0);
            sprintf(log_str,"%s,x倾斜角:%.2f,y倾斜角:%.2f,z倾斜角:%.2f",
                    log_str,((double)x_offset.i_short)/10.0,((double)y_offset.i_short)/10.0,((double)z_offset.i_short)/10.0);
            step += 6;
        }
        else if(subType == 3) //环境亮度传感器
        {
            char2short environmentLuminance;
            environmentLuminance.c_char[0] = data[step+1];
            environmentLuminance.c_char[1] = data[step];
            sprintf(update_sql,"%s environment_luminance=%u,",update_sql,(unsigned short)environmentLuminance.i_short);
            sprintf(log_str,"%s,环境光亮度:%u",log_str,(unsigned short)environmentLuminance.i_short);
            step += 2;
        }
        else if(subType == 4) //线温传感器
        {
            line_count += 1;
            //20210408 线温非255才更新
            if(data[step] != 255)
            {
                sprintf(update_sql,"%s line_temperature%d=%d,",update_sql,line_count,(char)data[step]);
                sprintf(log_str,"%s,线温%d:%d",log_str,line_count,(char)data[step]);
            }
            step += 1;
        }
        else
        {
            log("↑ %s,类型:%d,无法识别节点类型! \n",mac,subType);
            return 1;
        }
        
    }

    if(strlen(update_sql)>25)
    {
        sprintf(update_sql+strlen(update_sql)-1," where device_mac='%s'",mac);
        excuteSql(_db,update_sql);
        //sendtoqueue(m_pque,update_sql);
        log("%s \n",log_str);
    }
    else
    {
        log("%s,无传感器数据或者数据无效!\n",log_str);
    }

    return 0;
} 


//设备电气数据
int  do_DT_EPD(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
{
    //debug("topicName=%s payloadlen=%d\n", topicName, pubmsg->payloadlen);
    UNUSED(topicName);

    //校验长度,变长只比较最小
    if(pubmsg->payloadlen<12)
        return 1;

    char update_sql[512] = {0};
    char update_sql2[512] = {0};
    char insert_sql[512] = {0};
    char head_str[100] = {0};
    char data_str[1024] = {0};
    char string_mac[30][13] = {};//保存子节点数组
    char mac[25] = {0};
    int count; //节点数量
    int step = 13;
    int i= 0;
    struct timeval tv;
    unsigned int time1; //开始时间
    unsigned int time2; //结束时间

    unsigned char* data = pubmsg->payload;
    Hex2Str((const char*)data,mac,12);

    count = data[12]; 
    if(count<1) return 1;

    gettimeofday(&tv,NULL);
    time1 = tv.tv_sec*1000*1000 + tv.tv_usec;

    sprintf(update_sql2,"update dev_status set current_online = 1,update_time = now() where device_mac = '%s'",mac);
    excuteSql(_db,update_sql2);

    for(i = 0;i<count;i++)
    {
        if(step == pubmsg->payloadlen)
            return 1;

        char subMac[30] = {0}; //节点mac
        Hex2Str((const char*)(data+step),subMac,6);
        strcpy(string_mac[i],subMac);
        step += 6; 

        int subType = data[step]; //节点类型
        step += 1;

        if(i == 0) 
        {
            sprintf(head_str,"↑ 收到电气数据,网关:%s",mac);
        }

        if(subType == 1) //总表
        {
            char2int energy;
            memcpy(energy.c_char,data+step,4);
            EndianSwap((unsigned char*)energy.c_char,0,4);
            step += 4;

            //如果子节点上传数据重复,跳过解析当前节点
            if(isexist_mac(subMac,string_mac,i) == 1) continue;

            //log("%d.总表:%s,电能:%d",i+1,subMac,energy.i_int);
            sprintf(data_str,"%s %d.总表:%s,电能:%d ;",data_str,i+1,subMac,energy.i_int);
            
            sprintf(update_sql,"update dev_status set current_online = 1,current_energy = %d,update_time = now() where device_mac = '%s'",
                    energy.i_int,subMac);
            excuteSql(_db,update_sql);
            //sendtoqueue(m_pque,update_sql);

            sprintf(insert_sql,"insert into %s(device_mac,gateway_mac,device_type,current_energy,update_time)values('%s','%s',1,%d,%d)",
                    GetCurrentNYTable(),subMac,mac,energy.i_int,timeGloble_g);
            excuteSql(_db,insert_sql);
            //sendtoqueue(m_pque,insert_sql);
                
        }
        else if(subType == 2) //智能空开节点
        {
            char2int energy;
            char2int power;
            char2int current;
            char2int voltage;
            char2int relay;
            char dMac[30] = {0};

            //电压
            memcpy(voltage.c_char,data+step,4);
            EndianSwap((unsigned char*)voltage.c_char,0,4);
            step += 4;

            //电流
            memcpy(current.c_char,data+step,4);
            EndianSwap((unsigned char*)current.c_char,0,4);
            step += 4;

            //功率
            memcpy(power.c_char,data+step,4);
            EndianSwap((unsigned char*)power.c_char,0,4);
            step += 4;

            //电能
            memcpy(energy.c_char,data+step,4);
            EndianSwap((unsigned char*)energy.c_char,0,4);
            step += 4;
            
            //继电器状态,20210318 从2个字节改为4个字节
            relay.i_int = 0;
            relay.c_char[3] = data[step];
            relay.c_char[2] = data[step+1];
            relay.c_char[1] = data[step+2];
            relay.c_char[0] = data[step+3];
            step += 4;

            //如果子节点上传数据重复,跳过解析当前节点
            if(isexist_mac(subMac,string_mac,i) == 1) continue;
            
            int relay_state = 0;
            //0x0100 手动合闸,0x0 继电器合闸;20210303新增0x0400 远程合闸;20210318新增4个类型 0x00010000 定时合闸,0x080000 过压恢复合闸,0x100000 欠压恢复合闸,0x200000上电合闸
            if(relay.i_int == 0x0100 || relay.i_int == 0x0000 || relay.i_int == 0x0400 || relay.i_int == 0x010000 
                || relay.i_int == 0x080000 || relay.i_int == 0x100000 || relay.i_int == 0x200000)
            {
                relay_state = 1;
            }
            else if (relay.i_int == 0x0000FFFF) //0x0000FFFF 此状态不更新继电器状态
            {
                relay_state = -1;
            }
            

            //debug("relay:%d,char[0]:%02x,char[1]:%02x,char[2]:%02x,char[3]:%02x \n",relay.i_int,relay.c_char[0],relay.c_char[1],relay.c_char[2],relay.c_char[3]);

            if(strcmp(subMac,"0F0000000001") == 0) //网关数据
            {
                strcpy(dMac,mac);
            }
            else
            {
                strcpy(dMac,subMac);
                //log("↑ 收到电气数据,网关:%s,智能空开:%s,电压:%d,电流:%d,功率:%d,电能:%d,继电器状态:%d,code:%02x%02x \n",mac,subMac,voltage.i_int,current.i_int,power.i_int,energy.i_int,relay_state,relay.c_char[1],relay.c_char[0]); 
                sprintf(data_str,"%s %d.智能空开:%s,电压:%d,电流:%d,功率:%d,电能:%d,继电器状态:%d,code:%02x%02x%02x%02x;",
                                data_str,i+1,subMac,voltage.i_int,current.i_int,power.i_int,energy.i_int,relay_state,
                                (unsigned char)relay.c_char[3],(unsigned char)relay.c_char[2],(unsigned char)relay.c_char[1],(unsigned char)relay.c_char[0]);
            }
            
            if(relay_state == -1)//0x0000FFFF 此状态不更新继电器状态
            {
                sprintf(update_sql,"update dev_status set current_online = 1, current_voltage = %d,current_electric = %d,current_power = %d,current_energy = %d,update_time=now(),electric_update_flag=1 where device_mac ='%s'",
                    abs(voltage.i_int),abs(current.i_int),abs(power.i_int),abs(energy.i_int),dMac);
            }
            else
            {
                sprintf(update_sql,"update dev_status set current_online = 1, current_voltage = %d,current_electric = %d,current_power = %d,current_energy = %d,current_on_off = %d,original_relay = %d,update_time=now(),electric_update_flag=1 where device_mac ='%s'",
                    abs(voltage.i_int),abs(current.i_int),abs(power.i_int),abs(energy.i_int),relay_state,abs(relay.i_int),dMac);
            }
            //excuteSql(_db,update_sql);
            //sendtoqueue(m_pque,update_sql);

            if(relay_state == -1)//0x0000FFFF 此状态不更新继电器状态
            {
                sprintf(insert_sql,"insert into %s(device_mac,gateway_mac,device_type,current_voltage,current_electric,current_power,current_energy,update_time)"
                                "values('%s','%s',2,%d,%d,%d,%d,%d)",GetCurrentNYTable(),dMac,mac,voltage.i_int,current.i_int,power.i_int,energy.i_int,timeGloble_g);
            }
            else
            {
                sprintf(insert_sql,"insert into %s(device_mac,gateway_mac,device_type,current_voltage,current_electric,current_power,current_energy,current_on_off,update_time)"
                                "values('%s','%s',2,%d,%d,%d,%d,%d,%d)",GetCurrentNYTable(),dMac,mac,voltage.i_int,current.i_int,power.i_int,energy.i_int,relay_state,timeGloble_g);
            }
            //excuteSql(_db,insert_sql);
            //sendtoqueue(m_pque,insert_sql);

            //如果数据异常,放弃写入数据库
            if(voltage.i_int < 26000 && current.i_int < 300000 && voltage.i_int>=0 && current.i_int>=0)
            {
                excuteSql(_db,update_sql);
                excuteSql(_db,insert_sql);
            }

            //继电器报警
            //RelayAlarm(mac,subMac,relay.i_int,_db);

            //如果是网关,跳过最后打印日志
            if(strcmp(subMac,"0F0000000001") == 0)
            {
                gettimeofday(&tv,NULL);
                time2 = tv.tv_sec*1000*1000 + tv.tv_usec;
                log("↑ 收到电气数据,网关:%s,电压:%d,电流:%d,功率:%d,电能:%d,继电器状态:%d,code:%02x%02x%02x%02x,cost:%d微秒 \n",
                    mac,voltage.i_int,current.i_int,power.i_int,energy.i_int,relay_state,
                    (unsigned char)relay.c_char[3],(unsigned char)relay.c_char[2],(unsigned char)relay.c_char[1],(unsigned char)relay.c_char[0],(time2-time1));
                return 0;
            }
        }
        else if(subType == 3) //调光节点
        {
            //组号
            int group = data[step];
            step += 1;
            //亮度
            int luminance = data[step];
            step += 1;
            //调光模式
            int dimming_mode = data[step];
            step += 1;
            //调光类型
            int dimming_type = data[step];
            step += 1;
            //电压
            int voltage = data[step];
            step += 1;
            //电流
            char2int current;
            memset(&current,0x0,4);
            current.c_char[0] = data[step+1];
            current.c_char[1] = data[step];
            debug("电流: data[step]:%02X,data[step+1]:%02X,C_CHAR:%02X %02X %02X %02X \n",data[step],data[step+1],current.c_char[0],current.c_char[1],current.c_char[2],current.c_char[3]);
            step += 2;

            //如果子节点上传数据重复,跳过解析当前节点
            if(isexist_mac(subMac,string_mac,i) == 1) continue;
            
            //log("↑ 收到电气数据,网关:%s,调光节点:%s,组号:%d,亮度:%d,调光模式:%d,调光类型:%d,电压:%d,电流:%d \n",mac,subMac,group,luminance,
                //dimming_mode,dimming_type,abs(voltage),abs(current.i_int));
            sprintf(data_str,"%s %d.调光节点:%s,组号:%d,亮度:%d,调光模式:%d,调光类型:%d,电压:%d,电流:%d ;",
                    data_str,i+1,subMac,group,luminance,dimming_mode,dimming_type,abs(voltage),abs(current.i_int));

            sprintf(update_sql,"update dev_status set current_online = 1, current_luminance = %d,device_group = %d,current_voltage = %d,current_electric = %d,"
                    "update_time=now() where device_mac ='%s'",luminance,group,abs(voltage),abs(current.i_int),subMac);
            excuteSql(_db,update_sql);
            //sendtoqueue(m_pque,update_sql);

            sprintf(insert_sql,"insert into %s(device_mac,gateway_mac,device_type,device_luminance,device_group,current_voltage,current_electric,update_time)"
                                "values('%s','%s',3,%d,%d,%d,%d,%d)",GetCurrentNYTable(),subMac,mac,luminance,group,voltage,current.i_int,timeGloble_g);
            excuteSql(_db,insert_sql);
            //sendtoqueue(m_pque,insert_sql);

        }
        else if(subType == 4) //烟感节点
        {
            //报警状态
            int alarm = 0;
            alarm = data[step];
            step += 1;

            //如果子节点上传数据重复,跳过解析当前节点
            if(isexist_mac(subMac,string_mac,i) == 1) continue;

            //log("↑ 收到电气数据,网关:%s,烟感节点:%s,报警状态:%d",mac,subMac,alarm);
            sprintf(data_str,"%s %d.烟感节点:%s,报警状态:%d ;",data_str,i+1,subMac,alarm);
        }
        else
        {
            log("↑ 收到电气数据,网关:%s,节点:%s,类型:%d,无法识别节点类型! \n",mac,subMac,subType);
            return 1;
        }

    }

    //log("%s,%s \n",head_str,data_str);
    gettimeofday(&tv,NULL);
    time2 = tv.tv_sec*1000*1000 + tv.tv_usec;
    log("%s,%scost:%d微秒\n",head_str,data_str,(time2-time1));

    return 0;
}


//SIM卡ICCID号
int  do_DT_SIM(char *topicName, MQTTAsync_message *pubmsg, MYSQL *_db)
{
    //debug("topicName=%s payloadlen=%d\n", topicName, pubmsg->payloadlen);

    //校验长度
    if(CheckMsgLength(topicName, pubmsg,32))
        return 1;

    char update_sql[512] = {0};
    char mac[25] = {0};
    char sim[21] = {0};

    unsigned char* data = pubmsg->payload;
    Hex2Str((const char*)data,mac,12);
    memcpy(sim,data+12,20);
    log("↑ 收到SIM卡ICCID数据,网关:%s,iccid:%s \n",mac,sim);

    
    sprintf(update_sql,"update dev_info_gateway set sim_iccid = '%s' where device_mac ='%s'",sim,mac);
    excuteSql(_db,update_sql);

    return 0;
}