energy_storage/src/protocol/MqttEntity.cpp

#include "MqttEntity.h"
#include "common/Spdlogger.h"
#include "common/Utils.h"
#include "app/Application.h"
#include "app/AppData.h"
#include "app/Station.h"
#include "app/Device.h"

#define TIMEOUT 10000L

std::string REGAddrOffset(std::string addr, int offset)
{
	unsigned int val;
	std::stringstream ss;
	ss << std::hex << addr;
	ss >> val;
	return Utils::toHexStr(val + offset);
}

static std::map<std::string, std::map<std::string, REGInfo>> g_mapRegInfo;

void MqttClient::loadDataStruct(std::string filename)
{
	njson json;
	JSON::load(filename, json);

	// 遍历 JSON 对象
	for (auto& jsonitem : json.items()) 
	{
		std::string name = jsonitem.key();
		auto& jsonnodeItem = jsonitem.value();
		//int count = jsonnodeItem["count"];
		auto jsonaddrs = jsonnodeItem["addr"];

		auto& mapItem = g_mapRegInfo[name];
		int size = 0;
		for (int i = 0; i<2; ++i)
		{
			for (auto& item : jsonaddrs)
			{
				std::string addr = item["key"];
				if (i > 0)
				{
					addr = REGAddrOffset(addr, size*i);
				}
				mapItem[addr] = REGInfo(addr, item["datatype"], item["remark"]);
				if (i ==0)
				{
					size += mapItem[addr].bytes;
				}
			}
		}
	}
}
//

int MqttClient::init(string addr, string clientId, string username, string password)
{
	if (isConnected)
	{
		return MQTTASYNC_SUCCESS;
	}
	if (addr.empty())
	{
		return MQTTASYNC_FAILURE;
	}

	this->addr = addr;
	this->clientId = clientId;

	this->mapTopicInfo["EMS_YX"] = TopicInfo("EMS_YX", 101);
	this->mapTopicInfo["EMS_YC"] = TopicInfo("EMS_YC", 101);
	this->mapTopicInfo["EMS_YT"] = TopicInfo("EMS_YT", 101);
	this->mapTopicInfo["PCS_YX"] = TopicInfo("PCS_YX", 102, 1);
	this->mapTopicInfo["PCS_YC"] = TopicInfo("PCS_YC", 102, 1);
	this->mapTopicInfo["PCU_YX"] = TopicInfo("PCU_YX", 103);
	this->mapTopicInfo["PCU_YC"] = TopicInfo("PCU_YC", 103);
	this->mapTopicInfo["BMS_YX"] = TopicInfo("BMS_YX", 104);
	this->mapTopicInfo["BMS_YC"] = TopicInfo("BMS_YC", 104);
	this->mapTopicInfo["BCU_YX"] = TopicInfo("BCU_YX", 105, 1);
	this->mapTopicInfo["BCU_YC"] = TopicInfo("BCU_YC", 105, 1);
	this->mapTopicInfo["MEM_YC"] = TopicInfo("MEM_YC", 3);
	this->mapTopicInfo["Cooling_YC"] = TopicInfo("Cooling_YC", 110);
	this->mapTopicInfo["TH_YC"] = TopicInfo("TH_YC", 111);
	this->mapTopicInfo["Gateway_YX"] = TopicInfo("Gateway_YX", 112);
	this->mapTopicInfo["Charger_YC"] = TopicInfo("Charger_YC", 113);

	MQTTAsync_connectOptions option = MQTTAsync_connectOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	int rc {0};

	// "tcp://localhost:1883"
	std::string str = "ESS-" + std::to_string(Utils::random(1000, 9999)) + "-" +clientId;
	rc = MQTTAsync_create(&client, addr.c_str(), str.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL);
	if (rc != MQTTASYNC_SUCCESS)
	{
		spdlog::error("[mqtt] MQTTAsync_create error: {}", rc);
		return rc;
	}

	MQTTAsync_connectionLost* onConnectionLost = 
		[](void* context, char* cause)
		{
			static_cast<MqttClient*>(context)->onConnectionLost(cause);
		};
	MQTTAsync_messageArrived* onMessageArrived =
		[](void* context, char* topicName, int topicLen, MQTTAsync_message* message)->int 
		{
			return static_cast<MqttClient*>(context)->onMessageArrived(topicName, topicLen, message);
		};
	MQTTAsync_deliveryComplete* onDeliveryComplete =
		[](void* context, MQTTAsync_token token) 
		{
		};


	//设置连接丢失、接受消息的回调函数
	rc = MQTTAsync_setCallbacks(client, this, onConnectionLost, onMessageArrived, onDeliveryComplete);
	if (rc != MQTTASYNC_SUCCESS)
	{
		spdlog::error("[mqtt] MQTTAsync_setCallbacks error");
		this->destory();
		return rc;
	}

	option.keepAliveInterval = 20;
	option.cleansession = 1;
	option.onSuccess = [](void* context, MQTTAsync_successData* resp)  { static_cast<MqttClient*>(context)->onConnectSuccess(resp); };;
	option.onFailure = [](void* context, MQTTAsync_failureData* resp) { static_cast<MqttClient*>(context)->onConnectFaiure(resp); };
	option.context = this;
	option.username = username.c_str();
	option.password = password.c_str();

	//断开重连设置
	option.automaticReconnect = 1;//设置非零，断开自动重连
	option.minRetryInterval = 5; //单位秒，重连间隔次数，每次重新连接失败时，重试间隔都会加倍，直到最大间隔
	option.maxRetryInterval = 60;//单位秒，最大重连尝试间隔

	rc = MQTTAsync_connect(client, &option);
	if (rc != MQTTASYNC_SUCCESS)
	{
		spdlog::error("[mqtt] MQTTAsync_connect error");
		return rc;
	}
	return 0;

	//MQTTAsync_disconnect(client, NULL);
	//MQTTAsync_destroy(&client);
}

void MqttClient::destory()
{
	return;
	if (client)
	{
		MQTTAsync_destroy(&client);
	}
}

struct SubscribInfo
{
	std::function<void(int id)> callback;
};

void MqttClient::subscribe()
{
	MQTTAsync_onSuccess* funcSuccess = [](void* context, MQTTAsync_successData* response)
		{
			spdlog::info("[mqtt] subscribe {} success.", (char*)context);
		};
	MQTTAsync_onFailure* funcFailure = [](void* context, MQTTAsync_failureData* response)
		{
			spdlog::error("[mqtt] subscribe {} failed.", (char*)context);
		};

	MQTTAsync_responseOptions options = MQTTAsync_responseOptions_initializer;
	options.onSuccess = funcSuccess;
	options.onFailure = funcFailure;
	for (auto& item: mapTopicInfo)
	{
		std::string topic = "up/json/" + clientId + "/" + item.first;
		options.context = (void*)&item.first;
		int rc = MQTTAsync_subscribe(client, topic.data(), qos, &options);
		if (rc != MQTTASYNC_SUCCESS)
		{
			spdlog::error("[mqtt] subscribe [{},{}] failed, err={}", topic, qos, rc);
		}
		else
		{
			spdlog::info("[mqtt] subscribe [{},{}] ", topic, qos);
		}
	}
}

int MqttClient::publish(std::string topic, std::string text)
{
	MQTTAsync_responseOptions options = MQTTAsync_responseOptions_initializer;
	options.onSuccess = [](void* context, MQTTAsync_successData* response) {};
	options.onFailure = [](void* context, MQTTAsync_failureData* response) {};
	options.context = this;

	MQTTAsync_message msg = MQTTAsync_message_initializer;
	msg.qos = this->qos;
	msg.payload = text.data();
	msg.payloadlen = text.size();
	msg.retained = 0;

	std::string topicName = "down/json/" + clientId + "/" + topic;
	int rc = MQTTAsync_sendMessage(client, topicName.c_str(), &msg, &options);
	if (rc == MQTTASYNC_SUCCESS)
	{
		spdlog::info("[mqtt] publish MQTTAsync_sendMessage success, topic={}, text={}", topicName, text);
	}
	else
	{
		spdlog::error("[mqtt] publish MQTTAsync_sendMessage error, topic={}, text={}", topicName, text);
	}
	return 0;
}

int MqttClient::polling()
{
	if (!isConnected)
	{
		return 0;
	}

	njson json;
	json["ts"] = Utils::time();
	json["no"] = 0; // 设备编号

	auto& appdata = Application::data();
	auto station = appdata.getStationByCode(clientId);

	for (auto& item: mapTopicInfo)
	{
		auto& topicInfo = item.second;
		if (topicInfo.polling)
		{
			if (station)
			{
				std::vector<std::shared_ptr<Device>> vecDevice;
				station->getDeviceByType(topicInfo.deviceType, vecDevice);
				for (auto device: vecDevice)
				{
					json["no"] = Utils::toInt(device->code);
					this->publish(topicInfo.name, json.dump());
				}
			}
		}
	}
	return 0;
}

void MqttClient::onConnectionLost(char* cause)
{
	this->isConnected = false;
	//this->destory();
	spdlog::error("MQTT connection lost, cause={}", cause);
}

std::string GetSubStr(std::string c, std::string& str)
{
	std::string v;
	int pos = str.find_first_of("/");
	if (pos != string::npos)
	{
		v = str.substr(0, pos);
		str = str.substr(pos+1);
	}
	else
	{
		v = str;
		str = "";
	}
	return v;
}

int MqttClient::onMessageArrived(char* topic, int topicLen, MQTTAsync_message* msg)
{
	std::string topicStr = topic;
	int len = msg->payloadlen;
	std::string payload((const char*)msg->payload, len);


	// <数据方向>/<数据格式>/<厂家ID>/<指合>/<设备标识，上行可选>
	std::string direction = GetSubStr("/", topicStr);
	std::string datatype = GetSubStr("/", topicStr);
	std::string stationNo = GetSubStr("/", topicStr);
	std::string command = GetSubStr("/", topicStr);
	std::string deviceCode = GetSubStr("/", topicStr);

	spdlog::info("[mqtt] <<<<<<<<<< message arrived: topic=[{},{}], len={}, payload={}", topic, msg->qos, len, payload);

	njson json;
	bool ret = JSON::parse(payload, json);
	if (!ret)
	{
		spdlog::error("[mqtt] json parse error.");
		return 1;
	}
	auto station = Application::data().getStationByCode(stationNo);
	if (!station)
	{
		spdlog::error("[mqtt] get station error, clientId={}, stationId={}", clientId, stationNo);
		return 1;
	}

	auto iter = g_mapRegInfo.find(command);
	if (iter == g_mapRegInfo.end())
	{
		spdlog::error("[mqtt] get register add info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
		return 1;
	}
	std::map<std::string, REGInfo>& mapRegInfo = iter->second;


	auto iterTopic = mapTopicInfo.find(command);
	if (iterTopic == mapTopicInfo.end())
	{
		spdlog::error("[mqtt] get topic info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
		return 1;
	}
	TopicInfo& topicInfo = iterTopic->second;

	int deviceNo = -1;
	JSON::read(json, "no", deviceNo);
	auto device = station->getDeviceByType(topicInfo.deviceType, Utils::toStr(deviceNo));
	if (!device)
	{
		return 1;
	}

	for (auto& item: json.items())
	{
		std::string key = item.key();
		if (key != "ts" && key != "no")
		{
			auto data = json.at(key);
			if (data.is_array())
			{
				auto iter = mapRegInfo.find(key);
				for (int i = 0; i<data.size(); ++i)
				{
					if (iter != mapRegInfo.end())
					{
						auto addr = iter->first;
						auto& val = data[i];
						//spdlog::info("[mqtt] read register addr: [{}]={}, {}", addr, val, iter->second.remark);
						device->setParam(addr, val);
						++iter;
					}
				}
			}
			else if (data.is_number())
			{
				device->setParam(key, data.get<int>());
			}
			else if (data.is_string())
			{
				device->setParam(key, Utils::toInt(data.get<std::string>()));
			}
		}
	}

	// 必须释放消息内存！
	MQTTAsync_freeMessage(&msg);
	MQTTAsync_free(topic);
	return 1; // 1表示消息已经处理
}

// 交付完成回调（可选）
void MqttClient::onDeliveryComplete(MQTTAsync_token token) 
{
	//spdlog::info("MQTT delivery complete, token={}", token);
}

void MqttClient::onConnectSuccess( MQTTAsync_successData* resp)
{
	spdlog::info("[mqtt] connect to {} success, clientId={}.", addr, clientId);
	this->isConnected = true;
	this->subscribe();
}

void MqttClient::onConnectFaiure(MQTTAsync_failureData* resp)
{
	spdlog::error("[mqtt] connect to {} error, clientId={}.", addr, clientId);
	this->isConnected = false;
	this->destory();
}

void MqttClient::parseEMS_YX(std::shared_ptr<Station> station, njson& json, std::map<std::string, REGInfo>& mapRegInfo)
{
	int deviceNo = -1;
	JSON::read(json, "no", deviceNo);
	auto device = station->getDeviceByType(101, Utils::toStr(deviceNo));
	if (!device)
	{
		return;
	}

	for (auto& item: json.items())
	{
		std::string key = item.key();
		if (key != "ts" && key != "no")
		{
			auto data = json.at(key);
			if (data.is_array())
			{
				auto iter = mapRegInfo.find(key);
				for (int i = 0; i<data.size(); ++i)
				{
					if (iter != mapRegInfo.end())
					{
						auto addr = iter->first;
						device->mapParams[addr] = JSON::readStr(data[i], addr);
						++iter;
					}
				}
			}
			else if (data.is_number())
			{
				device->mapParams[key] = Utils::toStr(data.get<int>());
			}
			else if (data.is_string())
			{
				device->mapParams[key] = Utils::toStr(data.get<int>());
			}
		}
	}
}


string MQTT::pack(std::string name)
{
	njson json;
	json["ts"] = Utils::time();
	json["no"] = 1;

	if (name == "EMS_YC")
	{
		//A相电压	R	uint32	1V			0x107E
		//B相电压	R	uint32	1V			0x1080
		//C相电压	R	uint32	1V			0x1082
		//A相电流	R	int32	1A			0x1084
		//B相电流	R	int32	1A			0x1086
		//C相电流	R	int32	1A			0x1088

		//储能系统SOC	R	uint16	0.1			0x107A
		//储能系统SOH	R	uint16	0.1			0x107B

		json["addr"] = {"0x107A", "0x107B", "0x107E", "0x1080", "0x1082", "0x1084", "0x1086", "0x1088"};
	}
	else if (name == "PCS_YC")
	{
		//总充电量	R	uint32	1kWh			0x0003
		//总放电量	R	uint32	1kWh			0x0005

		//A相电压	R	int16	1V			0x0010
		//B相电压	R	int16	1V			0x0011
		//C相电压	R	int16	1V			0x0012

		//A相电流	R	int16	1A			0x0019
		//B相电流	R	int16	1A			0x001A
		//C相电流	R	int16	1A			0x001B
	
		//三相总有功功率	R	int16	1kW			0x0025
		//三相总无功功率	R	int16	1kVar			0x0026
		//三相总视在功率	R	int16	1kVA			0x0027
		//三相总功率因数	R	int16	1			0x0028

		//充电功率	R	int16	1kW			0x002C
		//放电功率	R	int16	1kW			0x002D
		json["addr"] = {"0x0003", "0x0005", "0x0010", "0x0011", "0x0012", "0x0019", "0x001A", "0x001B", "0x0025", "0x0026", "0x0027", "0x0028", "0x002C", "0x002D"};
	}
	else if (name == "PCU_YC")
	{
		//PCS侧线A相电压	R	int16	1v			0x0013
		//PCS侧线B相电压	R	int16	1v			0x0014
		//PCS侧线C相电压	R	int16	1v			0x0015

		//PCS侧功率因数A	R	int16	1			0x0019
		//PCS侧功率因数B	R	int16	1			0x001A
		//PCS侧功率因数C	R	int16	1			0x001B

		//PCS侧相电流A	R	int16	1A			0x001C
		//PCS侧相电流B	R	int16	1A			0x001D
		//PCS侧相电流C	R	int16	1A			0x001E

		//PCS侧三相总有功功率	R	int16	1kW			0x0028
		//PCS侧三相总无功功率	R	int16	1kVar			0x0029
		//PCS侧三相总视在功率	R	int16	1kVA			0x002A
		//PCS侧三相总功率因数	R	int16	1			0x002B

		json["addr"] = {"0x0013", "0x0014", "0x0015", "0x1080", "0x1082", "0x1084", "0x1086", "0x1088"};
	}
	else if (name == "BMS_YC")
	{
		//SOC	R	uint16	0.1			0x0001
		//SOH	R	uint16	0.1			0x0002
		//电压	R	uint32	0.1V			0x0003
		//电流	R	int32	0.1A			0x0005
		//可充电量	R	uint32	1kWh			0x0007
		//可放电量	R	uint32	1kWh			0x0009
		//可充电状态	R	uint16			1：可充电；0：不可充电	0x0047
		//可放电状态	R	uint16			1：可放电；0：不可放电	0x0048
		//运行状态	R	uint16			运行状态 0-正常 1-告警 2-保护	0x0049
		//充放电状态 	R	uint16			0-待机 1-充电 2-放电	0x004A

		json["addr"] = {"0x0001", "0x0002", "0x0003", "0x0005", "0x0007", "0x0009", "0x0047", "0x0048", "0x0049",  "0x004A"};
	}
	else if (name == "BCU_YC")
	{
		//电表类型	R	uint16		"0：储能站总表 1：逆变前侧电表 2：逆变后侧电表 3：配电柜电表 4：并网口电表"		0x0008
		//A相电压	R	uint32	1V			0x000B
		//B相电压	R	uint32	1V			0x000D
		//C相电压	R	uint32	1V			0x000F
		//A相电流	R	int32	1A			0x0011
		//B相电流	R	int32	1A			0x0013
		//C相电流	R	int32	1A			0x0015

		//尖段电价	R	uint32	1RMB			0x0027
		//峰段电价	R	uint32	1RMB			0x0029
		//平段电价	R	uint32	1RMB			0x002B
		//谷段电价	R	uint32	1RMB			0x002D
		//日充电电量	R	uint32	1kWh			0x002F
		//日放电电量	R	uint32	1kWh			0x0031
		//日充电费用	R	uint32	1RMB			0x0033
		//日放电费用	R	uint32	1RMB			0x0035
		//日收益	R	int32	1RMB			0x0037

		//总充电电量	R	uint32	1kWh			0x004D
		//总放电电量	R	uint32	1kWh			0x004F
		//总充电费用	R	uint32	1RMB			0x0051
		//总放电费用	R	uint32	1RMB			0x0053
		//总收益	R	int32	1RMB			0x0055
	}
	else if (name == "TH_YC")
	{
		//所属通道号	R	uint16		1		0x0001
		//所属温湿度号	R	uint16		1~10		0x0002
		//温度	R	int16	0.1℃			0x0003
		//湿度	R	int16	0.1℃			0x0004

	}
	return json.dump();
}