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"


void master_task(uint8_t *string,uint16_t stringlength);
void slave_task();
int READ_MODBUS_DATA(DEVICE_PARAMS *current_device,char* buf);
///////////////////////////////////////////////////////////////////////////////////////////
char string[512];
void  findDifference(char* data1, char* data2, char* string) ;
int jsoncunt = 1;

////////////////////////////////////////////////////////////////////////////////////////////
uint16_t BufferSize;
uint8_t Buffer[256];
uint32_t rx_num = 0;

uint8_t PingMsg[] = "PING\0";
uint8_t PongMsg[] = "PONG\0";
tRadioDriver *Radio=NULL;
/*
*********************************************************************************************************
*	函 数 名: void data_task(void *pdata)
*	功能说明: 主要是data_task处理线程，优先级高。其运行逻辑是将nandflash中的数据解析出来轮询发送数据
*	形    参：无
*	返 回 值: 无
*********************************************************************************************************
*/

void data_task(void *pdata)
{
	OS_CPU_SR cpu_sr;
	pdata = pdata;
//	Radio = RadioDriverInit();
//	Radio->Init();
	mmodbus_init(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();
	int deviceIndex=0;
	DEVICE_PARAMS *current_device=get->device_params;
	
//	uint32_t baudrate = get->baudrate;
//	uint8_t databits = get->dataBits;
//	uint8_t stopbits = get->stopBit;
//	uint8_t parity = get->parity;
//	uint8_t flowcontrol = get->flowControl;
//	Config_485_Port(baudrate, databits, stopbits, parity, flowcontrol);
	 

	uint16_t bufferLength;
	OS_Q_DATA Qnum;
	StringInfo message;
	char *mqttRecv;
	uint8_t err;
	
	while (current_device!=NULL)
	{		
		char buf[1000];	// 接收modbus读取的数据	
		READ_MODBUS_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 - 10000)
				{
					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();
				}//10s进行一次
				
				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;
		}			
	}
}

	#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_MODBUS_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;
	
	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, 70);
			OSIntEnter();
			if(power)
			{
				slaveAddress = currentModbusParams->slaveAddress;
				number =currentModbusParams->registerAddress;
				data = power->valueint;
				mmodbus_writeHoldingRegister16i(slaveAddress, number, data);
			}
			if(temp)
			{
				currentModbusParams = currentModbusParams->nextParams;
				slaveAddress = currentModbusParams->slaveAddress;
				number =currentModbusParams->registerAddress;
				data = temp->valueint;
				mmodbus_writeHoldingRegister16i(slaveAddress, number, data);
			}
			if(mode)
			{
				currentModbusParams = currentModbusParams->nextParams;
				slaveAddress = currentModbusParams->slaveAddress;
				number =currentModbusParams->registerAddress;
				data = mode->valueint;
				mmodbus_writeHoldingRegister16i(slaveAddress, number, data);
			}
			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 数组，找到变化的部分
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);
}