dtu_lan_gateway/app/System/source/task.c

#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);
}