dtu_simple/app/System/source/data_task.c

#include "data_task.h"
#include "usart.h"
#include "sys_mqtt.h"
#include "sys_http.h"
#include "mmodbus.h"
#include "dlt645_port.h"
#include "gd32_flash.h"
#include "protocol.h"
#include "timer.h"
#include "led.h"
#include "tcp_server.h"
#include "log.h"

void protocolsModeFunc(GATEWAY_PARAMS* current_device, char* buf, char* string);
void transparentModeFunc(DEVICE_PARAMS* current_device);

int ID = 1;
uint8_t count = 0;
uint8_t startFlag = 0;// 读取数据初始标志位
int time1,time2;
void data_task(void *pdata)
{			
		dlt645_init(1);	// 若读不到数据，则延时 参数 秒
		mmodbus_init(10);// 若读不到数据，则延时 参数 秒
		GATEWAY_PARAMS *get;
		// 如果flash有config数据，则解析，进行下一步；若flash无config数据，则等待上位机发送数据
		do{
				char *device_config_json=pvPortMalloc( 10 *1024 );
				memset(device_config_json,0,strlen(device_config_json));
				portENTER_CRITICAL();
				read_data_from_flash(device_config_json);		
				addGatewayParams(device_config_json);
				vPortFree(device_config_json);
				portEXIT_CRITICAL();
				get= get_gateway_config_params();
				delay_ms(100);
		}while(get->device_params == NULL);
		LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"device params not empty");
		DEVICE_PARAMS *current_device=get->device_params;
		
		char *buf = pvPortMalloc(1024);			// 接收读取的全部数据
		char *string = pvPortMalloc(1024);	// 接收读取的不同数据
		if(buf == NULL || string == NULL)
		{
			LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"buf or string malloc fail");
		}
		memset(buf,0,strlen(buf));
		memset(string,0,strlen(string));
		while (current_device!=NULL)
		{				
			if(ProtocolsModeFlag)
			{		
				protocolsModeFunc(get, buf, string);	
			}
			else if(TransparentModeFlag)
			{
				transparentModeFunc(current_device);
			}
			current_device=get->device_params;	
			vTaskDelay(1000);
		}	
		vPortFree(buf);
		vPortFree(string);
		LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"data_task return");
}

/*
*********************************************************************************************************
*	函 数 名: int compareArrays(uint8_t arr1[], uint8_t arr2[], int size)
*	功能说明: 比较两个数组是否相同
*	形    参: arr1[] 数组1，arr2[] 数组2，size 比较数组的大小
*	返 回 值: 1: 相同 0:不相同
*********************************************************************************************************
*/
int compareArrays(uint8_t arr1[], uint8_t arr2[], int size) {
    for (int i = 0; i < size; ++i) {
        if (arr1[i] != arr2[i]) {
            return 1; // 两个数组不相同，返回0
        }
    }   
    return 0; // 两个数组相同，返回1
}
/*
*********************************************************************************************************
*	函 数 名: int READ_MODBUS_DATA(DEVICE_PARAMS *device)
*	功能说明: 读取当前节点上的modbus数据
*	形    参: DEVICE_PARAMS *device  当前设备
*	返 回 值: 1: 成功 0:失败
*********************************************************************************************************
*/
int read_device_data(DEVICE_PARAMS *device, char* buf, char* string)
{
		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)
		{	
				gd_eval_led_toggle(LED_485TX);
				if (current_device->protocol == MODBUS_READ)
				{
						uint8_t cmd;	//开关水阀命令
						uint8_t state;	// 水阀状态
						uint16_t data[currentModbusParams->registerByteNum /2]; // modbus寄存器长度
						
						mmodbus_set16bitOrder(current_device->MDBbigLittleFormat);
						// 读水阀状态
						if(currentModbusParams->functionCode == 0x01)
						{
							bool success = mmodbus_readCoil(currentModbusParams->slaveAddress,
																		 currentModbusParams->registerByteNum /2,
																		 &state);
							if(success)
							{
								if(state == 0)
								{
									sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":close},", 
																									current_device->deviceID, currentModbusParams->keyword);
								}else{
									sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":open},", 
																									current_device->deviceID, currentModbusParams->keyword);
								}
							}
							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;
										}
								}		
						}
						// 读单个寄存器
						if (currentModbusParams->functionCode == 0x03)
						{
//  							bool success = mmodbus_readHoldingRegisters16i(0x17,0x00,0x02,data);
								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((value - currentModbusParams->value) != 0)
										{
												count++;
												sprintf(string + strlen(string), "{\"deviceId\":\"%s\",\"%s\":%d},", 
																									current_device->deviceID, currentModbusParams->keyword, value);
										}
										else
										{									
												sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":%d},", 
																									current_device->deviceID, currentModbusParams->keyword, value);
										}
										if (currentModbusParams->decimalPoint == 0)
										{
												currentModbusParams->value = value;

										}
										else
										{ 	
												currentModbusParams->value = value / my_pow(10,currentModbusParams->decimalPoint);
										}
																	
								}					
								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, ""); 
												sprintf(string + strlen(string) - 1, ""); 
												return 1;
										}
								}												
						}
						// 开关水阀
						if(currentModbusParams->functionCode == 0x05)
						{
							bool success = mmodbus_writeCoil(currentModbusParams->slaveAddress,
																							 currentModbusParams->registerByteNum /2,
																							 cmd);
							if(success)
							{
								
									sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\": success},", 
																									current_device->deviceID, currentModbusParams->keyword);
							}
							else{
									sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\": fail},", 
																									current_device->deviceID, currentModbusParams->keyword);
							}
							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;
									}
							}
						}
						// 写单个寄存器
						if(currentModbusParams->functionCode == 0x06)
						{
							bool success = mmodbus_writeHoldingRegisters16i(currentModbusParams->slaveAddress,
																		 currentModbusParams->registerAddress,
																		 currentModbusParams->registerByteNum /2,
																		 data);
							if(success)
							{
								sprintf(buf + strlen(buf), "{\"deviceId\":\"%s\",\"%s\":write success},", 
																									current_device->deviceID, currentModbusParams->keyword);
							}
							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)
				{
						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(compareArrays(read_buf,currentDLT645Params->data,10))// 不相同1，相同0
								{
										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(string + strlen(string), "{\"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);
										count++;
								}
								else
								{
										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);
								}
						}
		
								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 send_mqtt(char*buf, int jsonCunt)
*	功能说明: 将数据发送到mqtt
*	形    参: 参数1:读取数据 参数2:第一次发送标志
*	返 回 值: 无
*********************************************************************************************************
*/
void send_mqtt(char*buf){
		LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"send to mqtt");
		GATEWAY_PARAMS *get;
		get= get_gateway_config_params();
		char *pubJsonString = pvPortMalloc(700);
	  sprintf(pubJsonString,"ID: %d {\"DEVICEID\":\"%s\",\"data\":[%s]}",ID, get->deviceId, buf);	// 组成要发送的json语句	
		mqtt_publish_data(pubJsonString, QOS0, strlen(pubJsonString), (char*)&get->messageTopic);	
		vPortFree(pubJsonString);
}

/*
*********************************************************************************************************
*	函 数 名:void WRITE_MODBUS_DATA(char* cJSONstring)
*	功能说明: 接收mqtt数据并写入modbus寄存器
*	形    参：char* cJSONstring mqtt接收到的数据
*	返 回 值: 无
*********************************************************************************************************
*/
void write_modbus_data(char* JSON_STRING)
{	
		JSON_CMD jsonMsg;
		GATEWAY_PARAMS* get;
		get = get_gateway_config_params();
		DEVICE_PARAMS* current_device = get->device_params;
	
		parseStringField(JSON_STRING, "\"deviceId\":\"", jsonMsg.deviceId);
    jsonMsg.function = parseIntField(JSON_STRING, "\"function\":");
		jsonMsg.cmd = parseIntField(JSON_STRING, "\"cmd\":");
		jsonMsg.power = parseIntField(JSON_STRING, "\"power\":");
		jsonMsg.temp = parseIntField(JSON_STRING, "\"temp\":");
		jsonMsg.mode = parseIntField(JSON_STRING, "\"mode\":");
		jsonMsg.fan = parseIntField(JSON_STRING, "\"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,jsonMsg.deviceId)) //匹配ID
				{
					portENTER_CRITICAL();
					delay_ms(50);
					if(jsonMsg.function == 5)
					// 开关阀门
					{			
						mmodbus_writeCoil(jsonMsg.slaveAddress,jsonMsg.registerAddress,jsonMsg.cmd);
					}
					if(jsonMsg.function == 6)
					{
					/* 写入寄存器操作 */
						if(jsonMsg.power)
						{
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, jsonMsg.power);
						}
						delay_ms(10);
						if(jsonMsg.temp)
						{
								currentModbusParams = currentModbusParams->nextParams;							
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, jsonMsg.temp);	
						}
						delay_ms(10);
						if(jsonMsg.mode)
						{
								currentModbusParams = currentModbusParams->nextParams;
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, jsonMsg.mode);
						}
						delay_ms(10);
					  if(jsonMsg.fan)
						{
								currentModbusParams = currentModbusParams->nextParams;
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, jsonMsg.fan);	
						}
					}	
					delay_ms(10);
					portEXIT_CRITICAL();
				}
				current_device = current_device->nextDevice;
		}
}
// 重定义pow函数
uint32_t my_pow(int base, int exponent) {
    uint32_t result = 1;
    for(int i = 0; i < exponent; i++) {
        result *= base;
    }
    return result;
}

void protocolsModeFunc(GATEWAY_PARAMS* get, char* buf, char* string)
{
		if(mqtt_connectFlag)
		{
				time1 = GetCurrentTime();

				if(startFlag && time2  <= time1 - ( 10 * 1000))// 60s进行一次全数据发送
				{
						LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"protocolsMode:All data");
						read_device_data(get->device_params, buf, string);
						send_mqtt(buf);				
						time2 = GetCurrentTime();
				}
				else
				{
						read_device_data(get->device_params, buf, string);
						if(count > 0)// count检测string是否含有数据
						{
							LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"protocolsMode:Different data");
							count = 0;
							send_mqtt(string);	
							time2 = GetCurrentTime();		
						}					
				}
				startFlag = 1;	
				memset(buf,0,strlen(buf));				
				memset(string,0,strlen(string));	
				
		}				
}



int transparent_data(DEVICE_PARAMS *device)
{
		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)
		{	
				gd_eval_led_toggle(LED_485TX);			
				if (current_device->protocol == MODBUS_READ)
				{
						uint16_t data[currentModbusParams->registerByteNum /2]; // modbus寄存器长度		
						mmodbus_set16bitOrder(current_device->MDBbigLittleFormat);
						// 读单个寄存器
						if (currentModbusParams->functionCode == 0x03)
						{
//								bool success = mmodbus_readHoldingRegisters16i(0x17,0x00,0x02,data);
  							bool success = mmodbus_readHoldingRegisters16i(currentModbusParams->slaveAddress,
																															 currentModbusParams->registerAddress,
																															 currentModbusParams->registerByteNum /2,
																															 data);
								if (success)
								{
										uint32_t value;		
										if((value - currentModbusParams->value) != 0)
										{
											for (uint8_t i = 0; i < mmodbus.rxBuf[2]; i += 2)
											{
												uint8_t H = mmodbus.rxBuf[i + 3];
												mmodbus.rxBuf[i + 3] = mmodbus.rxBuf[i + 3 + 1];
												mmodbus.rxBuf[i + 3 + 1] = H;
											}
											gd_com_485_send(mmodbus.rxBuf,mmodbus.rxIndex);
											count++;
											vTaskDelay(100);
										}									
								}					
								currentModbusParams = currentModbusParams->nextParams;
								if (currentModbusParams == NULL)	
								{
										current_device = current_device->nextDevice;
										currentModbusParams = current_device->params->gateway_read_modbus_command;
										if(current_device == NULL)
										{
												return 1;
										}
								}												
						}
				}
				else if (current_device->protocol == DLT645_2007 || current_device->protocol == DLT645_97)
				{
						uint8_t read_buf[10];
										
						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(compareArrays(read_buf,currentDLT645Params->data,10))// 不相同1，相同0
								{
									portENTER_CRITICAL();
									gd_com_485_send(mmodbus.rxBuf,mmodbus.rxIndex);
									portEXIT_CRITICAL();
									vTaskDelay(100);
								}
						}
							currentDLT645Params = currentDLT645Params->nextParams;		
							if (currentDLT645Params == NULL)	
							{
									current_device = current_device->nextDevice;
									currentDLT645Params = current_device->params->gateway_read_dlt645_command;
									if(current_device == NULL)
									{
											return 1;
									}
							}										
				}	
		gd_eval_led_off(LED_485TX);					
		}		
		return 1;
	
}
void transparentModeFunc(DEVICE_PARAMS* current_device)
{
			LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"transparentMode:All data");
			printf("transparentMode:All data\n");
			transparent_data(current_device);		
}