feat: 储能数据同步，新运维大楼大屏的配置

yanderong

2026-09-15 868b19dddc1a28cdcf2703dac7a07df4b0b74221

feat: 储能数据同步，新运维大楼大屏的配置

 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;
}

 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;
    }

 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个小时
}

 ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/EmsState.java

New file
@@ -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;
        }
    }
}

 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;


 ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDatabaseRecord.java

File was deleted

 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; }
}

 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;
    }
}

 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);
    }

    /** 消费和消息处理异常回调。 */

 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 != ''">

 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 '设备编号';