jsyhsec/energy_storage
7
0
Fork 0
mirror of https://gitee.com/js-yhsec/energy_storage.git synced 2026-05-27 18:59:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

调试修改MQTT通讯问题

Browse Source
This commit is contained in:
lixiaoyuan
2025-09-17 19:55:59 +08:00
parent 94d6d8a8db
commit c55da0bddc
18 changed files with 661 additions and 494 deletions
Download Patch File Download Diff File Expand all files Collapse all files

349
src/protocol/MqttEntity.cpp
View File

@@ -9,6 +9,20 @@
#define TIMEOUT 10000L
bool MqttClient::load(std::string filename)
{
njson jsonroot;
bool ret = JSON::load(filename, jsonroot);
if (!ret)
{
spdlog::error("[mqtt] load config file failed, filename={}", filename);
return false;
}
return true;
}
std::map<std::string, TopicInfo> MqttClient::s_mapTopicInfo;
int MqttClient::init(string addr, string clientId, string username, string password)
{
@@ -25,7 +39,7 @@ int MqttClient::init(string addr, string clientId, string username, string passw
this->clientId = clientId;
//this->mapTopicInfo["EMS_YX"] = TopicInfo("EMS_YX", 101);
this->mapTopicInfo["EMS_YC"] = TopicInfo("EMS_YC", 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);
@@ -35,14 +49,14 @@ int MqttClient::init(string addr, string clientId, string username, string passw
//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, 1);
//this->mapTopicInfo["MEM_YC"] = TopicInfo("MEM_YC", 3); // 不召测
//this->mapTopicInfo["TH_YC"] = TopicInfo("TH_YC", 10, 1);
//this->mapTopicInfo["Fire40_YX"] = TopicInfo("Fire40_YX", 7, 1);
this->mapTopicInfo["Cooling_YC"] = TopicInfo("Cooling_YC", 14, 1);
this->mapTopicInfo["Cooling_YX"] = TopicInfo("Cooling_YX", 14, 1);
//this->mapTopicInfo["Cooling_YC"] = TopicInfo("Cooling_YC", 14, 1);
//this->mapTopicInfo["Cooling_YX"] = TopicInfo("Cooling_YX", 14, 1);
//this->mapTopicInfo["Gateway_YX"] = TopicInfo("Gateway_YX", 15, 1);
//this->mapTopicInfo["Gateway_YC"] = TopicInfo("Gateway_YC", 15, 1);
//this->mapTopicInfo["Charger_YC"] = TopicInfo("Charger_YC", 106, 1);
//this->mapTopicInfo["Charger_YC"] = TopicInfo("Charger_YC", 106);
MQTTAsync_connectOptions option = MQTTAsync_connectOptions_initializer;
MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
@@ -135,18 +149,21 @@ void MqttClient::subscribe()
MQTTAsync_responseOptions options = MQTTAsync_responseOptions_initializer;
options.onSuccess = funcSuccess;
options.onFailure = funcFailure;
for (auto& item: mapTopicInfo)
for (auto& item: MqttClient::s_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)
if (item.second.enabled)
{
spdlog::error("[mqtt] subscribe [{},{}] failed, err={}", topic, qos, rc);
}
else
{
spdlog::info("[mqtt] subscribe [{},{}] ", topic, qos);
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);
}
}
}
}
@@ -199,10 +216,10 @@ int MqttClient::polling()
return 0;
}
for (auto& item: mapTopicInfo)
for (auto& item: MqttClient::s_mapTopicInfo)
{
auto& topicInfo = item.second;
if (topicInfo.polling)
if (topicInfo.polling && topicInfo.enabled)
{
std::vector<std::shared_ptr<Device>> vecDevice;
station->getDeviceByType(topicInfo.deviceType, vecDevice);
@@ -268,96 +285,6 @@ void MqttClient::onConnectFaiure(MQTTAsync_failureData* resp)
this->destory();
}
void MqttClient::ParseArrivedMessage(njson& json, string clientId, string command, std::shared_ptr<Station> station)
{
std::string stationNo = clientId;
auto mapRegPtr = REGAddr::getRegMap(command);
if (!mapRegPtr)
{
spdlog::error("[mqtt] get register add info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
return;
}
auto iterTopic = mapTopicInfo.find(command);
if (iterTopic == mapTopicInfo.end())
{
spdlog::error("[mqtt] get topic info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
return;
}
TopicInfo& topicInfo = iterTopic->second;
int deviceNo = -1;
JSON::read(json, "no", deviceNo);
auto device = station->getDeviceByType(topicInfo.deviceType, 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())
{
std::string addrText;
auto iter = mapRegPtr->find(key);
for (int i = 0; i<data.size(); ++i)
{
if (iter != mapRegPtr->end())
{
auto addr = iter->first;
int val = data[i];
device->setParam(addr, val);
spdlog::info("[mqtt] read [{}]={},{}", addr, val, iter->second.remark);
if (command == "EMS_YC" && addr == "0x110C")
{
int a = 30;
a = 100;
}
if (command == "EMS_YC")
{
station->setRuntimeData(addr, val);
}
else if (command == "Fire40_YX")
{
station->setFire40Data(deviceNo, addr, val);
}
else if (command == "TH_YC")
{
station->setTHData(deviceNo, addr, val);
}
else if (command == "Cooling_YX" || command == "Cooling_YC")
{
station->setCoolingData(deviceNo, addr, val);
}
else if (command == "Gateway_YX")
{
//if (key == "CDZ") "CDZ": 1, //充电桩 1：在线，0：离线
//else if (key == "EMU") //储能 1：在线，0：离线
}
++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>()));
}
}
}
}
int MqttClient::onMessageArrived(char* topic, int topicLen, MQTTAsync_message* msg)
{
std::string topicStr = topic;
@@ -412,7 +339,7 @@ int MqttClient::onMessageArrived(char* topic, int topicLen, MQTTAsync_message* m
}
else
{
ParseArrivedMessage(json, clientId, command, station);
ParseArrivedMessage(json, command, station);
}
// 必须释放消息内存！
@@ -422,117 +349,119 @@ int MqttClient::onMessageArrived(char* topic, int topicLen, MQTTAsync_message* m
}
string MQTT::pack(std::string name)
void MqttClient::ParseArrivedMessage(njson& json, string command, std::shared_ptr<Station> station)
{
njson json;
json["ts"] = Utils::time();
json["no"] = 1;
std::string stationNo = clientId;
if (name == "EMS_YC")
auto iterTopic = MqttClient::s_mapTopicInfo.find(command);
if (iterTopic == MqttClient::s_mapTopicInfo.end())
{
//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"};
spdlog::error("[mqtt] get topic info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
return;
}
else if (name == "PCS_YC")
TopicInfo& topicInfo = iterTopic->second;
int deviceNo = -1;
JSON::read(json, "no", deviceNo);
auto device = station->getDeviceByType(topicInfo.deviceType, Utils::toStr(deviceNo));
if (!device)
{
//总充电量 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"};
spdlog::error("[mqtt] get device info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
return;
}
else if (name == "PCU_YC")
auto mapRegPtr = REGAddr::getRegMap(command);
if (!mapRegPtr)
{
//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"};
spdlog::error("[mqtt] get register add info error, clientId={}, stationId={}, command={}", clientId, stationNo, command);
return;
}
else if (name == "BMS_YC")
for (auto& item: json.items())
{
//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
std::string key = item.key();
if (key == "ts" || key == "no")
{
continue;
}
auto data = json[key];
if (data.is_array())
{
if (command == "Charger_YC")
{
if (key == "1") key = "11";
else if (key == "2") key = "21";
}
std::string addrText;
auto iter = mapRegPtr->find(key);
if (iter != mapRegPtr->end())
{
for (int i = 0; i<data.size(); ++i)
{
auto addr = iter->first;
int val = data[i];
device->setParam(addr, val);
spdlog::debug("[mqtt] read [{}]={}, {}", addr, val, iter->second.remark);
json["addr"] = {"0x0001", "0x0002", "0x0003", "0x0005", "0x0007", "0x0009", "0x0047", "0x0048", "0x0049", "0x004A"};
if (command == "MEM_YC") { station->setRuntimeData(deviceNo, addr, val); }
else if (command == "Fire40_YX") { station->setFire40Data(deviceNo, addr, val); }
else if (command == "TH_YC") { station->setTHData(deviceNo, addr, val); }
else if (command == "Cooling_YX" || command == "Cooling_YC") { station->setCoolingData(deviceNo, addr, val); }
else if (command == "Gateway_YX")
{
//if (key == "CDZ") "CDZ": 1, //充电桩 1：在线，0：离线
//else if (key == "EMU") //储能 1：在线，0：离线
}
++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>()));
}
}
}
std::vector<std::string> KEY_CHARGER_1 = {"31071", "31073", "31075", "31077", "31079", "31081", "31083"};
std::vector<std::string> KEY_CHARGER_2 = {"31072", "31074", "31076", "31078", "31080", "31082", "31084"};
void MqttClient::ParseMessageCharge(njson& json, string command, std::shared_ptr<Station> station, std::shared_ptr<Device> device)
{
if (json.contains("1"))
{
auto& jsondata = json["1"];
if (jsondata.is_array())
{
for (int i = 0; i<jsondata.size(); i++)
{
if (i<KEY_CHARGER_1.size())
{
auto& addr = KEY_CHARGER_1[i];
auto val = jsondata[i].get<int>();
device->setParam(addr, val);
spdlog::info("[mqtt] read: 枪1 [{}]={}", addr, val);
}
}
}
}
if (json.contains("2"))
{
auto& jsondata = json["2"];
if (jsondata.is_array())
{
for (int i = 0; i<jsondata.size(); i++)
{
if (i<KEY_CHARGER_2.size())
{
auto& addr = KEY_CHARGER_2[i];
auto val = jsondata[i].get<int>();
device->setParam(addr, val);
spdlog::info("[mqtt] read: 枪2 [{}]={}", addr, val);
}
}
}
}
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();
}