|
|
@@ -11,6 +11,7 @@
|
|
|
package com.ruoyi.ems.gcc.handle;
|
|
|
|
|
|
import com.alibaba.fastjson2.JSON;
|
|
|
+import com.alibaba.fastjson2.JSONArray;
|
|
|
import com.alibaba.fastjson2.JSONObject;
|
|
|
import com.huashe.common.utils.DateUtils;
|
|
|
import com.ruoyi.ems.common.handle.BaseDevHandler;
|
|
|
@@ -18,6 +19,7 @@ import com.ruoyi.ems.common.mqtt.MqttMessageHandler;
|
|
|
import com.ruoyi.ems.common.mqtt.MqttTemplate;
|
|
|
import com.ruoyi.ems.domain.ElecStoreH;
|
|
|
import com.ruoyi.ems.domain.EmsDevice;
|
|
|
+import com.ruoyi.ems.domain.EmsObjAbilityCallLog;
|
|
|
import com.ruoyi.ems.domain.EmsObjAttrValue;
|
|
|
import com.ruoyi.ems.enums.DevOnlineStatus;
|
|
|
import com.ruoyi.ems.gcc.model.GccSite;
|
|
|
@@ -106,29 +108,26 @@ public class GccHandler extends BaseDevHandler implements MqttMessageHandler {
|
|
|
public static final String ABILITY_APPLY_STRATEGY_IDLE = "applyStrategyIdle";
|
|
|
|
|
|
/**
|
|
|
- * 下行/应答topic后缀
|
|
|
+ * 站点策略属性key前缀(每站点一条:strategy_{deviceId})
|
|
|
*/
|
|
|
- private static final String TOPIC_SUFFIX_COMMAND = "/command";
|
|
|
- private static final String TOPIC_SUFFIX_COMMAND_RESPONSE = "/commandResponse";
|
|
|
+ private static final String STRATEGY_ATTR_PREFIX = "strategy_";
|
|
|
|
|
|
/**
|
|
|
- * 策略时段保守功率预置值(kW,待厂家确认PCS额定功率后调整)
|
|
|
+ * 策略报文code(厂家模板:strategy1)与系统标志位(协议info_1)
|
|
|
*/
|
|
|
- private static final int SLOT_POWER_LIMIT_KW = 100;
|
|
|
+ private static final String STRATEGY_CODE = "strategy1";
|
|
|
+ private static final int STRATEGY_SYS_FLAG_DEFAULT = 128;
|
|
|
|
|
|
/**
|
|
|
- * 默认策略时段(厂家描述:02-06谷时充电、11-13谷时补电;光伏随动时段协议无字段暂禁止)
|
|
|
- * 字段:{峰谷选择, 开始小时, 开始分钟, 结束小时, 结束分钟, 充电最大功率kW, 放电最大功率kW}
|
|
|
+ * 下行/应答topic后缀
|
|
|
*/
|
|
|
- private static final int[][] DEFAULT_STRATEGY_SLOTS = {
|
|
|
- { 4, 2, 0, 6, 0, SLOT_POWER_LIMIT_KW, 0 },
|
|
|
- { 4, 11, 0, 13, 0, SLOT_POWER_LIMIT_KW, 0 }
|
|
|
- };
|
|
|
+ private static final String TOPIC_SUFFIX_COMMAND = "/command";
|
|
|
+ private static final String TOPIC_SUFFIX_COMMAND_RESPONSE = "/commandResponse";
|
|
|
|
|
|
/**
|
|
|
- * 全静置策略:无有效时段(24时段全部禁止)
|
|
|
+ * 待回写应答的调用日志:deviceId -> call_log_id
|
|
|
*/
|
|
|
- private static final int[][] IDLE_STRATEGY_SLOTS = {};
|
|
|
+ private final Map<String, Long> pendingCallLogIds = new ConcurrentHashMap<>();
|
|
|
|
|
|
/**
|
|
|
* MQTT操作模板(由订阅器建连后绑定,用于策略下发)
|
|
|
@@ -344,10 +343,9 @@ public class GccHandler extends BaseDevHandler implements MqttMessageHandler {
|
|
|
|
|
|
@Override
|
|
|
public void handle(String topic, byte[] payload) {
|
|
|
- // 策略下发应答:当前阶段仅记录日志,不走遥测解析链路
|
|
|
+ // 策略下发应答:记日志并回写调用日志(含topic与payload)
|
|
|
if (StringUtils.endsWith(topic, TOPIC_SUFFIX_COMMAND_RESPONSE)) {
|
|
|
- log.info("储能策略下发应答, topic[{}], payload[{}]", topic,
|
|
|
- new String(payload, StandardCharsets.UTF_8));
|
|
|
+ handleCommandResponse(topic, payload);
|
|
|
return;
|
|
|
}
|
|
|
|
|
|
@@ -1036,8 +1034,8 @@ public class GccHandler extends BaseDevHandler implements MqttMessageHandler {
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 能力调用:策略下发(当前支持默认策略/全静置策略)
|
|
|
- * 链路:前端 -> ems-server能力转发 -> 本方法 -> 逐站点发布command报文 -> EMS应答commandResponse(记日志)
|
|
|
+ * 能力调用:策略下发(当前支持站点策略下发/全静置策略)
|
|
|
+ * 链路:前端 -> ems-server能力转发 -> 本方法 -> 逐站点发布command报文 -> EMS应答commandResponse(回写调用日志)
|
|
|
*
|
|
|
* @param abilityParam 能力参数
|
|
|
* @return 调用结果
|
|
|
@@ -1049,74 +1047,166 @@ public class GccHandler extends BaseDevHandler implements MqttMessageHandler {
|
|
|
}
|
|
|
|
|
|
String abilityKey = abilityParam.getAbilityKey();
|
|
|
- int[][] slots;
|
|
|
if (StringUtils.equals(ABILITY_APPLY_DEFAULT_STRATEGY, abilityKey)) {
|
|
|
- slots = DEFAULT_STRATEGY_SLOTS;
|
|
|
+ return applySiteStrategy(abilityParam);
|
|
|
}
|
|
|
else if (StringUtils.equals(ABILITY_APPLY_STRATEGY_IDLE, abilityKey)) {
|
|
|
- slots = IDLE_STRATEGY_SLOTS;
|
|
|
+ return applyIdleStrategy(abilityParam);
|
|
|
}
|
|
|
- else {
|
|
|
- return new CallResponse<>(-1, "不支持的能力: " + abilityKey);
|
|
|
+
|
|
|
+ return new CallResponse<>(-1, "不支持的能力: " + abilityKey);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 下发站点策略:读取系统策略属性(每站点一条,strategy_{deviceId})逐站点下发
|
|
|
+ * abilityParam为空下发全部站点,传站点deviceId则仅下发指定站点;时段内容以attr_value维护的JSON为准(Web端可编辑)
|
|
|
+ *
|
|
|
+ * @param abilityParam 能力参数
|
|
|
+ * @return 调用结果
|
|
|
+ */
|
|
|
+ private CallResponse<Void> applySiteStrategy(AbilityPayload abilityParam) {
|
|
|
+ String targetDeviceId = StringUtils.trimToNull(abilityParam.getAbilityParam());
|
|
|
+
|
|
|
+ // 查询系统属性值,筛选策略属性(attrKey以strategy_开头)
|
|
|
+ List<EmsObjAttrValue> attrValues = objAttrValueService.selectByObjCode(SYS_MODEL_CODE, SUBSYSTEM_CODE);
|
|
|
+ List<EmsObjAttrValue> strategyAttrs = new ArrayList<>();
|
|
|
+
|
|
|
+ for (EmsObjAttrValue value : attrValues) {
|
|
|
+ if (!StringUtils.startsWith(value.getAttrKey(), STRATEGY_ATTR_PREFIX)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (null == targetDeviceId
|
|
|
+ || StringUtils.equals(STRATEGY_ATTR_PREFIX + targetDeviceId, value.getAttrKey())) {
|
|
|
+ strategyAttrs.add(value);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (strategyAttrs.isEmpty()) {
|
|
|
+ return new CallResponse<>(-1,
|
|
|
+ null == targetDeviceId ? "未配置站点策略" : "站点[" + targetDeviceId + "]无策略配置");
|
|
|
}
|
|
|
|
|
|
+ int cnt = 0;
|
|
|
+ StringBuilder detail = new StringBuilder();
|
|
|
+
|
|
|
+ for (EmsObjAttrValue attr : strategyAttrs) {
|
|
|
+ EmsObjAbilityCallLog callLog = null;
|
|
|
+ String deviceId = null;
|
|
|
+
|
|
|
+ try {
|
|
|
+ JSONObject strategy = JSON.parseObject(attr.getAttrValue());
|
|
|
+ deviceId = strategy.getString("deviceId");
|
|
|
+ if (StringUtils.isEmpty(deviceId)) {
|
|
|
+ log.warn("储能策略配置缺少deviceId, attr[{}]", attr.getAttrKey());
|
|
|
+ detail.append(attr.getAttrKey()).append(":缺少deviceId ");
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ String payloadJson = buildStrategyPayload(strategy);
|
|
|
+ String topic = "/v1/devices/" + deviceId + TOPIC_SUFFIX_COMMAND;
|
|
|
+
|
|
|
+ // 1. 先插入下发日志(与其他子系统一致,留痕优先)
|
|
|
+ callLog = saveCallLog(abilityParam.getObjCode(), abilityParam.getModelCode(),
|
|
|
+ abilityParam.getAbilityKey(), 0, buildLogPayload(topic, payloadJson), null);
|
|
|
+
|
|
|
+ // 2. 下发指令
|
|
|
+ log.info("储能策略下发, 站点[{}], topic[{}], payload[{}]", deviceId, topic, payloadJson);
|
|
|
+ mqttTemplate.sendString(topic, payloadJson, 0, false);
|
|
|
+
|
|
|
+ // 3. 记录日志id,MQTT应答(commandResponse)到达后回写同一条记录
|
|
|
+ if (null != callLog && null != callLog.getId()) {
|
|
|
+ pendingCallLogIds.put(deviceId, callLog.getId());
|
|
|
+ }
|
|
|
+
|
|
|
+ cnt++;
|
|
|
+ detail.append(deviceId).append(' ');
|
|
|
+ }
|
|
|
+ catch (Exception e) {
|
|
|
+ log.error("储能策略下发失败, attr[{}]", attr.getAttrKey(), e);
|
|
|
+ // 失败状态回写到同一条日志记录
|
|
|
+ updateCallLogResult(null != callLog ? callLog.getId() : null, -1, e.getMessage());
|
|
|
+ detail.append(attr.getAttrKey()).append(":失败 ");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return new CallResponse<>(0, "已下发" + cnt + "个站点: " + detail);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 下发全静置策略:24时段全部禁止(报文结构与厂家模板一致,仅无有效时段)
|
|
|
+ *
|
|
|
+ * @param abilityParam 能力参数
|
|
|
+ * @return 调用结果
|
|
|
+ */
|
|
|
+ private CallResponse<Void> applyIdleStrategy(AbilityPayload abilityParam) {
|
|
|
if (CollectionUtils.isEmpty(sites)) {
|
|
|
return new CallResponse<>(-1, "储能站点配置未加载");
|
|
|
}
|
|
|
|
|
|
- String payloadJson = buildStrategyPayload(sites.get(0).getDeviceId(), slots);
|
|
|
- int cnt = 0;
|
|
|
+ // 构造静置策略:无时段内容(报文仅info_1)
|
|
|
+ JSONObject idleStrategy = new JSONObject();
|
|
|
+ idleStrategy.put("sysFlag", STRATEGY_SYS_FLAG_DEFAULT);
|
|
|
+ idleStrategy.put("slots", new JSONArray());
|
|
|
+ String payloadJson = buildStrategyPayload(idleStrategy);
|
|
|
|
|
|
- // 逐站点下发(报文仅deviceId不同,其余一致)
|
|
|
+ int cnt = 0;
|
|
|
for (GccSite site : sites) {
|
|
|
if (StringUtils.isEmpty(site.getDeviceId())) {
|
|
|
continue;
|
|
|
}
|
|
|
|
|
|
String topic = "/v1/devices/" + site.getDeviceId() + TOPIC_SUFFIX_COMMAND;
|
|
|
- String msg = buildStrategyPayload(site.getDeviceId(), slots);
|
|
|
- log.info("储能策略下发, 能力[{}], 站点[{}], topic[{}], payload[{}]", abilityKey, site.getSiteName(),
|
|
|
- topic, msg);
|
|
|
- mqttTemplate.sendString(topic, msg, 0, false);
|
|
|
- cnt++;
|
|
|
+ EmsObjAbilityCallLog callLog = null;
|
|
|
+
|
|
|
+ try {
|
|
|
+ // 1. 先插入下发日志(留痕优先)
|
|
|
+ callLog = saveCallLog(abilityParam.getObjCode(), abilityParam.getModelCode(),
|
|
|
+ abilityParam.getAbilityKey(), 0, buildLogPayload(topic, payloadJson), null);
|
|
|
+
|
|
|
+ // 2. 下发指令
|
|
|
+ log.info("储能全静置策略下发, 站点[{}], topic[{}], payload[{}]", site.getSiteName(), topic, payloadJson);
|
|
|
+ mqttTemplate.sendString(topic, payloadJson, 0, false);
|
|
|
+
|
|
|
+ // 3. 记录日志id,应答到达后回写同一条记录
|
|
|
+ if (null != callLog && null != callLog.getId()) {
|
|
|
+ pendingCallLogIds.put(site.getDeviceId(), callLog.getId());
|
|
|
+ }
|
|
|
+ cnt++;
|
|
|
+ }
|
|
|
+ catch (Exception e) {
|
|
|
+ log.error("储能全静置策略下发失败, 站点[{}]", site.getSiteName(), e);
|
|
|
+ updateCallLogResult(null != callLog ? callLog.getId() : null, -1, e.getMessage());
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- // 记录能力调用日志(应答异步到达,仅记录下发结果)
|
|
|
- saveCallLog(abilityParam.getObjCode(), abilityParam.getModelCode(), abilityKey, 0, payloadJson,
|
|
|
- "已下发" + cnt + "个站点,等待EMS应答");
|
|
|
- return new CallResponse<>(0, "策略已下发" + cnt + "个站点,等待EMS应答");
|
|
|
+ return new CallResponse<>(0, "全静置策略已下发" + cnt + "个站点,等待EMS应答");
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 构建策略报文(平台下行sheet:serviceCode=strategy, code=strategy01, 24时段×5字段)
|
|
|
- * 时间编码:小时×256+分钟;未定义时段峰谷选择=0(禁止)
|
|
|
+ * 构建策略报文(厂家模板:code=strategy1, deviceid小写, info_1=系统标志位, 每时段5字段)
|
|
|
+ * 时间编码:小时*256+分钟;时段字段:峰谷选择/开始/结束/充电最大功率/放电最大功率
|
|
|
*
|
|
|
- * @param deviceId 站点客户端ID
|
|
|
- * @param slots 有效时段定义(最多24组)
|
|
|
+ * @param strategy 策略JSON(deviceId/sysFlag/slots)
|
|
|
* @return 报文JSON字符串
|
|
|
*/
|
|
|
- private String buildStrategyPayload(String deviceId, int[][] slots) {
|
|
|
+ private String buildStrategyPayload(JSONObject strategy) {
|
|
|
+ Integer sysFlag = strategy.getInteger("sysFlag");
|
|
|
+ JSONArray slots = strategy.getJSONArray("slots");
|
|
|
+
|
|
|
JSONObject data = new JSONObject();
|
|
|
- data.put("code", "strategy01");
|
|
|
- // info_1 保留
|
|
|
- data.put("info_1", 0);
|
|
|
-
|
|
|
- for (int n = 1; n <= 24; n++) {
|
|
|
- int base = 2 + (n - 1) * 5;
|
|
|
- int[] slot = (n <= slots.length) ? slots[n - 1] : null;
|
|
|
-
|
|
|
- if (null == slot) {
|
|
|
- // 时段禁止:全部填0
|
|
|
- for (int i = 0; i < 5; i++) {
|
|
|
- data.put("info_" + (base + i), 0);
|
|
|
- }
|
|
|
- }
|
|
|
- else {
|
|
|
- data.put("info_" + base, slot[0]);
|
|
|
- data.put("info_" + (base + 1), slot[1] * 256 + slot[2]);
|
|
|
- data.put("info_" + (base + 2), slot[3] * 256 + slot[4]);
|
|
|
- data.put("info_" + (base + 3), slot[5]);
|
|
|
- data.put("info_" + (base + 4), slot[6]);
|
|
|
+ data.put("code", STRATEGY_CODE);
|
|
|
+ data.put("info_1", null == sysFlag ? STRATEGY_SYS_FLAG_DEFAULT : sysFlag);
|
|
|
+
|
|
|
+ int idx = 2;
|
|
|
+ if (null != slots) {
|
|
|
+ for (int i = 0; i < slots.size(); i++) {
|
|
|
+ JSONObject slot = slots.getJSONObject(i);
|
|
|
+ data.put("info_" + idx++, slot.getIntValue("peakValley"));
|
|
|
+ data.put("info_" + idx++, encodeTime(slot.getString("start")));
|
|
|
+ data.put("info_" + idx++, encodeTime(slot.getString("end")));
|
|
|
+ data.put("info_" + idx++, slot.getIntValue("chargeKw"));
|
|
|
+ data.put("info_" + idx++, slot.getIntValue("dischargeKw"));
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -1125,7 +1215,7 @@ public class GccHandler extends BaseDevHandler implements MqttMessageHandler {
|
|
|
service.put("data", data);
|
|
|
|
|
|
JSONObject device = new JSONObject();
|
|
|
- device.put("deviceId", deviceId);
|
|
|
+ device.put("deviceid", strategy.getString("deviceId"));
|
|
|
device.put("service", service);
|
|
|
|
|
|
JSONObject root = new JSONObject();
|
|
|
@@ -1133,6 +1223,92 @@ public class GccHandler extends BaseDevHandler implements MqttMessageHandler {
|
|
|
return root.toJSONString();
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 时间编码:HH:mm -> 小时*256+分钟(如 06:00 -> 1536)
|
|
|
+ *
|
|
|
+ * @param time HH:mm格式时间
|
|
|
+ * @return 编码值
|
|
|
+ */
|
|
|
+ private int encodeTime(String time) {
|
|
|
+ String[] parts = StringUtils.split(time, ':');
|
|
|
+ if (null == parts || parts.length < 2) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ return Integer.parseInt(parts[0]) * 256 + Integer.parseInt(parts[1]);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 处理策略下发应答:记日志并将应答(含topic与payload)回写对应调用日志
|
|
|
+ *
|
|
|
+ * @param topic 应答主题
|
|
|
+ * @param payload 应答报文
|
|
|
+ */
|
|
|
+ private void handleCommandResponse(String topic, byte[] payload) {
|
|
|
+ String payloadStr = new String(payload, StandardCharsets.UTF_8);
|
|
|
+ log.info("储能策略下发应答, topic[{}], payload[{}]", topic, payloadStr);
|
|
|
+
|
|
|
+ try {
|
|
|
+ JSONObject root = JSON.parseObject(payloadStr);
|
|
|
+ JSONObject device = root.getJSONObject("device");
|
|
|
+ if (null == device) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 兼容厂家大小写:deviceid / deviceId
|
|
|
+ String deviceId = device.getString("deviceid");
|
|
|
+ if (StringUtils.isEmpty(deviceId)) {
|
|
|
+ deviceId = device.getString("deviceId");
|
|
|
+ }
|
|
|
+
|
|
|
+ if (StringUtils.isEmpty(deviceId)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long logId = pendingCallLogIds.remove(deviceId);
|
|
|
+ updateCallLogResult(logId, 0, buildLogPayload(topic, payloadStr));
|
|
|
+ }
|
|
|
+ catch (Exception e) {
|
|
|
+ log.warn("储能策略应答解析失败, topic[{}]", topic, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 构建日志载体内容(纯字符串):Topic 换行 MsgBody,便于日志详情直接展示
|
|
|
+ *
|
|
|
+ * @param topic MQTT主题
|
|
|
+ * @param msgBody 报文字符串
|
|
|
+ * @return 日志载体字符串
|
|
|
+ */
|
|
|
+ private String buildLogPayload(String topic, String msgBody) {
|
|
|
+ return "Topic: " + topic + "\nMsgBody: " + msgBody;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 回写调用日志(同一条记录):状态/响应时间/响应内容;日志不存在时忽略,不影响主流程
|
|
|
+ *
|
|
|
+ * @param logId 日志主键
|
|
|
+ * @param callStatus 调用状态 0-成功 -1-失败
|
|
|
+ * @param resPayload 响应内容
|
|
|
+ */
|
|
|
+ private void updateCallLogResult(Long logId, int callStatus, String resPayload) {
|
|
|
+ if (null == logId) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ EmsObjAbilityCallLog update = new EmsObjAbilityCallLog();
|
|
|
+ update.setId(logId);
|
|
|
+ update.setCallStatus(callStatus);
|
|
|
+ update.setResTime(new Date());
|
|
|
+ update.setResPayload(resPayload);
|
|
|
+ objAbilityCallLogService.updateLog(update);
|
|
|
+ }
|
|
|
+ catch (Exception e) {
|
|
|
+ log.warn("回写能力调用日志失败, logId[{}]", logId, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
@Override
|
|
|
public List<EmsDevice> getDeviceList() {
|
|
|
QueryDevice queryDevice = new QueryDevice();
|