#include "task.h" #include "cjson.h" #include "myFile.h" #include "gateway_message.h" #include "log.h" #include "malloc.h" #include "sx1276.h" #include "protocol.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" int READ_DATA(DEVICE_PARAMS *current_device,char* buf); void WRITE_MODBUS_DATA(char* cJSONstring); /////////////////////////////////////////////////////////////////////////////////////////// char string[512]; void findDifference(char* data1, char* data2, char* string) ; int jsoncunt = 1; int commd = 1; //////////////////////////////////////////////////////////////////////////////////////////// /* ********************************************************************************************************* * �� �� ��: void data_task(void *pdata) * ����˵��: ��Ҫ��data_task�����̣߳����ȼ��ߡ��������߼��ǽ�nandflash�е����ݽ���������ѯ�������� * �� �Σ��� * �� �� ֵ: �� ********************************************************************************************************* */ void data_task(void *pdata) { OS_CPU_SR cpu_sr; pdata = pdata; mmodbus_init(1); char buf[1000]; // ����modbus��ȡ������ 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(); int deviceIndex=0; DEVICE_PARAMS *current_device=get->device_params; // Config_485_Port(get->baudrate, get->dataBits, get->stopBit, get->parity, get->flowControl); while (current_device!=NULL) { uint8_t err; StringInfo *message=NULL; //���ն�����Ϣ�ṹ�� message=(StringInfo *)OSQPend(JsonQ,1000,&err); if(commd) { READ_DATA(current_device, buf); if(current_device->nextDevice!=NULL) { current_device=current_device->nextDevice; } else { sprintf(buf + strlen(buf) - 1, "}"); time1 = OSTimeGet(); if( jsoncunt || time2 <= time1 - 20000) { memset(pubJsonStringCopy,0, strlen(pubJsonStringCopy)); memcpy(pubJsonStringCopy + strlen(pubJsonStringCopy), buf, strlen(buf));//������һ�ε����� sprintf(pubJsonString,"{\"DEVICEID\":\"%s\",\"data\":[%s]",get->deviceId, buf); int msg = MBOX_USER_PUBLISHQOS2; if(mqtt_connectFlag==1) OSMboxPost(mqtt_sendMseeageMbox, &msg); current_device=get->device_params; jsoncunt = 0; time2 = OSTimeGet(); }//20s����һ�� else { if(strcmp(buf,pubJsonStringCopy)) { memset(string, 0 , strlen(string)); findDifference(buf, pubJsonStringCopy, string); memset(pubJsonString,0, strlen(pubJsonString)); sprintf(pubJsonString,"{\"deviceId\":\"%s\",\"data\":[{%s}]}",get->deviceId, string); memset(pubJsonStringCopy,0, strlen(pubJsonStringCopy)); sprintf(pubJsonStringCopy, buf, strlen(buf)); int msg = MBOX_USER_PUBLISHQOS2; if(mqtt_connectFlag==1) OSMboxPost(mqtt_sendMseeageMbox, &msg); } } memset(buf,0,strlen(buf)); current_device=get->device_params; } } else { if(message != NULL) { WRITE_MODBUS_DATA(message->p); } printf("sx1278_task_recv : %s",message->p); myfree(SRAMIN ,message->p); commd = 1; } OSTimeDly(1000); } } #if 0 //���Դ��벻����ת��ֱ�ӽ�����Ӧ�Ľ��� SlaveProtocolAnalysis(string,bufferLength); data_acquisition(); uint8_t node_string[256]; uint16_t node_string_Length; nodeSendReaddValue(node_string,&node_string_Length); GatewayProtocolAnalysis(node_string,node_string_Length); #endif /* ********************************************************************************************************* * �� �� ��: int READ_MODBUS_DATA(DEVICE_PARAMS *device) * ����˵��: ��ȡ��ǰ�ڵ��ϵ�modbus���� * �� �Σ�DEVICE_PARAMS *device ��ǰ�豸 * �� �� ֵ: 1 ���ɹ� 0��ʧ�� ********************************************************************************************************* */ int READ_DATA(DEVICE_PARAMS *device, char* buf) { DEVICE_PARAMS *current_device=device; GATEWAY_PARAMS* get; get = get_gateway_config_params(); 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); currentModbusParams = currentModbusParams->nextParams; if (currentModbusParams == NULL) { sprintf(buf + strlen(buf) - 1, ""); return 0; } } } } } return 1; } /* ********************************************************************************************************* * �� �� ��:void WRITE_MODBUS_DATA(char* cJSONstring) * ����˵��: ����mqtt���ݲ�д��modbus�Ĵ��� * �� �Σ�char* cJSONstring mqtt���յ������� * �� �� ֵ: �� ********************************************************************************************************* */ void WRITE_MODBUS_DATA(char* cJSONstring) { uint16_t data; uint16_t number; uint16_t slaveAddress; GATEWAY_PARAMS* get; get = get_gateway_config_params(); DEVICE_PARAMS* current_device = get->device_params; if(cJSONstring[0] != '\0'){ 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)) { OSTimeDlyHMSM(0, 0, 0, 100); //OSIntEnter(); if(power) { slaveAddress = currentModbusParams->slaveAddress; number =currentModbusParams->registerAddress; data = power->valueint; mmodbus_writeHoldingRegister16i(slaveAddress, number, data); } OSTimeDlyHMSM(0, 0, 0, 100); if(temp) { currentModbusParams = currentModbusParams->nextParams; slaveAddress = currentModbusParams->slaveAddress; number =currentModbusParams->registerAddress; data = temp->valueint; mmodbus_writeHoldingRegister16i(slaveAddress, number, data); } OSTimeDlyHMSM(0, 0, 0, 100); if(mode) { currentModbusParams = currentModbusParams->nextParams; slaveAddress = currentModbusParams->slaveAddress; number =currentModbusParams->registerAddress; data = mode->valueint; mmodbus_writeHoldingRegister16i(slaveAddress, number, data); } OSTimeDlyHMSM(0, 0, 0, 100); if(fan) { currentModbusParams = currentModbusParams->nextParams; slaveAddress = currentModbusParams->slaveAddress; number =currentModbusParams->registerAddress; data = fan->valueint; mmodbus_writeHoldingRegister16i(slaveAddress, number, data); } //OSIntExit(); } current_device = current_device->nextDevice; } cJSON_Delete(root); } } // �Ƚ����� JSONItem ���飬�ҵ��仯�IJ��� void findDifference(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); }