#include "task.h" #include "cjson.h" #include "myFile.h" #include "gateway_message.h" #include "log.h" #include "malloc.h" #include "sx1276.h" #include "dlt645.h" #include "usart.h" #include "node_data_acquisition.h" #include "sys_mqtt.h" #include "sys_http.h" #include "node_message.h" #include "usart.h" #include "mmodbus.h" #include "sys_mqtt.h" #include "gateway_message.h" #include "MQTTClient.h" #include "cJSON.h" #include "time_count.h" #include "dlt645_1997_private.h" char string[512]; /* ********************************************************************************************************* * 函 数 名: void data_task(void *pdata) * 功能说明: 主要是data_task处理线程,优先级高。其运行逻辑是将nandflash中的数据解析出来轮询发送数据 * 形 参:无 * 返 回 值: 无 ********************************************************************************************************* */ void data_task(void *pdata) { OS_CPU_SR cpu_sr; pdata = pdata; dlt645_init(100); mmodbus_init(1); int jsonCunt = 1; char *device_config_json = mymalloc(SRAMEX, 9 * 1024); read_file("device.txt", device_config_json); addGatewayParams(device_config_json); myfree(SRAMEX, device_config_json); GATEWAY_PARAMS *get; get= get_gateway_config_params(); DEVICE_PARAMS *current_device=get->device_params; // Config_485_Port(get->baudrate, get->dataBits, get->stopBit, get->parity, get->flowControl); char *buf = mymalloc(SRAMEX, 9 * 1024); // 接收读取的数据 char *pubJsonStringCopy = mymalloc(SRAMEX, 9 * 1024); // 备份数据 memset(buf, 0, 9 * 1024); while (current_device!=NULL) { read_device_data(current_device, buf); //读取数据 send_mqtt(buf, jsonCunt,pubJsonStringCopy); //发送数据 jsonCunt = 0; memset(buf,0,strlen(buf)); current_device=get->device_params; OSTimeDly(1); } myfree(SRAMEX, buf); myfree(SRAMEX,pubJsonStringCopy); } /* ********************************************************************************************************* * 函 数 名: void mqtt_to_device() * 功能说明: 将接收到的数据发送至设备 * 形 参: * 返 回 值: ********************************************************************************************************* */ void mqtt_to_device(){ uint8_t err; StringInfo *message; message = (StringInfo*)OSMboxPend(mqtt_recvMseeageMbox, 1000, &err); if(message != NULL) //包含消息 { write_modbus_data(message->p); //写入数据 myfree(SRAMEX ,message->p);//释放内部数据 //OSTimeDly(1000); } } /* ********************************************************************************************************* * 函 数 名: int READ_MODBUS_DATA(DEVICE_PARAMS *device) * 功能说明: 读取当前节点上的modbus数据 * 形 参: DEVICE_PARAMS *device 当前设备 * 返 回 值: 1: 成功 0:失败 ********************************************************************************************************* */ int read_device_data(DEVICE_PARAMS *device, char* buf) { DEVICE_PARAMS *current_device=device; GATEWAY_READ_MODBUS_COMMAND *currentModbusParams = current_device->params->gateway_read_modbus_command; GATEWAY_READ_DLT645_COMMAND *currentDLT645Params = current_device->params->gateway_read_dlt645_command; while(current_device->params != NULL) { if (current_device->protocol == MODBUS_READ) { protocol_485=1; uint16_t data[currentModbusParams->registerByteNum /2]; // modbus寄存器长度 mmodbus_set16bitOrder(current_device->MDBbigLittleFormat); if (currentModbusParams->functionCode == 0x03 | currentModbusParams->functionCode == 0x01) { bool success = mmodbus_readHoldingRegisters16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, currentModbusParams->registerByteNum /2, data); if (success) { uint32_t value; if (currentModbusParams->registerByteNum == 4) { value = (uint32_t)data[0] | data[1]; } else if (currentModbusParams->registerByteNum == 2) { value = data[0]; } if (currentModbusParams->decimalPoint == 0) { currentModbusParams->value[0] = value; currentModbusParams->value[1] = value << 8; currentModbusParams->value[2] = value << 16; currentModbusParams->value[3] = value << 24; } else { float convertedValue = (float)value / pow(10, currentModbusParams->decimalPoint); memcpy(currentModbusParams->value, &convertedValue, 4); } sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":%d},", current_device->deviceID, currentModbusParams->keyword, value); } // else // { // printf("read modbus register fail\n"); // return 0; // } /* 每读完一个寄存器,进行message判断 */ mqtt_to_device(); currentModbusParams = currentModbusParams->nextParams; if (currentModbusParams == NULL) { current_device = current_device->nextDevice; currentModbusParams = current_device->params->gateway_read_modbus_command; if(current_device == NULL) { sprintf(buf + strlen(buf) - 1, "}"); return 1; } } } } else if (current_device->protocol == DLT645_2007 || current_device->protocol == DLT645_97) { protocol_485=2; uint8_t read_buf[10]; uint32_t dltValue; currentDLT645Params->rxLen = 0; memset(read_buf, 0, 10); memset(currentDLT645Params->data, 0, 10); dlt645_set_addr(&dlt645, currentDLT645Params->deviceID645); int8_t rs; if (current_device->protocol == DLT645_2007) { rs = dlt645_read_data(&dlt645, currentDLT645Params->Identification, read_buf, DLT645_2007); } else if (current_device->protocol == DLT645_1997) { rs = dlt645_read_data(&dlt645, currentDLT645Params->Identification, read_buf, DLT645_1997); } if (rs != -1) { if (rs <= 4) { memcpy(currentDLT645Params->data, read_buf, 4); currentDLT645Params->rxLen = 4; } else if (rs == 5) { memcpy(currentDLT645Params->data, read_buf, 5); currentDLT645Params->rxLen = 5; } else if (rs > 5) { memcpy(currentDLT645Params->data, read_buf, 9); currentDLT645Params->rxLen = 9; } dltValue = currentDLT645Params->data[0] << 24 | currentDLT645Params->data[1] << 16| currentDLT645Params->data[2] << 8 | currentDLT645Params->data[3]; sprintf(buf + strlen(buf), "{\"identifier\":\"%s\",\"deviceID645\":\"%02x%02x%02x%02x%02x%02x\",\"identifier645\":%d,\"value\":%X}", currentDLT645Params->keyword, currentDLT645Params->deviceID645[0], currentDLT645Params->deviceID645[1],currentDLT645Params->deviceID645[2], currentDLT645Params->deviceID645[3],currentDLT645Params->deviceID645[4], currentDLT645Params->deviceID645[5],currentDLT645Params->Identification,dltValue); } // else // { // currentDLT645Params->rxLen = 0; // printf("read DLT current data fail\n"); // } currentDLT645Params = currentDLT645Params->nextParams; if (currentDLT645Params == NULL) { current_device = current_device->nextDevice; currentDLT645Params = current_device->params->gateway_read_dlt645_command; if(current_device == NULL) { sprintf(buf + strlen(buf) - 1, "}"); return 1; } } } } return 1; } /* ********************************************************************************************************* * 函 数 名:void WRITE_MODBUS_DATA(char* cJSONstring) * 功能说明: 接收mqtt数据并写入modbus寄存器 * 形 参:char* cJSONstring mqtt接收到的数据 * 返 回 值: 无 ********************************************************************************************************* */ void write_modbus_data(char* cJSONstring) { GATEWAY_PARAMS* get; get = get_gateway_config_params(); DEVICE_PARAMS* current_device = get->device_params; /* 利用cJSOn_Parse解析数据,获取各类型数据 */ cJSON *root = cJSON_Parse(cJSONstring); const char *deviceId = cJSON_GetStringValue(cJSON_GetObjectItem(root, "deviceId")); const cJSON *power = cJSON_GetObjectItemCaseSensitive(root, "power"); const cJSON *temp = cJSON_GetObjectItemCaseSensitive(root, "temp"); const cJSON *mode = cJSON_GetObjectItemCaseSensitive(root, "mode"); const cJSON *fan = cJSON_GetObjectItemCaseSensitive(root, "fan"); while(current_device) { char* device_ID = (char*)current_device->deviceID; GATEWAY_WRITE_MODBUS_COMMAND *currentModbusParams = current_device->params->gateway_write_modbus_command; if(!strcmp(device_ID,deviceId)) //匹配ID { OSTimeDly(100); /* 写入寄存器操作 */ if(power) { mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, power->valueint); } OSTimeDly(100); if(temp) { currentModbusParams = currentModbusParams->nextParams; mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, temp->valueint); } OSTimeDly(100); if(mode) { currentModbusParams = currentModbusParams->nextParams; mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, mode->valueint); } OSTimeDly(100); if(fan) { currentModbusParams = currentModbusParams->nextParams; mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, fan->valueint); } } current_device = current_device->nextDevice; } cJSON_Delete(root); } /* ********************************************************************************************************* * 函 数 名: void find_difference(char* buf, char* pubJsonStringCopy, char* string) * 功能说明: 比较出参数1和参数2的不同处 * 形 参: 参数1:新数据 参数2:旧数据 参数3:输出参数 * 返 回 值: 无 ********************************************************************************************************* */ void find_difference(char* buf, char* pubJsonStringCopy, char* string) { const char* delimiter = "{}"; char* saveptr1; char* saveptr2; char* data1 = malloc(strlen(buf) + 1); char* data2 = malloc(strlen(pubJsonStringCopy) + 1); ; memcpy(data1, buf, strlen(buf)); memcpy(data2, pubJsonStringCopy, strlen(pubJsonStringCopy)); // 利用strtok_r函数分割字符串,并逐一比较 char* token1 = strtok_r((char*)data1, delimiter, &saveptr1); char* token2 = strtok_r((char*)data2, delimiter, &saveptr2); while (token1 != NULL && token2 != NULL) { if (strcmp(token1, token2) != 0) { memcpy(string + strlen(string), token1, strlen(token1)); } token1 = strtok_r(NULL, delimiter, &saveptr1); token2 = strtok_r(NULL, delimiter, &saveptr2); } // // 如果有剩余字符串未比较,则打印剩余字符串 // while (token1 != NULL) { // sprintf(string + strlen(string),"%s,", token1); // token1 = strtok_r(NULL, delimiter, &saveptr1); // } // while (token2 != NULL) { // //sprintf(string + strlen(string),"{%s},", token2); // token2 = strtok_r(NULL, delimiter, &saveptr2); // } free(data1); free(data2); } /* ********************************************************************************************************* * 函 数 名: void send_mqtt(char*buf, int jsonCunt) * 功能说明: 将数据发送到mqtt * 形 参: 参数1:读取数据 参数2:第一次发送标志 * 返 回 值: 无 ********************************************************************************************************* */ void send_mqtt(char*buf, int jsonCunt, char*pubJsonStringCopy){ GATEWAY_PARAMS *get; get= get_gateway_config_params(); if(get->device_params->protocol == MODBUS_READ) send_modbus_data(get, buf, jsonCunt,pubJsonStringCopy);// modbus数据通道 else if(get->device_params->protocol == DLT645_2007 || get->device_params->protocol == DLT645_1997) send_dlt645_data(get, buf, jsonCunt,pubJsonStringCopy);// DLT645数据通道 } /* ********************************************************************************************************* * 函 数 名: void send_modbus_data(GATEWAY_PARAMS *get, char*buf, int jsonCunt, char*pubJsonStringCopy) * 功能说明: 发送modubs数据需求的通道 * 形 参: 参数1:网关信息 参数2: 读取的信息 参数3: 第一次发送标志 参数4: 备份信息 * 返 回 值: 无 ********************************************************************************************************* */ void send_modbus_data(GATEWAY_PARAMS *get, char*buf, int jsonCunt, char*pubJsonStringCopy){ time1 = GetCurrentTime();// 获取当前时间 if(jsonCunt || time2 <= time1 - (20 * 1000)) // 20s进行一次 { memset(pubJsonStringCopy,0, strlen(pubJsonStringCopy)); memcpy(pubJsonStringCopy + strlen(pubJsonStringCopy), buf, strlen(buf));//备份上一次的数据 sprintf(pubJsonString,"{\"DEVICEID\":\"%s\",\"data\":[%s]",get->deviceId, buf); // 组成要发送的json语句 int msg = MBOX_USER_PUBLISHQOS0; if(mqtt_connectFlag==1) OSMboxPost(mqtt_sendMseeageMbox, &msg); time2 = GetCurrentTime(); // 获取当前时间 } else { if(strcmp(buf,pubJsonStringCopy)) // 比较两次数据是否不同 { memset(string, 0 , strlen(string)); find_difference(buf, pubJsonStringCopy, string);// 比较两次采集的数据不同之处 memset(pubJsonString,0, strlen(pubJsonString)); sprintf(pubJsonString,"{\"DEVICEID\":\"%s\",\"data\":[{%s}]}",get->deviceId, string);// 组成要发送的json语句 memset(pubJsonStringCopy,0, strlen(pubJsonStringCopy)); sprintf(pubJsonStringCopy, buf, strlen(buf));// 备份当前数据 int msg = MBOX_USER_PUBLISHQOS0; if(mqtt_connectFlag==1) OSMboxPost(mqtt_sendMseeageMbox, &msg); } } } /* ********************************************************************************************************* * 函 数 名: void send_dlt645_data(GATEWAY_PARAMS *get, char*buf, int jsonCunt, char*pubJsonStringCopy) * 功能说明: 发送modubs数据需求的通道 * 形 参: 参数1:网关信息 参数2: 读取的信息 参数3: 第一次发送标志 参数4: 备份信息 * 返 回 值: 无 ********************************************************************************************************* */ void send_dlt645_data(GATEWAY_PARAMS *get, char*buf, int jsonCunt, char*pubJsonStringCopy){ memset(pubJsonStringCopy,0, strlen(pubJsonStringCopy)); memcpy(pubJsonStringCopy + strlen(pubJsonStringCopy), buf, strlen(buf));//备份上一次的数据 sprintf(pubJsonString,"{\"DEVICEID\":\"%s\",\"data\":[%s]",get->deviceId, buf); // 组成要发送的json语句 int msg = MBOX_USER_PUBLISHQOS0; if(mqtt_connectFlag==1) OSMboxPost(mqtt_sendMseeageMbox, &msg); }