From 868b19dddc1a28cdcf2703dac7a07df4b0b74221 Mon Sep 17 00:00:00 2001
From: yanderong <307678080@qq.com>
Date: 星期二, 15 九月 2026 11:40:22 +0800
Subject: [PATCH] feat: 储能数据同步，新运维大楼大屏的配置
---
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/EmsState.java | 488 ++++++++++++++++++++++++++++
/dev/null | 26 -
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDeviceState.java | 69 ++-
ruoyi-service/ruoyi-other/src/main/resources/sql/t_screen_content_alter.sql | 6
ruoyi-service/ruoyi-other/src/main/resources/mapper/other/EmindStationMapper.xml | 2
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/service/EmindStationManagementService.java | 4
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/PowerMeterState.java | 7
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/util/UdanEmsKafkaUtil.java | 39 +
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/config/UdanEmsKafkaProperties.java | 2
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/domain/EmindStation.java | 4
ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/service/UdanEmsMessageProcessor.java | 319 ++++++++++++++----
11 files changed, 822 insertions(+), 144 deletions(-)
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/domain/EmindStation.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/domain/EmindStation.java
index becfb3d..f65b128 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/domain/EmindStation.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/domain/EmindStation.java
@@ -85,4 +85,8 @@
/** Last update time. */
@ApiModelProperty("Last update time")
private LocalDateTime updateTime;
+ /**
+ * 电站设备编号
+ */
+ private String stationCode;
}
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/service/EmindStationManagementService.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/service/EmindStationManagementService.java
index 0dbeda7..95f124e 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/service/EmindStationManagementService.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/emind/service/EmindStationManagementService.java
@@ -48,13 +48,14 @@
station.setMaxPower(input.getMaxPower());
station.setCapacity(input.getCapacity());
station.setCommissioningTime(input.getCommissioningTime());
- station.setPlatformName(EmindStation.DEFAULT_PLATFORM_NAME);
+ station.setPlatformName(input.getPlatformName());
station.setSourceType(EmindStation.SOURCE_MANUAL);
station.setManagementDeleted(0);
station.setDelFlag(0);
station.setSyncedAt(now);
station.setCreateTime(now);
station.setUpdateTime(now);
+ station.setStationCode(input.getStationCode());
mapper.insert(station);
return station;
}
@@ -80,6 +81,7 @@
station.setCapacity(input.getCapacity());
station.setCommissioningTime(input.getCommissioningTime());
station.setUpdateTime(LocalDateTime.now(clock));
+ station.setStationCode(input.getStationCode());
mapper.updateById(station);
return station;
}
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/config/UdanEmsKafkaProperties.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/config/UdanEmsKafkaProperties.java
index 9a26c50..439c541 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/config/UdanEmsKafkaProperties.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/config/UdanEmsKafkaProperties.java
@@ -16,5 +16,5 @@
private String topic;
private String groupId = "group-device-state";
private String autoOffsetReset = "latest";
- private int pollSeconds = 3600;
+ private int pollSeconds = 3600; // 1个小时
}
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/EmsState.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/EmsState.java
new file mode 100644
index 0000000..49adeaa
--- /dev/null
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/EmsState.java
@@ -0,0 +1,488 @@
+package com.ruoyi.other.udan.model;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 优旦储能-设备
+ * @author Lux
+ * @version 1.0.0
+ * @since 2026/9/14 17:59
+ */
+@Data
+public class EmsState implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 设备id （可关联系统配置的电站设备号）
+ * 只能由字母、数字和下划线组成，长度不超过50个字节
+ */
+ private String deviceId;
+
+ /**
+ * 系统状态
+ */
+ private String emsStatus;
+
+ /**
+ * 需量
+ */
+ private Float demandPower;
+
+ /**
+ * 4G卡
+ */
+ private String emsSimCard;
+
+ /**
+ * 时间
+ * 1970-01-01T00:00:00Z 以来的毫秒数
+ */
+ private Long time;
+
+ /**
+ * proto协议版本
+ */
+ private String protoProtocolVersion;
+
+ /**
+ * 运行状态
+ */
+ private EmsRunStatus emsRunStatus;
+
+ /**
+ * 工作状态
+ */
+ private EmsWorkStatus emsWorkStatus;
+
+ /**
+ * 故障状态
+ */
+ private ErrorStatus faultStatus;
+
+ /**
+ * 并离网状态
+ */
+ private String onGridStatus;
+
+ /**
+ * 控制源
+ */
+ private ControlSrc controlSrc;
+
+ /**
+ * 指令模式
+ */
+ private EmuModeCode emuModeCode;
+
+ /**
+ * 计划模式
+ */
+ private EmuAutoMode emuAutoMod;
+
+ /**
+ * 防逆流控制
+ */
+ private YesNo overDischargeProtect;
+
+ /**
+ * 防过载控制
+ */
+ private YesNo overChargeProtect;
+
+ /**
+ * 需量控制
+ */
+ private YesNo demandControl;
+
+ /**
+ * 动态增容控制
+ */
+ private YesNo capacityExpansionModeCtrl;
+
+ /**
+ * 日充电量
+ * 单位：kWh
+ */
+ private Float dailyChargeEnergyKwh;
+
+ /**
+ * 日放电量
+ * 单位：kWh
+ */
+ private Float dailyDischargeEnergyKwh;
+
+ /**
+ * 月充电量
+ * 单位：kWh
+ */
+ private Float monthlyChargeEnergyKwh;
+
+ /**
+ * 月放电量
+ * 单位：kWh
+ */
+ private Float monthlyDischargeEnergyKwh;
+
+ /**
+ * 累计充电量
+ * 单位：kWh
+ */
+ private Double totalChargeEnergyKwh;
+
+ /**
+ * 累计放电量
+ * 单位：kWh
+ */
+ private Double totalDischargeEnergyKwh;
+
+ /**
+ * PCS总有功功率
+ * 单位：kW
+ */
+ private Float pcsSumsActivePower;
+
+ /**
+ * PCS总视在功率
+ * 单位：kVar
+ */
+ private Float pcsSumsApparentPower;
+
+ /**
+ * 直流母线总电压
+ * 单位：V
+ */
+ private Float dcBusVoltage;
+
+ /**
+ * 直流母线总电流
+ * 单位：A
+ */
+ private Float dcBusCurrent;
+
+ /**
+ * 总SOC
+ * 单位：%
+ */
+ private Float sumsSoc;
+
+ /**
+ * 电池总节数
+ */
+ private Integer cellSums;
+
+ /**
+ * 电池平均温度
+ * 单位：°C
+ */
+ private Float batteryAverageTempr;
+
+ /**
+ * 电池最大温差
+ * 单位：°C
+ */
+ private Float batteryMaxTemprDiff;
+
+ /**
+ * 最高单体电压值
+ * 单位：mV
+ */
+ private Integer maxCellVoltage;
+
+ /**
+ * 最低单体电压值
+ * 单位：mV
+ */
+ private Integer minCellVoltage;
+
+ /**
+ * 系统最大允许充电功率
+ * 单位：kW
+ */
+ private Float systemMaxAllowChgPower;
+
+ /**
+ * 系统最大允许放电功率
+ * 单位：kW
+ */
+ private Float systemMaxAllowDisPower;
+
+ /**
+ * 可充电量
+ * 单位：kWh
+ */
+ private Float rechargeableKwh;
+
+ /**
+ * 可放电量
+ * 单位：kWh
+ */
+ private Float dischargeableKwh;
+
+ /**
+ * 当前时刻最大允许充电功率（汇总）
+ * 单位：kW
+ */
+ private Float currentMaxAllowChgPower;
+
+ /**
+ * 当前时刻最大允许放电功率（汇总）
+ * 单位：kW
+ */
+ private Float currentMaxAllowDisPower;
+
+ /**
+ * SOC上限
+ * 单位：%
+ */
+ private Float socUpperLimit;
+
+ /**
+ * SOC下限
+ * 单位：%
+ */
+ private Float socLowerLimit;
+
+ /**
+ * 最大功率放电时长
+ * 单位：s
+ */
+ private Integer maxPowerDischargeDuration;
+
+ /**
+ * 最大功率充电时长
+ * 单位：s
+ */
+ private Integer maxPowerChargeDuration;
+
+ /**
+ * 与主机通讯状态
+ */
+ private String masterCommStatus;
+
+ /**
+ * 并网点正向有功电能
+ * 单位：kWh
+ */
+ private Double gridImportActiveEnergyKwh;
+
+ /**
+ * 并网点反向有功电能
+ * 单位：kWh
+ */
+ private Double gridExportActiveEnergyKwh;
+
+ /**
+ * 直流侧总功率
+ * 单位：kW
+ */
+ private Float dcTotalPower;
+
+ /**
+ * 负载功率
+ * 正放负充
+ * 单位：kW
+ */
+ private Float loadTotalPower;
+
+ /**
+ * 储能侧总功率
+ * 正放负充
+ * 单位：kW
+ */
+ private Float energyStoreTotalPower;
+
+ /**
+ * 高压侧电网功率
+ * 正放负充
+ * 单位：kW
+ */
+ private Float highSideGridPower;
+
+ /**
+ * 低压侧电网功率
+ * 正放负充
+ * 单位：kW
+ */
+ private Float lowSideGridPower;
+
+ /**
+ * 光伏功率
+ * 正放负充
+ * 单位：kW
+ */
+ private Float pvPower;
+
+ /**
+ * 总SOH
+ * 单位：%
+ */
+ private Float sumsSoh;
+
+ public enum ErrorStatus {
+ ERROR_STATUS_UNKNOWN(0), ERROR_STATUS_OCCURRING(1), ERROR_STATUS_NONE(2);
+ private final int code;
+ ErrorStatus(int code) { this.code = code; }
+ public int getCode() { return code; }
+ }
+
+
+ /**
+ * EMS运行状态
+ */
+ public enum EmsRunStatus {
+
+ /**
+ * 预留
+ */
+ EMS_RUN_STATUS_UNKNOWN(0),
+
+ /**
+ * 停止
+ */
+ EMS_RUN_STATUS_STOPPED(1),
+
+ /**
+ * 待机
+ */
+ EMS_RUN_STATUS_IDLE(2),
+
+ /**
+ * 运行
+ */
+ EMS_RUN_STATUS_RUNNING(3),
+
+ /**
+ * 测试
+ */
+ EMS_RUN_STATUS_TEST(4);
+
+ private final int code;
+
+ EmsRunStatus(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+ }
+
+ /**
+ * EMS工作状态
+ */
+ public enum EmsWorkStatus {
+
+ EMS_WORK_STATUS_UNKNOWN(0),
+
+ EMS_WORK_STATUS_CHARGING(1),
+
+ EMS_WORK_STATUS_DISCHARGING(2),
+
+ EMS_WORK_STATUS_IDLE(3);
+
+ private final int code;
+
+ EmsWorkStatus(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+ }
+
+ /**
+ * 控制源
+ */
+ public enum ControlSrc {
+
+ CONTROL_SRC_UNKNOWN(0),
+
+ CONTROL_SRC_LOCAL(1),
+
+ CONTROL_SRC_REMOTE(2),
+
+ CONTROL_SRC_OTHER(3);
+
+ private final int code;
+
+ ControlSrc(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+ }
+
+ /**
+ * 指令模式
+ */
+ public enum EmuModeCode {
+
+ EMU_MODE_CODE_UNKNOWN(0),
+
+ EMU_MODE_CODE_SUB_SYSTEM(1),
+
+ EMU_MODE_CODE_PLAN(2),
+
+ EMU_MODE_CODE_GENERAL_INSTRUCTION(3);
+
+ private final int code;
+
+ EmuModeCode(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+ }
+
+ /**
+ * 计划模式
+ */
+ public enum EmuAutoMode {
+
+ EMU_AUTO_MODE_UNKNOWN(0),
+
+ EMU_AUTO_MODE_24_HOUR_SCHEDULED_MODE(1),
+
+ EMU_AUTO_MODE_DEMAND_RESPONSE_SCHEDULED_MODE(2);
+
+ private final int code;
+
+ EmuAutoMode(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+ }
+
+ /**
+ * 保留
+ */
+ public enum YesNo {
+
+ YES_NO_UNKNOWN(0), // 保留
+
+ YES_NO_YES(1), // 是
+
+ YES_NO_NO(2); // 否
+
+ private final int code;
+
+ YesNo(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+ }
+}
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/PowerMeterState.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/PowerMeterState.java
index 707d629..4c5f63a 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/PowerMeterState.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/PowerMeterState.java
@@ -2,13 +2,18 @@
import lombok.Data;
+import java.io.Serializable;
+
/**
* @author Lux
* @version 1.0.0
* @since 2026/9/10 17:55
*/
@Data
-public class PowerMeterState {
+public class PowerMeterState implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
/** 峰用电总价，单位：元 */
private double cumulativeImportAmountPeak;
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDatabaseRecord.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDatabaseRecord.java
deleted file mode 100644
index f3fa9eb..0000000
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDatabaseRecord.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package com.ruoyi.other.udan.model;
-
-import lombok.Data;
-
-import java.time.Instant;
-
-/** 待写入数据库的业务对象；仅定义映射结果，不执行数据库操作。 */
-@Data
-public class UdanEmsDatabaseRecord {
- private String emsSystemId;
- private Instant eventTime;
- private int pcsCount;
- private int batteryStackCount;
- private int batteryClusterCount;
- private int electricMeterCount;
- private int powerMeterCount;
- private int airConditionerCount;
- private int fireCount;
- private int jpfCount;
- private int tmsCount;
- private int dehumidifierCount;
- private int upsCount;
- private int pvCount;
- private int chargeStationCount;
- private int extPeripheralCount;
-}
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDeviceState.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDeviceState.java
index 5c382c7..2119556 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDeviceState.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDeviceState.java
@@ -1,35 +1,60 @@
package com.ruoyi.other.udan.model;
+import com.ruoyi.other.emind.dto.EmindIncomeHistoryDto;
import lombok.Data;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
/** iems-open-v1.0.5.proto 中 IEmsAllDeviceState 对应的业务对象。 */
@Data
public class UdanEmsDeviceState {
private String emsSystemId;
private long time;
- private EmsState ems;
- private List<PcsState> pcsStates = new ArrayList<>();
- private List<BatteryStackState> batteryStackStates = new ArrayList<>();
- private List<BatteryClusterState> batteryClusterStates = new ArrayList<>();
- private List<ElectricMeterState> electricMeterStates = new ArrayList<>();
- private List<PvState> pvStates = new ArrayList<>();
- private List<ChargeStationState> chargeStationStates = new ArrayList<>();
- private int powerMeterCount;
- private int airConditionerCount;
- private int fireCount;
- private int jpfCount;
- private int tmsCount;
- private int dehumidifierCount;
- private int upsCount;
- private int extPeripheralCount;
+ private String emsDeviceId;
+ private EmsState emsState;
+ private PowerMeterState powerMeterState;
- @Data public static class EmsState { private String deviceId; private int emsStatus; private float demandPower; private long time; private String protoProtocolVersion; private int emsRunStatus; private int emsWorkStatus; private int faultStatus; private int onGridStatus; private float dailyChargeEnergyKwh; private float dailyDischargeEnergyKwh; private float monthlyChargeEnergyKwh; private float monthlyDischargeEnergyKwh; private double totalChargeEnergyKwh; private double totalDischargeEnergyKwh; private float pcsSumsActivePower; private float sumsSoc; private float sumsSoh; private float dcBusVoltage; private float dcBusCurrent; }
- @Data public static class PcsState { private long time; private String deviceId; private int errorStatus; private int alarmStatus; private int runStatus; private int onGridStatus; private float acTotalActivePower; private float acTotalReactivePower; private float acTotalApparentPower; private float acTotalPowerFactor; private List<Float> dcTotalVoltage = new ArrayList<>(); private List<Float> dcCurrent = new ArrayList<>(); private List<Float> dcPower = new ArrayList<>(); private float dailyChargeEnergyKwh; private float dailyDischargeEnergyKwh; private double totalChargeEnergyKwh; private double totalDischargeEnergyKwh; private int controlMode; private int workStatus; private float ambientTemp; private int commStatus; }
- @Data public static class BatteryStackState { private long time; private String deviceId; private int workStatus; private int errorStatus; private float tv; private float tc; private float soh; private float soc; private float rechargeableCapacityKwh; private float dischargeableCapacityKwh; private double cumulativeDischargeKwh; private double cumulativeChargeKwh; private int avgCellVoltageMv; private int avgCellTemp; private int batteryStatus; private int commStatus; }
- @Data public static class BatteryClusterState { private long time; private String deviceId; private String stackDeviceId; private int workStatus; private float tv; private float tc; private float soh; private float soc; private float rechargeableCapacityKwh; private float dischargeableCapacityKwh; private double cumulativeChargeKwh; private double cumulativeDischargeKwh; private int avgCellTemp; private int avgCellVoltMv; private int commStatus; }
- @Data public static class ElectricMeterState { private long time; private String deviceId; private int meterType; private double cumulativeImportKwh; private double cumulativeExportKwh; private float acTotalActivePower; private float aPhaseActivePower; private float bPhaseActivePower; private float cPhaseActivePower; private int commStatus; }
- @Data public static class PvState { private String deviceId; private int commStatus; private float activePower; private float reactivePower; private float powerFactor; private float dailyPowerGeneration; private double totalPowerGeneration; private int runStatus; private int errorStatus; private int alarmStatus; }
- @Data public static class ChargeStationState { private String deviceId; private long time; private int runStatus; private int commStatus; }
+ public List<EmindIncomeHistoryDto> convert(Long stationId) {
+ List<EmindIncomeHistoryDto> dailyList = new ArrayList<>();
+
+ if(Objects.nonNull(this.getPowerMeterState())) {
+ EmindIncomeHistoryDto daily = new EmindIncomeHistoryDto();
+ daily.setStationId(stationId);
+ Optional.of(getPowerMeterState().getTime()).ifPresent(t -> {
+ LocalDate localDate = Instant.ofEpochMilli(t).atZone(ZoneId.systemDefault()).toLocalDate();
+ daily.setDate(localDate);
+ });
+
+ daily.setPrice(getPowerMeterState().getDailyImportAmount());
+ daily.setPriceName("");
+ Optional.of(getPowerMeterState().getDailyProfit()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setIncome);
+ Optional.of(getPowerMeterState().getDailyExportKwhValley()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setDischargedEnergyValley);
+ Optional.of(getPowerMeterState().getDailyExportKwhSharp()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setDischargedEnergySharp);
+ Optional.of(getPowerMeterState().getDailyExportKwhPeak()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setDischargedEnergyPeak);
+ Optional.of(getPowerMeterState().getDailyExportKwhShoulder()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setDischargedEnergyFlat);
+ Optional.of(getPowerMeterState().getDailyExportKwh()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setDischargedEnergy);
+ Optional.of(getPowerMeterState().getDailyImportKwhValley()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setChargedEnergyValley);
+ Optional.of(getPowerMeterState().getDailyImportKwhSharp()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setChargedEnergySharp);
+ Optional.of(getPowerMeterState().getDailyImportKwhPeak()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setChargedEnergyPeak);
+ Optional.of(getPowerMeterState().getDailyImportKwhShoulder()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setChargedEnergyFlat);
+ Optional.of(getPowerMeterState().getDailyImportKwh()).map(String::valueOf).map(BigDecimal::new).ifPresent(daily::setChargedEnergy);
+ dailyList.add(daily);
+ }
+ return dailyList;
+ }
+
+ // @Data public static class EmsState { private String deviceId; private int emsStatus; private float demandPower; private long time; private String protoProtocolVersion; private int emsRunStatus; private int emsWorkStatus; private int faultStatus; private int onGridStatus; private float dailyChargeEnergyKwh; private float dailyDischargeEnergyKwh; private float monthlyChargeEnergyKwh; private float monthlyDischargeEnergyKwh; private double totalChargeEnergyKwh; private double totalDischargeEnergyKwh; private float pcsSumsActivePower; private float sumsSoc; private float sumsSoh; private float dcBusVoltage; private float dcBusCurrent; }
+ // @Data public static class PcsState { private long time; private String deviceId; private int errorStatus; private int alarmStatus; private int runStatus; private int onGridStatus; private float acTotalActivePower; private float acTotalReactivePower; private float acTotalApparentPower; private float acTotalPowerFactor; private List<Float> dcTotalVoltage = new ArrayList<>(); private List<Float> dcCurrent = new ArrayList<>(); private List<Float> dcPower = new ArrayList<>(); private float dailyChargeEnergyKwh; private float dailyDischargeEnergyKwh; private double totalChargeEnergyKwh; private double totalDischargeEnergyKwh; private int controlMode; private int workStatus; private float ambientTemp; private int commStatus; }
+// @Data public static class BatteryStackState { private long time; private String deviceId; private int workStatus; private int errorStatus; private float tv; private float tc; private float soh; private float soc; private float rechargeableCapacityKwh; private float dischargeableCapacityKwh; private double cumulativeDischargeKwh; private double cumulativeChargeKwh; private int avgCellVoltageMv; private int avgCellTemp; private int batteryStatus; private int commStatus; }
+// @Data public static class BatteryClusterState { private long time; private String deviceId; private String stackDeviceId; private int workStatus; private float tv; private float tc; private float soh; private float soc; private float rechargeableCapacityKwh; private float dischargeableCapacityKwh; private double cumulativeChargeKwh; private double cumulativeDischargeKwh; private int avgCellTemp; private int avgCellVoltMv; private int commStatus; }
+// @Data public static class ElectricMeterState { private long time; private String deviceId; private int meterType; private double cumulativeImportKwh; private double cumulativeExportKwh; private float acTotalActivePower; private float aPhaseActivePower; private float bPhaseActivePower; private float cPhaseActivePower; private int commStatus; }
+// @Data public static class PvState { private String deviceId; private int commStatus; private float activePower; private float reactivePower; private float powerFactor; private float dailyPowerGeneration; private double totalPowerGeneration; private int runStatus; private int errorStatus; private int alarmStatus; }
+// @Data public static class ChargeStationState { private String deviceId; private long time; private int runStatus; private int commStatus; }
}
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 95e8088..d305f4d 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
@@ -3,112 +3,281 @@
import com.alibaba.fastjson.JSON;
import com.google.protobuf.CodedInputStream;
import com.google.protobuf.WireFormat;
+import com.ruoyi.other.emind.domain.EmindStation;
+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.udan.model.EmsState;
+import com.ruoyi.other.udan.model.PowerMeterState;
import com.ruoyi.other.udan.model.UdanEmsDeviceState;
import com.ruoyi.other.udan.util.UdanEmsKafkaUtil;
import lombok.extern.slf4j.Slf4j;
+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.LocalDateTime;
+import java.util.List;
+import java.util.Objects;
-/** 将 iems-open-v1.0.5.proto 消息解析为业务对象，并映射为待入库对象。 */
+/** 将 IEmsAllDeviceState 中模型已有的 EMS 和能耗计字段转换为业务对象。 */
@Slf4j
@Component
@ConditionalOnProperty(prefix = "udan.ems.kafka", name = "enabled", havingValue = "true")
public class UdanEmsMessageProcessor {
private final UdanEmsKafkaUtil kafkaUtil;
+ private final EmindStorageDashboardMapper dashboardMapper;
+ private final EmindStationMapper stationMapper;
- /** 注入 Kafka 消费工具。 */
- public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil) { this.kafkaUtil = kafkaUtil; }
+ public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil, EmindStorageDashboardMapper dashboardMapper, EmindStationMapper stationMapper) {
+ this.kafkaUtil = kafkaUtil;
+ this.dashboardMapper = dashboardMapper;
+ this.stationMapper = stationMapper;
+ }
- /** 注册消息回调，避免消费者只消费但没有业务处理。 */
- //@PostConstruct
- public void register() { kafkaUtil.setMessageHandler(this::process); }
+ @PostConstruct
+ public void register() {
+ kafkaUtil.setMessageHandler(this::process);
+ }
- /** 解析消息、转换待入库对象并输出日志；此处暂不调用数据库。 */
- public void process(String emsSystemId, byte[] protobufBytes) {
+ public void process(String id, byte[] bytes, String emsDeviceId) {
try {
- UdanEmsDeviceState state = parse(emsSystemId, protobufBytes);
- log.info("2、UDAN EMS实际对象: {}", JSON.toJSONString(state));
+ LocalDateTime now = LocalDateTime.now();
+ UdanEmsDeviceState udanEmsDeviceState = parse(id, emsDeviceId, bytes);
+ if(StringUtils.isNotEmpty(udanEmsDeviceState.getEmsDeviceId())) {
+ log.info("2、UDAN EMS实际对象: {}", JSON.toJSONString(udanEmsDeviceState));
+ List<EmindStation> stations = stationMapper.selectRemoteActiveStations();
+ 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);
+ }else{
+ log.info("UDAN EMS数据同步错误-电站信息不存在！");
+ }
+ }
+
} catch (IOException e) {
- log.error("UDAN EMS protobuf解析失败: emsSystemId={}, bytes={}", emsSystemId, protobufBytes == null ? 0 : protobufBytes.length, e);
+ log.error("UDAN EMS protobuf解析失败: emsSystemId={}, bytes={}", id,
+ bytes == null ? 0 : bytes.length, e);
}
}
- /** 按顶层 IEmsAllDeviceState 字段号读取并构造真实业务对象。 */
+ /** 兼容不带 Header 的直接调用。 */
+ public void process(String id, byte[] bytes) {
+ process(id, bytes, null);
+ }
+
+ public UdanEmsDeviceState parse(String id, String emsDeviceId, byte[] bytes) throws IOException {
+ UdanEmsDeviceState state = new UdanEmsDeviceState();
+ state.setEmsSystemId(id);
+ state.setEmsDeviceId(emsDeviceId);
+ 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 == 8) state.setPowerMeterState(parsePowerMeter(in.readBytes().toByteArray()));
+ else skip(in, type);
+ });
+ return state;
+ }
+
+ /** 兼容不带 Header 的直接解析调用。 */
public UdanEmsDeviceState parse(String id, byte[] bytes) throws IOException {
- UdanEmsDeviceState s = new UdanEmsDeviceState();
- s.setEmsSystemId(id);
- CodedInputStream in = CodedInputStream.newInstance(bytes == null ? new byte[0] : bytes);
- while (!in.isAtEnd()) {
- int tag = in.readTag(); if (tag == 0) break;
- int field = WireFormat.getTagFieldNumber(tag); int type = WireFormat.getTagWireType(tag);
- if (field == 1 && type == WireFormat.WIRETYPE_VARINT) s.setTime(in.readUInt64());
- else if (type == WireFormat.WIRETYPE_LENGTH_DELIMITED) parseChild(s, field, in.readBytes().toByteArray());
- else in.skipField(tag);
- }
- return s;
+ return parse(id, null, bytes);
}
-
- /** 解析顶层各子消息，字段号来自 iems-open-v1.0.5.proto。 */
- private void parseChild(UdanEmsDeviceState s, int field, byte[] data) throws IOException {
- switch (field) {
- case 2: s.setEms(parseEms(data)); break;
- case 4: s.getPcsStates().add(parsePcs(data)); break;
- case 5: s.getBatteryStackStates().add(parseStack(data)); break;
- case 6: s.getBatteryClusterStates().add(parseCluster(data)); break;
- case 7: s.getElectricMeterStates().add(parseMeter(data)); break;
- case 8: s.setPowerMeterCount(s.getPowerMeterCount() + 1); break;
- case 9: s.setAirConditionerCount(s.getAirConditionerCount() + 1); break;
- case 11: s.setFireCount(s.getFireCount() + 1); break;
- case 13: s.setJpfCount(s.getJpfCount() + 1); break;
- case 14: s.setTmsCount(s.getTmsCount() + 1); break;
- case 18: s.setDehumidifierCount(s.getDehumidifierCount() + 1); break;
- case 19: s.setUpsCount(s.getUpsCount() + 1); break;
- case 20: s.getPvStates().add(parsePv(data)); break;
- case 21: s.getChargeStationStates().add(parseChargeStation(data)); break;
- case 23: s.setExtPeripheralCount(s.getExtPeripheralCount() + 1); break;
- default: break;
- }
- }
-
- /** 解析 EMS 状态对象中的实际字段。 */
- private UdanEmsDeviceState.EmsState parseEms(byte[] b) throws IOException {
- UdanEmsDeviceState.EmsState x = new UdanEmsDeviceState.EmsState();
- read(b, (f, t, in) -> { if (t == 2 && f == 1) x.setDeviceId(in.readString()); else if (t == 5 && f == 18) x.setDemandPower(in.readFloat()); else if (t == 2 && f == 21) x.setProtoProtocolVersion(in.readString()); else if (t == 5 && f == 33) x.setDailyChargeEnergyKwh(in.readFloat()); else if (t == 5 && f == 34) x.setDailyDischargeEnergyKwh(in.readFloat()); else if (t == 1 && f == 37) x.setTotalChargeEnergyKwh(in.readDouble()); else if (t == 1 && f == 38) x.setTotalDischargeEnergyKwh(in.readDouble()); else if (t == 5 && f == 43) x.setSumsSoc(in.readFloat()); else if (t == 5 && f == 69) x.setSumsSoh(in.readFloat()); else if (t == 5 && f == 41) x.setDcBusVoltage(in.readFloat()); else if (t == 5 && f == 42) x.setDcBusCurrent(in.readFloat()); else skip(in, t); });
+ private EmsState parseEms(byte[] bytes) throws IOException {
+ EmsState x = new EmsState();
+ read(bytes, (f, t, in) -> {
+ switch (f) {
+ case 1: x.setDeviceId(str(in,t)); break; case 9: x.setEmsStatus(enumName(in,t,"RunStatus")); break;
+ case 18: x.setDemandPower(f32(in,t)); break; case 20: x.setTime(u64(in,t)); break; case 21: x.setProtoProtocolVersion(str(in,t)); break;
+ case 22: x.setEmsRunStatus(enumAt(EmsState.EmsRunStatus.values(),in,t)); break; case 23: x.setEmsWorkStatus(enumAt(EmsState.EmsWorkStatus.values(),in,t)); break;
+ case 24: x.setFaultStatus(enumAt(EmsState.ErrorStatus.values(),in,t)); break; case 25: x.setOnGridStatus(enumName(in,t,"GridStatus")); break;
+ case 26: x.setControlSrc(enumAt(EmsState.ControlSrc.values(),in,t)); break; case 27: x.setEmuModeCode(enumAt(EmsState.EmuModeCode.values(),in,t)); break;
+ case 28: x.setEmuAutoMod(enumAt(EmsState.EmuAutoMode.values(),in,t)); break; case 29: x.setOverDischargeProtect(enumAt(EmsState.YesNo.values(),in,t)); break;
+ case 30: x.setOverChargeProtect(enumAt(EmsState.YesNo.values(),in,t)); break; case 31: x.setDemandControl(enumAt(EmsState.YesNo.values(),in,t)); break; case 32: x.setCapacityExpansionModeCtrl(enumAt(EmsState.YesNo.values(),in,t)); break;
+ case 33: x.setDailyChargeEnergyKwh(f32(in,t)); break; case 34: x.setDailyDischargeEnergyKwh(f32(in,t)); break; case 35: x.setMonthlyChargeEnergyKwh(f32(in,t)); break; case 36: x.setMonthlyDischargeEnergyKwh(f32(in,t)); break;
+ case 37: x.setTotalChargeEnergyKwh(f64(in,t)); break; case 38: x.setTotalDischargeEnergyKwh(f64(in,t)); break; case 39: x.setPcsSumsActivePower(f32(in,t)); break; case 40: x.setPcsSumsApparentPower(f32(in,t)); break;
+ case 41: x.setDcBusVoltage(f32(in,t)); break; case 42: x.setDcBusCurrent(f32(in,t)); break; case 43: x.setSumsSoc(f32(in,t)); break; case 44: x.setCellSums(u32(in,t)); break;
+ case 45: x.setBatteryAverageTempr(f32(in,t)); break; case 46: x.setBatteryMaxTemprDiff(f32(in,t)); break; case 47: x.setMaxCellVoltage(u32(in,t)); break; case 48: x.setMinCellVoltage(u32(in,t)); break;
+ case 49: x.setSystemMaxAllowChgPower(f32(in,t)); break; case 50: x.setSystemMaxAllowDisPower(f32(in,t)); break; case 51: x.setRechargeableKwh(f32(in,t)); break; case 52: x.setDischargeableKwh(f32(in,t)); break;
+ case 53: x.setCurrentMaxAllowChgPower(f32(in,t)); break; case 54: x.setCurrentMaxAllowDisPower(f32(in,t)); break; case 55: x.setSocUpperLimit(f32(in,t)); break; case 56: x.setSocLowerLimit(f32(in,t)); break;
+ case 57: x.setMaxPowerDischargeDuration(u32(in,t)); break; case 58: x.setMaxPowerChargeDuration(u32(in,t)); break; case 60: x.setMasterCommStatus(enumName(in,t,"CommStatus")); break;
+ case 61: x.setGridImportActiveEnergyKwh(f64(in,t)); break; case 62: x.setGridExportActiveEnergyKwh(f64(in,t)); break; case 63: x.setDcTotalPower(f32(in,t)); break; case 64: x.setLoadTotalPower(f32(in,t)); break;
+ case 65: x.setEnergyStoreTotalPower(f32(in,t)); break; case 66: x.setHighSideGridPower(f32(in,t)); break; case 67: x.setLowSideGridPower(f32(in,t)); break; case 68: x.setPvPower(f32(in,t)); break; case 69: x.setSumsSoh(f32(in,t)); break;
+ default: skip(in,t);
+ }
+ });
return x;
}
- /** 解析 PCS 实际字段。 */
- private UdanEmsDeviceState.PcsState parsePcs(byte[] b) throws IOException {
- UdanEmsDeviceState.PcsState x = new UdanEmsDeviceState.PcsState();
- read(b, (f, t, in) -> { if (t == 2 && f == 2) x.setDeviceId(in.readString()); else if (t == 5 && f == 13) x.setAcTotalActivePower(in.readFloat()); else if (t == 5 && f == 21) x.setAcTotalReactivePower(in.readFloat()); else if (t == 5 && f == 22) x.setAcTotalApparentPower(in.readFloat()); else if (t == 5 && f == 23) x.setAcTotalPowerFactor(in.readFloat()); else if (t == 5 && f == 17) x.setDailyChargeEnergyKwh(in.readFloat()); else if (t == 5 && f == 18) x.setDailyDischargeEnergyKwh(in.readFloat()); else if (t == 1 && f == 19) x.setTotalChargeEnergyKwh(in.readDouble()); else if (t == 1 && f == 20) x.setTotalDischargeEnergyKwh(in.readDouble()); else if (t == 5 && f == 59) x.setAmbientTemp(in.readFloat()); else skip(in, t); });
- return x;
+ private PowerMeterState parsePowerMeter(byte[] bytes) throws IOException {
+ PowerMeterState value = new PowerMeterState();
+ read(bytes, (field, type, input) -> {
+ switch (field) {
+ case 1: value.setCumulativeImportAmountPeak(f64(input, type)); break;
+ case 2: value.setCumulativeImportAmountShoulder(f64(input, type)); break;
+ case 3: value.setCumulativeImportAmountValley(f64(input, type)); break;
+ case 4: value.setCumulativeImportAmount(f64(input, type)); break;
+ case 5: value.setDailyImportAmount((float) f64(input, type)); break;
+ case 6: value.setMonthlyImportAmount(f64(input, type)); break;
+ case 7: value.setDailyProfit(f64(input, type)); break;
+ case 8: value.setYesterdayProfit(f64(input, type)); break;
+ case 9: value.setMonthlyProfit(f64(input, type)); break;
+ case 10: value.setLastMonthlyProfit(f64(input, type)); break;
+ case 11: value.setCumulativeProfit(f64(input, type)); break;
+ case 12: value.setCumulativeImportAmountSharp(f64(input, type)); break;
+ case 15: value.setCumulativeImportKwh(f64(input, type)); break;
+ case 16: value.setCumulativeImportKwhSharp(f64(input, type)); break;
+ case 17: value.setCumulativeImportKwhPeak(f64(input, type)); break;
+ case 18: value.setCumulativeImportKwhShoulder(f64(input, type)); break;
+ case 19: value.setCumulativeImportKwhValley(f64(input, type)); break;
+ case 20: value.setCumulativeExportKwh(f64(input, type)); break;
+ case 21: value.setCumulativeExportKwhSharp(f64(input, type)); break;
+ case 22: value.setCumulativeExportKwhPeak(f64(input, type)); break;
+ case 23: value.setCumulativeExportKwhShoulder(f64(input, type)); break;
+ case 24: value.setCumulativeExportKwhValley(f64(input, type)); break;
+ case 25: value.setMonthlyImportKwh(f64(input, type)); break;
+ case 26: value.setMonthlyImportKwhSharp(f64(input, type)); break;
+ case 27: value.setMonthlyImportKwhPeak(f64(input, type)); break;
+ case 28: value.setMonthlyImportKwhShoulder(f64(input, type)); break;
+ case 29: value.setMonthlyImportKwhValley(f64(input, type)); break;
+ case 30: value.setMonthlyExportKwh(f64(input, type)); break;
+ case 31: value.setMonthlyExportKwhSharp(f64(input, type)); break;
+ case 32: value.setMonthlyExportKwhPeak(f64(input, type)); break;
+ case 33: value.setMonthlyExportKwhShoulder(f64(input, type)); break;
+ case 34: value.setMonthlyExportKwhValley(f64(input, type)); break;
+ case 35: value.setDailyImportKwh(f32(input, type)); break;
+ case 36: value.setDailyImportKwhSharp(f32(input, type)); break;
+ case 37: value.setDailyImportKwhPeak(f32(input, type)); break;
+ case 38: value.setDailyImportKwhShoulder(f32(input, type)); break;
+ case 39: value.setDailyImportKwhValley(f32(input, type)); break;
+ case 40: value.setDailyExportKwh(f32(input, type)); break;
+ case 41: value.setDailyExportKwhSharp(f32(input, type)); break;
+ case 42: value.setDailyExportKwhPeak(f32(input, type)); break;
+ case 43: value.setDailyExportKwhShoulder(f32(input, type)); break;
+ case 44: value.setDailyExportKwhValley(f32(input, type)); break;
+ case 45: value.setCumulativeExportAmount(f64(input, type)); break;
+ case 46: value.setCumulativeExportAmountSharp(f64(input, type)); break;
+ case 47: value.setCumulativeExportAmountPeak(f64(input, type)); break;
+ case 48: value.setCumulativeExportAmountShoulder(f64(input, type)); break;
+ case 49: value.setCumulativeExportAmountValley(f64(input, type)); break;
+ case 50: value.setMonthlyImportAmountSharp(f64(input, type)); break;
+ case 51: value.setMonthlyImportAmountPeak(f64(input, type)); break;
+ case 52: value.setMonthlyImportAmountShoulder(f64(input, type)); break;
+ case 53: value.setMonthlyImportAmountValley(f64(input, type)); break;
+ case 54: value.setMonthlyExportAmount(f64(input, type)); break;
+ case 55: value.setMonthlyExportAmountSharp(f64(input, type)); break;
+ case 56: value.setMonthlyExportAmountPeak(f64(input, type)); break;
+ case 57: value.setMonthlyExportAmountShoulder(f64(input, type)); break;
+ case 58: value.setMonthlyExportAmountValley(f64(input, type)); break;
+ case 59: value.setDailyImportAmountSharp(f32(input, type)); break;
+ case 60: value.setDailyImportAmountPeak(f32(input, type)); break;
+ case 61: value.setDailyImportAmountShoulder(f32(input, type)); break;
+ case 62: value.setDailyImportAmountValley(f32(input, type)); break;
+ case 63: value.setDailyExportAmount(f32(input, type)); break;
+ case 64: value.setDailyExportAmountSharp(f32(input, type)); break;
+ case 65: value.setDailyExportAmountPeak(f32(input, type)); break;
+ case 66: value.setDailyExportAmountShoulder(f32(input, type)); break;
+ case 67: value.setDailyExportAmountValley(f32(input, type)); break;
+ case 68: value.setTime(u64(input, type)); break;
+ default: skip(input, type);
+ }
+ });
+ return value;
}
- /** 解析电池堆实际字段。 */
- private UdanEmsDeviceState.BatteryStackState parseStack(byte[] b) throws IOException { UdanEmsDeviceState.BatteryStackState x = new UdanEmsDeviceState.BatteryStackState(); read(b, (f,t,in)-> { if(t==2&&f==2)x.setDeviceId(in.readString()); else if(t==5&&f==7)x.setTv(in.readFloat()); else if(t==5&&f==8)x.setTc(in.readFloat()); else if(t==5&&f==9)x.setSoh(in.readFloat()); else if(t==5&&f==10)x.setSoc(in.readFloat()); else if(t==5&&f==31)x.setRechargeableCapacityKwh(in.readFloat()); else if(t==5&&f==32)x.setDischargeableCapacityKwh(in.readFloat()); else if(t==1&&f==33)x.setCumulativeDischargeKwh(in.readDouble()); else if(t==1&&f==34)x.setCumulativeChargeKwh(in.readDouble()); else skip(in,t); }); return x; }
- /** 解析电池簇实际字段。 */
- private UdanEmsDeviceState.BatteryClusterState parseCluster(byte[] b) throws IOException { UdanEmsDeviceState.BatteryClusterState x = new UdanEmsDeviceState.BatteryClusterState(); read(b,(f,t,in)->{if(t==2&&f==2)x.setDeviceId(in.readString());else if(t==2&&f==3)x.setStackDeviceId(in.readString());else if(t==5&&f==7)x.setTv(in.readFloat());else if(t==5&&f==8)x.setTc(in.readFloat());else if(t==5&&f==9)x.setSoh(in.readFloat());else if(t==5&&f==10)x.setSoc(in.readFloat());else skip(in,t);});return x; }
- /** 解析电表实际字段。 */
- private UdanEmsDeviceState.ElectricMeterState parseMeter(byte[] b) throws IOException { UdanEmsDeviceState.ElectricMeterState x = new UdanEmsDeviceState.ElectricMeterState(); read(b,(f,t,in)->{if(t==2&&f==2)x.setDeviceId(in.readString());else if(t==1&&f==17)x.setCumulativeImportKwh(in.readDouble());else if(t==1&&f==22)x.setCumulativeExportKwh(in.readDouble());else if(t==5&&f==54)x.setAcTotalActivePower(in.readFloat());else skip(in,t);});return x; }
- /** 解析光伏实际字段。 */
- private UdanEmsDeviceState.PvState parsePv(byte[] b) throws IOException { UdanEmsDeviceState.PvState x = new UdanEmsDeviceState.PvState(); read(b,(f,t,in)->{if(t==2&&f==1)x.setDeviceId(in.readString());else if(t==5&&f==18)x.setActivePower(in.readFloat());else if(t==5&&f==19)x.setReactivePower(in.readFloat());else if(t==5&&f==20)x.setPowerFactor(in.readFloat());else if(t==5&&f==21)x.setDailyPowerGeneration(in.readFloat());else if(t==1&&f==22)x.setTotalPowerGeneration(in.readDouble());else skip(in,t);});return x; }
- /** 解析充电桩实际字段。 */
- private UdanEmsDeviceState.ChargeStationState parseChargeStation(byte[] b) throws IOException { UdanEmsDeviceState.ChargeStationState x = new UdanEmsDeviceState.ChargeStationState(); read(b,(f,t,in)->{if(t==2&&f==1)x.setDeviceId(in.readString());else if(t==0&&f==2)x.setTime(in.readUInt64());else skip(in,t);});return x; }
+ private void read(byte[] bytes, Reader reader) throws IOException {
+ CodedInputStream input = CodedInputStream.newInstance(bytes == null ? new byte[0] : bytes);
+ while (!input.isAtEnd()) {
+ int tag = input.readTag();
+ if (tag == 0) {
+ break;
+ }
+ reader.read(WireFormat.getTagFieldNumber(tag), WireFormat.getTagWireType(tag), input);
+ }
+ }
- /** 遍历嵌套 protobuf 字段，并把字段号、线类型和读取器交给解析器。 */
- private void read(byte[] b, FieldReader reader) throws IOException { CodedInputStream in=CodedInputStream.newInstance(b); while(!in.isAtEnd()){int tag=in.readTag();if(tag==0)break;reader.read(WireFormat.getTagFieldNumber(tag),WireFormat.getTagWireType(tag),in);} }
- /** 跳过当前不需要映射的字段。 */
- private void skip(CodedInputStream in, int type) throws IOException {
+ private static String str(CodedInputStream input, int type) throws IOException {
+ if (type == WireFormat.WIRETYPE_LENGTH_DELIMITED) {
+ return input.readString();
+ }
+ skip(input, type);
+ return null;
+ }
+
+ private static long u64(CodedInputStream input, int type) throws IOException {
+ if (type == WireFormat.WIRETYPE_VARINT) {
+ return input.readUInt64();
+ }
+ skip(input, type);
+ return 0L;
+ }
+
+ private static int u32(CodedInputStream input, int type) throws IOException {
+ return (int) u64(input, type);
+ }
+
+ private static float f32(CodedInputStream input, int type) throws IOException {
+ if (type == WireFormat.WIRETYPE_FIXED32) {
+ return input.readFloat();
+ }
+ skip(input, type);
+ return 0F;
+ }
+
+ private static double f64(CodedInputStream input, int type) throws IOException {
+ if (type == WireFormat.WIRETYPE_FIXED64) {
+ return input.readDouble();
+ }
+ skip(input, type);
+ return 0D;
+ }
+ private static String enumName(CodedInputStream input, int type, String prefix) throws IOException {
+ if (type != WireFormat.WIRETYPE_VARINT) {
+ skip(input, type);
+ return null;
+ }
+ int number = input.readInt32();
+ if ("RunStatus".equals(prefix)) {
+ return new String[]{"RUN_STATUS_UNKNOWN", "RUN_STATUS_STOPPED", "RUN_STATUS_RUNNING"}[number < 0 || number > 2 ? 0 : number];
+ }
+ if ("GridStatus".equals(prefix)) {
+ return new String[]{"GRID_STATUS_UNKNOWN", "GRID_STATUS_ON_GRID", "GRID_STATUS_OFF_GRID"}[number < 0 || number > 2 ? 0 : number];
+ }
+ if ("CommStatus".equals(prefix)) {
+ return new String[]{"COMM_STATUS_UNKNOWN", "COMM_STATUS_DISCONNECTED", "COMM_STATUS_CONNECTED"}[number < 0 || number > 2 ? 0 : number];
+ }
+ return prefix + "_" + number;
+ }
+ private static <T> T enumAt(T[] values, CodedInputStream input, int type) throws IOException {
+ int number = (int) u64(input, type);
+ return number >= 0 && number < values.length ? values[number] : null;
+ }
+
+ private static void skip(CodedInputStream input, int type) throws IOException {
switch (type) {
- case WireFormat.WIRETYPE_VARINT: in.readInt64(); break;
- case WireFormat.WIRETYPE_FIXED64: in.readFixed64(); break;
- case WireFormat.WIRETYPE_LENGTH_DELIMITED: in.readBytes(); break;
- case WireFormat.WIRETYPE_FIXED32: in.readFixed32(); break;
- default: break;
+ case WireFormat.WIRETYPE_VARINT:
+ input.readInt64();
+ break;
+ case WireFormat.WIRETYPE_FIXED64:
+ input.readFixed64();
+ break;
+ case WireFormat.WIRETYPE_LENGTH_DELIMITED:
+ input.readBytes();
+ break;
+ case WireFormat.WIRETYPE_FIXED32:
+ input.readFixed32();
+ break;
+ default:
+ throw new IOException("Unsupported protobuf wire type: " + type);
}
}
- @FunctionalInterface private interface FieldReader { void read(int field, int type, CodedInputStream input) throws IOException; }
+
+ @FunctionalInterface
+ private interface Reader {
+ void read(int field, int type, CodedInputStream input) throws IOException;
+ }
}
diff --git a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/util/UdanEmsKafkaUtil.java b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/util/UdanEmsKafkaUtil.java
index 45fc1e1..0da14ad 100644
--- a/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/util/UdanEmsKafkaUtil.java
+++ b/ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/util/UdanEmsKafkaUtil.java
@@ -8,7 +8,7 @@
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
@@ -20,13 +20,11 @@
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.Duration;
-import java.util.Base64;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
/**
* 优旦 EMS 平台 Kafka 消费工具。
@@ -48,7 +46,7 @@
});
private volatile KafkaConsumer<String, byte[]> consumer;
private volatile boolean running;
- private volatile MessageHandler messageHandler = (key, payload) -> { };
+ private volatile MessageHandler messageHandler = (key, payload, emsDeviceId) -> { };
private volatile ErrorHandler errorHandler = Throwable::printStackTrace;
/**
@@ -65,7 +63,7 @@
* 当 udan.ems.kafka.enabled=true 时由 Spring 在 Bean 初始化后自动调用。
* PostConstruct 创建并完成依赖注入后，自动执行这个方法一次
*/
- //@PostConstruct
+ @PostConstruct
public void start() {
validate();
running = true;
@@ -78,7 +76,7 @@
* @param handler 接收 EMS 系统 ID 和 protobuf 原始字节的处理器；传入 null 时忽略消息
*/
public void setMessageHandler(MessageHandler handler) {
- this.messageHandler = handler == null ? (key, payload) -> { } : handler;
+ this.messageHandler = handler == null ? (key, payload, emsDeviceId) -> { } : handler;
}
/**
@@ -93,16 +91,15 @@
/**
* 按配置的超时时间轮询一次 Kafka，并将取得的消息交给消息处理器。
*
- * @return 本次轮询取得的消息数量
* @throws IllegalStateException 消费者尚未启动时抛出
*/
- public int pollOnce() {
+ public void pollOnce() {
if (consumer == null) {
throw new IllegalStateException("Kafka consumer has not started");
}
ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(Math.max(1, properties.getPollSeconds())));
dispatch(records);
- return records.count();
+ records.count();
}
/**
@@ -125,8 +122,6 @@
consumer = c;
c.subscribe(Collections.singletonList(properties.getTopic()));
while (running) pollOnce();
- } catch (WakeupException ignored) {
- if (running) errorHandler.onError(ignored);
} catch (Throwable e) {
if (running) errorHandler.onError(e);
} finally {
@@ -147,11 +142,23 @@
log.info("1、UDAN EMS可读字段: emsSystemId={}, data={}", record.key(), toReadableProtobuf(payload));
// 此处即为业务接入点：解析 protobuf 后调用保存服务，当前仅由回调方决定如何处理。
- messageHandler.onMessage(record.key(), payload);
+ messageHandler.onMessage(record.key(), payload, getEmsDeviceId(record));
} catch (Throwable e) {
errorHandler.onError(e);
}
}
+ }
+
+ /** 获取消息 Header 中的 EMS 设备 ID。优先使用 emsDeviceId，兼容单 Header 报文。 */
+ private String getEmsDeviceId(ConsumerRecord<String, byte[]> record) {
+ Header header = record.headers().lastHeader("ems_device_id");
+ if (header == null && !Objects.isNull(record.headers())) {
+ header = record.headers().iterator().next();
+ }
+ if (header == null || header.value() == null) {
+ return null;
+ }
+ return new String(header.value(), StandardCharsets.UTF_8);
}
/**
@@ -237,7 +244,6 @@
/** 判断长度字段是否是可直接阅读的 UTF-8 文本。 */
private static boolean isReadableString(byte[] data) {
- if (data.length == 0) return true;
for (byte value : data) {
int c = value & 0xFF;
if (c < 0x20 || c > 0x7E) return false;
@@ -304,8 +310,9 @@
*
* @param emsSystemId Kafka Key，即 EMS 系统 ID
* @param protobufBytes Kafka Value，即 protobuf3 原始字节
+ * @param emsDeviceId Kafka Header 中的 EMS 设备 ID
*/
- void onMessage(String emsSystemId, byte[] protobufBytes);
+ void onMessage(String emsSystemId, byte[] protobufBytes, String emsDeviceId);
}
/** 消费和消息处理异常回调。 */
diff --git a/ruoyi-service/ruoyi-other/src/main/resources/mapper/other/EmindStationMapper.xml b/ruoyi-service/ruoyi-other/src/main/resources/mapper/other/EmindStationMapper.xml
index 759e98e..dae23eb 100644
--- a/ruoyi-service/ruoyi-other/src/main/resources/mapper/other/EmindStationMapper.xml
+++ b/ruoyi-service/ruoyi-other/src/main/resources/mapper/other/EmindStationMapper.xml
@@ -54,7 +54,7 @@
<select id="selectManagementPage" resultMap="stationResultMap">
SELECT id, station_id, name, address, province, city, longitude, latitude,
max_power, capacity, commissioning_time, platform_name, source_type,
- management_deleted, del_flag, synced_at, create_time, update_time
+ management_deleted, del_flag, synced_at, create_time, update_time, station_code
FROM t_emind_station
WHERE del_flag = 0 AND management_deleted = 0
<if test="query.platformName != null and query.platformName != ''">
diff --git a/ruoyi-service/ruoyi-other/src/main/resources/sql/t_screen_content_alter.sql b/ruoyi-service/ruoyi-other/src/main/resources/sql/t_screen_content_alter.sql
index 342d894..2328ae1 100644
--- a/ruoyi-service/ruoyi-other/src/main/resources/sql/t_screen_content_alter.sql
+++ b/ruoyi-service/ruoyi-other/src/main/resources/sql/t_screen_content_alter.sql
@@ -1,4 +1,8 @@
-- ----------------------------
--- 增加大屏名称配置字段
+-- 1、增加大屏名称配置字段
alter table t_screen_content add column screen_name varchar(255) default 'NULL' null comment '大屏名称';
+-- 2、增加储能电站表设备编号
+alter table t_emind_station add column station_code varchar(255) default 'NULL' null comment '设备编号';
+
+
--
Gitblit v1.7.1