From fd15e6f7dd6a2f6bccf27a8b06fc60211e7df28e Mon Sep 17 00:00:00 2001
From: puzhibing <393733352@qq.com>
Date: 星期一, 28 九月 2026 09:05:49 +0800
Subject: [PATCH] Merge branch '2.3' of http://120.76.84.145:10101/gitblit/r/java/mx_charging_pile into 2.3
---
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/service/UdanEmsMessageProcessor.java | 140 +++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 137 insertions(+), 3 deletions(-)
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/service/UdanEmsMessageProcessor.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/service/UdanEmsMessageProcessor.java
index d305f4d..08ea393 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/service/UdanEmsMessageProcessor.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/service/UdanEmsMessageProcessor.java
@@ -7,20 +7,29 @@
import com.ruoyi.other.emind.dto.EmindIncomeHistoryDto;
import com.ruoyi.other.emind.mapper.EmindStationMapper;
import com.ruoyi.other.emind.mapper.EmindStorageDashboardMapper;
+import com.ruoyi.other.emind.service.EmindEnergyChainDetailService;
import com.ruoyi.other.udan.model.EmsState;
+import com.ruoyi.other.udan.model.BatteryStackState;
+import com.ruoyi.other.udan.model.ElectricMeterState;
+import com.ruoyi.other.udan.model.PcsState;
import com.ruoyi.other.udan.model.PowerMeterState;
+import com.ruoyi.other.udan.model.TmsState;
import com.ruoyi.other.udan.model.UdanEmsDeviceState;
import com.ruoyi.other.udan.util.UdanEmsKafkaUtil;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.io.IOException;
+import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
/** 将 IEmsAllDeviceState 中模型已有的 EMS 和能耗计字段转换为业务对象。 */
@Slf4j
@@ -30,11 +39,13 @@
private final UdanEmsKafkaUtil kafkaUtil;
private final EmindStorageDashboardMapper dashboardMapper;
private final EmindStationMapper stationMapper;
+ private final EmindEnergyChainDetailService emindEnergyChainDetailService;
- public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil, EmindStorageDashboardMapper dashboardMapper, EmindStationMapper stationMapper) {
+ public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil, EmindStorageDashboardMapper dashboardMapper, EmindStationMapper stationMapper, EmindEnergyChainDetailService emindEnergyChainDetailService) {
this.kafkaUtil = kafkaUtil;
this.dashboardMapper = dashboardMapper;
this.stationMapper = stationMapper;
+ this.emindEnergyChainDetailService = emindEnergyChainDetailService;
}
@PostConstruct
@@ -52,10 +63,19 @@
EmindStation station = stations.stream().filter(s -> udanEmsDeviceState.getEmsDeviceId().equals(s.getStationCode())).findFirst().orElse(null);
if(Objects.nonNull(station)) {
List<EmindIncomeHistoryDto> daily = udanEmsDeviceState.convert(station.getStationId());
- dashboardMapper.upsertDaily(station.getStationId(), daily, now);
+ Set<LocalDate> dateSet = daily.stream().map(EmindIncomeHistoryDto::getDate).collect(Collectors.toSet());
+ // 删除电站下该天已存在的旧数据
+ if(CollectionUtils.isNotEmpty(dateSet)) {
+ dashboardMapper.deleteByStationIdDay(station.getStationId(), dateSet);
+ }
+ if(CollectionUtils.isNotEmpty(daily)) {
+ dashboardMapper.upsertDaily(station.getStationId(), daily, now);
+ }
}else{
- log.info("UDAN EMS数据同步错误-电站信息不存在！");
+ log.error("UDAN EMS数据同步错误-电站信息不存在！");
}
+ // 保存设备其它信息
+ emindEnergyChainDetailService.saveAll(udanEmsDeviceState);
}
} catch (IOException e) {
@@ -76,7 +96,19 @@
read(bytes, (field, type, in) -> {
if (field == 1 && type == 0) state.setTime(in.readUInt64());
else if (type == 2 && field == 2) state.setEmsState(parseEms(in.readBytes().toByteArray()));
+ else if (type == 2 && field == 4 && state.getPcsState() == null) {
+ state.setPcsState(parsePcs(in.readBytes().toByteArray()));
+ }
+ else if (type == 2 && field == 5 && state.getBatteryStackState() == null) {
+ state.setBatteryStackState(parseBatteryStack(in.readBytes().toByteArray()));
+ }
+ else if (type == 2 && field == 7 && state.getElectricMeterState() == null) {
+ state.setElectricMeterState(parseElectricMeter(in.readBytes().toByteArray()));
+ }
else if (type == 2 && field == 8) state.setPowerMeterState(parsePowerMeter(in.readBytes().toByteArray()));
+ else if (type == 2 && field == 14 && state.getTmsState() == null) {
+ state.setTmsState(parseTms(in.readBytes().toByteArray()));
+ }
else skip(in, type);
});
return state;
@@ -189,6 +221,96 @@
return value;
}
+ private PcsState parsePcs(byte[] bytes) throws IOException {
+ PcsState value = new PcsState();
+ read(bytes, (field, type, input) -> {
+ switch (field) {
+ case 2: value.setDeviceId(str(input, type)); break;
+ case 3: value.setErrorStatus(enumName(input, type, "ErrorStatus")); break;
+ case 4: value.setAlarmStatus(enumName(input, type, "AlarmStatus")); break;
+ case 5: value.setRunStatus(enumName(input, type, "RunStatus")); break;
+ case 6: value.setOnGridStatus(enumName(input, type, "GridStatus")); break;
+ case 13: value.setAcTotalActivePower((double) f32(input, type)); break;
+ case 17: value.setDailyChargeEnergyKwh((double) f32(input, type)); break;
+ case 18: value.setDailyDischargeEnergyKwh((double) f32(input, type)); break;
+ case 19: value.setTotalChargeEnergyKwh(f64(input, type)); break;
+ case 20: value.setTotalDischargeEnergyKwh(f64(input, type)); break;
+ case 31: value.setAPhaseActivePower((double) f32(input, type)); break;
+ case 34: value.setBPhaseActivePower((double) f32(input, type)); break;
+ case 37: value.setCPhaseActivePower((double) f32(input, type)); break;
+ default: skip(input, type);
+ }
+ });
+ return value;
+ }
+
+ private BatteryStackState parseBatteryStack(byte[] bytes) throws IOException {
+ BatteryStackState value = new BatteryStackState();
+ read(bytes, (field, type, input) -> {
+ switch (field) {
+ case 2: value.setDeviceId(str(input, type)); break;
+ case 4: value.setWorkStatus(enumName(input, type, "BatteryWorkStatus")); break;
+ case 5: value.setErrorStatus(enumName(input, type, "ErrorStatus")); break;
+ case 6: value.setWarningStatus(enumName(input, type, "AlarmStatus")); break;
+ case 7: value.setTv((double) f32(input, type)); break;
+ case 8: value.setTc((double) f32(input, type)); break;
+ case 9: value.setSoh((double) f32(input, type)); break;
+ case 10: value.setSoc((double) f32(input, type)); break;
+ case 15: value.setMinCellVoltMv((double) u32(input, type)); break;
+ case 23: value.setMaxCellVoltMv((double) u32(input, type)); break;
+ case 31: value.setRechargeableCapacityKwh((double) f32(input, type)); break;
+ case 32: value.setDischargeableCapacityKwh((double) f32(input, type)); break;
+ case 33: value.setCumulativeDischargeKwh(f64(input, type)); break;
+ case 34: value.setCumulativeChargeKwh(f64(input, type)); break;
+ case 36: value.setAvgCellTemp((double) u32(input, type) - 40D); break;
+ default: skip(input, type);
+ }
+ });
+ return value;
+ }
+
+ private ElectricMeterState parseElectricMeter(byte[] bytes) throws IOException {
+ ElectricMeterState value = new ElectricMeterState();
+ read(bytes, (field, type, input) -> {
+ switch (field) {
+ case 2: value.setDeviceId(str(input, type)); break;
+ case 48: value.setAPhaseVolt((double) f32(input, type)); break;
+ case 49: value.setBPhaseVolt((double) f32(input, type)); break;
+ case 50: value.setCPhaseVolt((double) f32(input, type)); break;
+ case 51: value.setAPhaseCurrent((double) f32(input, type)); break;
+ case 52: value.setBPhaseCurrent((double) f32(input, type)); break;
+ case 53: value.setCPhaseCurrent((double) f32(input, type)); break;
+ case 54: value.setAcTotalActivePower((double) f32(input, type)); break;
+ case 69: value.setFrequency((double) f32(input, type)); break;
+ case 74: value.setAcTotalPowerFactor((double) f32(input, type)); break;
+ default: skip(input, type);
+ }
+ });
+ return value;
+ }
+
+ private TmsState parseTms(byte[] bytes) throws IOException {
+ TmsState value = new TmsState();
+ read(bytes, (field, type, input) -> {
+ switch (field) {
+ case 1: value.setDeviceId(str(input, type)); break;
+ case 2: value.setRun_status(enumName(input, type, "TmsRunStatus")); break;
+ case 3: value.setError_status(enumName(input, type, "ErrorStatus")); break;
+ case 4: value.setSupply_temp((double) f32(input, type)); break;
+ case 5: value.setReturn_temp((double) f32(input, type)); break;
+ case 8: value.setEnv_temp((double) f32(input, type)); break;
+ case 9: value.setBackwater_press((double) f32(input, type)); break;
+ case 10: value.setOut_water_press((double) f32(input, type)); break;
+ case 13: value.setCompress_status(enumName(input, type, "RunStatus")); break;
+ case 14: value.setCompress_heat_status(enumName(input, type, "RunStatus")); break;
+ case 15: value.setElectric_heat_status(enumName(input, type, "RunStatus")); break;
+ case 16: value.setPump_status(enumName(input, type, "RunStatus")); break;
+ default: skip(input, type);
+ }
+ });
+ return value;
+ }
+
private void read(byte[] bytes, Reader reader) throws IOException {
CodedInputStream input = CodedInputStream.newInstance(bytes == null ? new byte[0] : bytes);
while (!input.isAtEnd()) {
@@ -250,6 +372,18 @@
if ("CommStatus".equals(prefix)) {
return new String[]{"COMM_STATUS_UNKNOWN", "COMM_STATUS_DISCONNECTED", "COMM_STATUS_CONNECTED"}[number < 0 || number > 2 ? 0 : number];
}
+ if ("ErrorStatus".equals(prefix)) {
+ return new String[]{"ERROR_STATUS_UNKNOWN", "ERROR_STATUS_OCCURRING", "ERROR_STATUS_NONE"}[number < 0 || number > 2 ? 0 : number];
+ }
+ if ("AlarmStatus".equals(prefix)) {
+ return new String[]{"ALARM_STATUS_UNKNOWN", "ALARM_STATUS_OCCURRING", "ALARM_STATUS_NONE"}[number < 0 || number > 2 ? 0 : number];
+ }
+ if ("BatteryWorkStatus".equals(prefix)) {
+ return new String[]{"BATTERY_WORK_STATUS_UNKNOWN", "BATTERY_WORK_STATUS_CHARGING", "BATTERY_WORK_STATUS_DISCHARGING", "BATTERY_WORK_STATUS_IDLE"}[number < 0 || number > 3 ? 0 : number];
+ }
+ if ("TmsRunStatus".equals(prefix)) {
+ return new String[]{"TMS_RUN_STATUS_UNKNOWN", "TMS_RUN_STATUS_IDLE", "TMS_RUN_STATUS_READY_TO_START", "TMS_RUN_STATUS_RUNNING", "TMS_RUN_STATUS_DELAYED_SHUTDOWN", "TMS_RUN_STATUS_SELF_TEST", "TMS_RUN_STATUS_ALARMING"}[number < 0 || number > 6 ? 0 : number];
+ }
return prefix + "_" + number;
}
private static <T> T enumAt(T[] values, CodedInputStream input, int type) throws IOException {
--
Gitblit v1.7.1