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.h"
#include "dlt645_1997_private.h"
#include "dlt645_port.h"
#include "gd32_flash.h"
#include "protocol.h"
#include "sys_http.h"
#include "jsmn.h"
#include "timer.h"

uint8_t time_count = 0, jsonCunt = 0, count = 0;
uint8_t protocol_485;
int time1,time2;
int ID = 0;

void data_task(void *pdata)
{
		 
		dlt645_init(1);
		mmodbus_init(1);
		while(!load_http_config)
		{
			vTaskDelay(100);
		}

		portENTER_CRITICAL();
		char *device_config_json=pvPortMalloc( 10 *1024 );
		read_data_from_flash(device_config_json);
		addGatewayParams(device_config_json);	
		vPortFree(device_config_json);
		device_config_json = NULL;
		portEXIT_CRITICAL();
		
		GATEWAY_PARAMS *get;
		get= get_gateway_config_params();
		DEVICE_PARAMS *current_device=get->device_params;
		
		char *buf = pvPortMalloc(1024);	// 接收读取的数据
		char *string = pvPortMalloc(1024);	// 接收读取的数据
		memset(buf,0,strlen(buf));
		memset(string,0,strlen(string));
		
		while (current_device!=NULL)
		{		
			time1 = GetCurrentTime();
			if(mqtt_connectFlag)
			{
				if(jsonCunt && time2  <= time1 - ( 10 * 1000))// 60s进行一次全数据发送
				{
						read_device_data(current_device, buf, string);
						send_mqtt(buf);
						memset(buf,0,strlen(buf));					
						time2 = GetCurrentTime();
					current_device=get->device_params;
				}
				else
				{
					read_device_data(current_device, buf, string);
					if(count > 0)// count检测是否含有数据
					{
						send_mqtt(string);
						memset(string,0,strlen(string));
						current_device=get->device_params;
						count = 0;
						if(!time_count)
						{
							jsonCunt = 1;
							time_count = 1;
							time2 = GetCurrentTime();
						}
					}
				}	
				memset(buf,0,strlen(buf));				
			}		
			
			vTaskDelay(100);
		}	
		vPortFree(buf);
		buf = NULL;
		
}	

/*
*********************************************************************************************************
*	函 数 名: 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)
		{				
				if (current_device->protocol == MODBUS_READ)
				{
						protocol_485=1;
						uint8_t state;
						uint16_t data[currentModbusParams->registerByteNum /2]; // modbus寄存器长度
						uint8_t data1[currentModbusParams->registerByteNum /2];
						mmodbus_set16bitOrder(current_device->MDBbigLittleFormat);
						// 读水阀状态
						if(currentModbusParams->functionCode == 0x01)
						{
							bool success = mmodbus_readCoil(currentModbusParams->slaveAddress,
																		 currentModbusParams->registerByteNum /2,
																		 data1);
							if(success)
							{
								uint8_t value;
								value = data1[0];
								if(value == 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
										{
//												float convertedValue = (float)value / pow(10, currentModbusParams->decimalPoint);
//												currentModbusParams->value=convertedValue;
										}
																	
								}					
								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,
																							 state);
							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)
				{
						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);
								count++;
			
						}
		
								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){
		GATEWAY_PARAMS *get;
		get= get_gateway_config_params();
//		char *pubJsonString = pvPortMalloc(strlen(buf)); 
		char pubJsonString[500];
	  sprintf(pubJsonString,"ID: %d {\"DEVICEID\":\"%s\",\"data\":[%s]}",ID++, get->deviceId, buf);	// 组成要发送的json语句							
	 	mqtt_qos_t qos = QOS0;
		uint16_t pub_length = strlen(pubJsonString);
		mqtt_publish_data(pubJsonString, qos, pub_length, (char*)&get->messageTopic);	
//		vPortFree(pubJsonString);
//		pubJsonString = NULL;
}

/*
*********************************************************************************************************
*	函 数 名: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
				{
					delay_ms(200);
					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(100);
						if(jsonMsg.temp)
						{
								currentModbusParams = currentModbusParams->nextParams;							
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, 
																								jsonMsg.temp);	
						}
						delay_ms(100);
						if(jsonMsg.mode)
						{
								currentModbusParams = currentModbusParams->nextParams;
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, 
																								jsonMsg.mode);
						}
						delay_ms(100);
					  if(jsonMsg.fan)
						{
								currentModbusParams = currentModbusParams->nextParams;
								mmodbus_writeHoldingRegister16i(currentModbusParams->slaveAddress, 
																								currentModbusParams->registerAddress, 
																								jsonMsg.fan);	
						}
						delay_ms(100);
					}	
				}
				current_device = current_device->nextDevice;
		}
}