#include "data_task.h" #include "usart.h" #include "sys_mqtt.h" #include "sys_http.h" #include "mmodbus.h" #include "dlt645.h" #include "dlt645_1997_private.h" #include "dlt645_port.h" #include "gd32_flash.h" #include "protocol.h" #include "sys_http.h" #include "jsmn.h" #include "timer.h" uint8_t time_count = 0, jsonCunt = 0, count = 0; uint8_t protocol_485; int time1,time2; int ID = 0; void data_task(void *pdata) { dlt645_init(1); mmodbus_init(1); while(!load_http_config) { vTaskDelay(100); } portENTER_CRITICAL(); char *device_config_json=pvPortMalloc( 10 *1024 ); read_data_from_flash(device_config_json); addGatewayParams(device_config_json); vPortFree(device_config_json); device_config_json = NULL; portEXIT_CRITICAL(); GATEWAY_PARAMS *get; get= get_gateway_config_params(); DEVICE_PARAMS *current_device=get->device_params; char *buf = pvPortMalloc(1024); // 接收读取的数据 char *string = pvPortMalloc(1024); // 接收读取的数据 memset(buf,0,strlen(buf)); memset(string,0,strlen(string)); while (current_device!=NULL) { time1 = GetCurrentTime(); if(mqtt_connectFlag) { if(jsonCunt && time2 <= time1 - ( 10 * 1000))// 60s进行一次全数据发送 { read_device_data(current_device, buf, string); send_mqtt(buf); memset(buf,0,strlen(buf)); time2 = GetCurrentTime(); current_device=get->device_params; } else { read_device_data(current_device, buf, string); if(count > 0)// count检测是否含有数据 { send_mqtt(string); memset(string,0,strlen(string)); current_device=get->device_params; count = 0; if(!time_count) { jsonCunt = 1; time_count = 1; time2 = GetCurrentTime(); } } } memset(buf,0,strlen(buf)); } vTaskDelay(100); } vPortFree(buf); buf = NULL; } /* ********************************************************************************************************* * 函 数 名: int READ_MODBUS_DATA(DEVICE_PARAMS *device) * 功能说明: 读取当前节点上的modbus数据 * 形 参: DEVICE_PARAMS *device 当前设备 * 返 回 值: 1: 成功 0:失败 ********************************************************************************************************* */ int read_device_data(DEVICE_PARAMS *device, char* buf, char* string) { 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; uint8_t state; uint16_t data[currentModbusParams->registerByteNum /2]; // modbus寄存器长度 uint8_t data1[currentModbusParams->registerByteNum /2]; mmodbus_set16bitOrder(current_device->MDBbigLittleFormat); // 读水阀状态 if(currentModbusParams->functionCode == 0x01) { bool success = mmodbus_readCoil(currentModbusParams->slaveAddress, currentModbusParams->registerByteNum /2, data1); if(success) { uint8_t value; value = data1[0]; if(value == 0) { sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":close},", current_device->deviceID, currentModbusParams->keyword); }else{ sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":open},", current_device->deviceID, currentModbusParams->keyword); } } 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; } } } // 读单个寄存器 if (currentModbusParams->functionCode == 0x03) { // bool success = mmodbus_readHoldingRegisters16i(0x17,0x00,0x02,data); 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((value - currentModbusParams->value) != 0) { count++; sprintf(string + strlen(string), "{\"deviceId\":\"%s\",\"%s\":%d},", current_device->deviceID, currentModbusParams->keyword, value); } else { sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":%d},", current_device->deviceID, currentModbusParams->keyword, value); } if (currentModbusParams->decimalPoint == 0) { currentModbusParams->value = value; } else { // float convertedValue = (float)value / pow(10, currentModbusParams->decimalPoint); // currentModbusParams->value=convertedValue; } } 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, ""); sprintf(string + strlen(string) - 1, ""); return 1; } } } // 开关水阀 if(currentModbusParams->functionCode == 0x05) { bool success = mmodbus_writeCoil(currentModbusParams->slaveAddress, currentModbusParams->registerByteNum /2, state); if(success) { sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\": success},", current_device->deviceID, currentModbusParams->keyword); } else{ sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\": fail},", current_device->deviceID, currentModbusParams->keyword); } 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; } } } // 写单个寄存器 if(currentModbusParams->functionCode == 0x06) { bool success = mmodbus_writeHoldingRegisters16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, currentModbusParams->registerByteNum /2, data); if(success) { sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":write success},", current_device->deviceID, currentModbusParams->keyword); } 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); count++; } 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 send_mqtt(char*buf, int jsonCunt) * 功能说明: 将数据发送到mqtt * 形 参: 参数1:读取数据 参数2:第一次发送标志 * 返 回 值: 无 ********************************************************************************************************* */ void send_mqtt(char*buf){ GATEWAY_PARAMS *get; get= get_gateway_config_params(); // char *pubJsonString = pvPortMalloc(strlen(buf)); char pubJsonString[500]; sprintf(pubJsonString,"ID: %d {\"DEVICEID\":\"%s\",\"data\":[%s]}",ID++, get->deviceId, buf); // 组成要发送的json语句 mqtt_qos_t qos = QOS0; uint16_t pub_length = strlen(pubJsonString); mqtt_publish_data(pubJsonString, qos, pub_length, (char*)&get->messageTopic); // vPortFree(pubJsonString); // pubJsonString = NULL; } /* ********************************************************************************************************* * 函 数 名:void WRITE_MODBUS_DATA(char* cJSONstring) * 功能说明: 接收mqtt数据并写入modbus寄存器 * 形 参:char* cJSONstring mqtt接收到的数据 * 返 回 值: 无 ********************************************************************************************************* */ void write_modbus_data(char* JSON_STRING) { JSON_CMD jsonMsg; GATEWAY_PARAMS* get; get = get_gateway_config_params(); DEVICE_PARAMS* current_device = get->device_params; parseStringField(JSON_STRING, "\"deviceId\":\"", jsonMsg.deviceId); jsonMsg.function = parseIntField(JSON_STRING, "\"function\":"); jsonMsg.cmd = parseIntField(JSON_STRING, "\"cmd\":"); jsonMsg.power = parseIntField(JSON_STRING, "\"power\":"); jsonMsg.temp = parseIntField(JSON_STRING, "\"temp\":"); jsonMsg.mode = parseIntField(JSON_STRING, "\"mode\":"); jsonMsg.fan = parseIntField(JSON_STRING, "\"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,jsonMsg.deviceId)) //匹配ID { delay_ms(200); if(jsonMsg.function == 5) // 开关阀门 { mmodbus_writeCoil(jsonMsg.slaveAddress,jsonMsg.registerAddress,jsonMsg.cmd); } if(jsonMsg.function == 6) { /* 写入寄存器操作 */ if(jsonMsg.power) { mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, jsonMsg.power); } delay_ms(100); if(jsonMsg.temp) { currentModbusParams = currentModbusParams->nextParams; mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, jsonMsg.temp); } delay_ms(100); if(jsonMsg.mode) { currentModbusParams = currentModbusParams->nextParams; mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, jsonMsg.mode); } delay_ms(100); if(jsonMsg.fan) { currentModbusParams = currentModbusParams->nextParams; mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, currentModbusParams->registerAddress, jsonMsg.fan); } delay_ms(100); } } current_device = current_device->nextDevice; } }