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);	// 接收读取的数据
		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);
}