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 "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);	// ���ն�ȡ������
		memset(buf, 0, 9 * 1024);	
		while (current_device!=NULL)
		{						
				read_device_data(current_device, buf);	//��ȡ����	
				send_mqtt(buf, jsonCunt);								//��������
				jsonCunt = 0;		
				memset(buf,0,strlen(buf));	
				current_device=get->device_params;		
				OSTimeDly(1);
		}
		myfree(SRAMEX, buf);
}	
/*
*********************************************************************************************************
*	�� �� ��: 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];
						
						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;
								}
								sprintf(buf + strlen(buf), "{\"identifier\":\"%s\",\"deviceID645\":\"%02x%02x%02x%02x%02x%02x\",\"identifier645\":%d}",
																				currentDLT645Params->keyword, currentDLT645Params->deviceID645[0],
																				currentDLT645Params->deviceID645[1],currentDLT645Params->deviceID645[2],
																				currentDLT645Params->deviceID645[3],currentDLT645Params->deviceID645[4],
																				currentDLT645Params->deviceID645[5],currentDLT645Params->Identification);
								
			
						}
//						else
//						{
//								currentDLT645Params->rxLen = 0;
//								printf("read DLT current data fail\n");
//						}
								/* ÿ����һ���Ĵ���������message�ж� */
								mqtt_to_device();
					
								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)
{
		uint16_t data;
		uint16_t number;
		uint16_t slaveAddress;
		
		GATEWAY_PARAMS* get;
		get = get_gateway_config_params();
		DEVICE_PARAMS* current_device = get->device_params;
		
		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))
				{
						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�IJ�ͬ��
*	��    ��: ����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){
		GATEWAY_PARAMS *get;
		get= get_gateway_config_params();
		time1 = GetCurrentTime();
		
		if(jsonCunt || time2  <= time1 - (3 * 1000)) // 20s����һ��
		{
				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_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);
						memset(pubJsonStringCopy,0, strlen(pubJsonStringCopy));
						sprintf(pubJsonStringCopy, buf, strlen(buf));
					
						int msg = MBOX_USER_PUBLISHQOS0;	
						if(mqtt_connectFlag==1) OSMboxPost(mqtt_sendMseeageMbox, &msg);	
				}
		}	
}