package com.ruoyi.other.udan.service; 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.emind.service.EmindEnergyChainDetailService; import com.ruoyi.other.udan.model.EmsState; import com.ruoyi.other.udan.model.BatteryStackState; import com.ruoyi.other.udan.model.ElectricMeterState; import com.ruoyi.other.udan.model.PcsState; import com.ruoyi.other.udan.model.PowerMeterState; import com.ruoyi.other.udan.model.TmsState; import com.ruoyi.other.udan.model.UdanEmsDeviceState; import com.ruoyi.other.udan.util.UdanEmsKafkaUtil; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.StringUtils; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.io.IOException; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.List; import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; /** 将 IEmsAllDeviceState 中模型已有的 EMS 和能耗计字段转换为业务对象。 */ @Slf4j @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; private final EmindEnergyChainDetailService emindEnergyChainDetailService; public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil, EmindStorageDashboardMapper dashboardMapper, EmindStationMapper stationMapper, EmindEnergyChainDetailService emindEnergyChainDetailService) { this.kafkaUtil = kafkaUtil; this.dashboardMapper = dashboardMapper; this.stationMapper = stationMapper; this.emindEnergyChainDetailService = emindEnergyChainDetailService; } @PostConstruct public void register() { kafkaUtil.setMessageHandler(this::process); } public void process(String id, byte[] bytes, String emsDeviceId) { try { LocalDateTime now = LocalDateTime.now(); UdanEmsDeviceState udanEmsDeviceState = parse(id, emsDeviceId, bytes); if(StringUtils.isNotEmpty(udanEmsDeviceState.getEmsDeviceId())) { log.info("2、UDAN EMS实际对象: {}", JSON.toJSONString(udanEmsDeviceState)); List stations = stationMapper.selectRemoteActiveStations(); EmindStation station = stations.stream().filter(s -> udanEmsDeviceState.getEmsDeviceId().equals(s.getStationCode())).findFirst().orElse(null); if(Objects.nonNull(station)) { List daily = udanEmsDeviceState.convert(station.getStationId()); Set dateSet = daily.stream().map(EmindIncomeHistoryDto::getDate).collect(Collectors.toSet()); // 删除电站下该天已存在的旧数据 if(CollectionUtils.isNotEmpty(dateSet)) { dashboardMapper.deleteByStationIdDay(station.getStationId(), dateSet); } if(CollectionUtils.isNotEmpty(daily)) { dashboardMapper.upsertDaily(station.getStationId(), daily, now); } }else{ log.error("UDAN EMS数据同步错误-电站信息不存在！"); } // 保存设备其它信息 emindEnergyChainDetailService.saveAll(udanEmsDeviceState); } } catch (IOException e) { log.error("UDAN EMS protobuf解析失败: emsSystemId={}, bytes={}", id, bytes == null ? 0 : bytes.length, e); } } /** 兼容不带 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 == 4 && state.getPcsState() == null) { state.setPcsState(parsePcs(in.readBytes().toByteArray())); } else if (type == 2 && field == 5 && state.getBatteryStackState() == null) { state.setBatteryStackState(parseBatteryStack(in.readBytes().toByteArray())); } else if (type == 2 && field == 7 && state.getElectricMeterState() == null) { state.setElectricMeterState(parseElectricMeter(in.readBytes().toByteArray())); } else if (type == 2 && field == 8) state.setPowerMeterState(parsePowerMeter(in.readBytes().toByteArray())); else if (type == 2 && field == 14 && state.getTmsState() == null) { state.setTmsState(parseTms(in.readBytes().toByteArray())); } else skip(in, type); }); return state; } /** 兼容不带 Header 的直接解析调用。 */ public UdanEmsDeviceState parse(String id, byte[] bytes) throws IOException { return parse(id, null, bytes); } 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; } 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 PcsState parsePcs(byte[] bytes) throws IOException { PcsState value = new PcsState(); read(bytes, (field, type, input) -> { switch (field) { case 2: value.setDeviceId(str(input, type)); break; case 3: value.setErrorStatus(enumName(input, type, "ErrorStatus")); break; case 4: value.setAlarmStatus(enumName(input, type, "AlarmStatus")); break; case 5: value.setRunStatus(enumName(input, type, "RunStatus")); break; case 6: value.setOnGridStatus(enumName(input, type, "GridStatus")); break; case 13: value.setAcTotalActivePower((double) f32(input, type)); break; case 17: value.setDailyChargeEnergyKwh((double) f32(input, type)); break; case 18: value.setDailyDischargeEnergyKwh((double) f32(input, type)); break; case 19: value.setTotalChargeEnergyKwh(f64(input, type)); break; case 20: value.setTotalDischargeEnergyKwh(f64(input, type)); break; case 31: value.setAPhaseActivePower((double) f32(input, type)); break; case 34: value.setBPhaseActivePower((double) f32(input, type)); break; case 37: value.setCPhaseActivePower((double) f32(input, type)); break; case 50: value.setControlMode(enumName(input, type, "PcsControlMode")); break; case 60: value.setCommStatus(enumName(input, type, "CommStatus")); break; default: skip(input, type); } }); return value; } private BatteryStackState parseBatteryStack(byte[] bytes) throws IOException { BatteryStackState value = new BatteryStackState(); read(bytes, (field, type, input) -> { switch (field) { case 2: value.setDeviceId(str(input, type)); break; case 4: value.setWorkStatus(enumName(input, type, "BatteryWorkStatus")); break; case 5: value.setErrorStatus(enumName(input, type, "ErrorStatus")); break; case 6: value.setWarningStatus(enumName(input, type, "AlarmStatus")); break; case 7: value.setTv((double) f32(input, type)); break; case 8: value.setTc((double) f32(input, type)); break; case 9: value.setSoh((double) f32(input, type)); break; case 10: value.setSoc((double) f32(input, type)); break; case 15: value.setMinCellVoltMv((double) u32(input, type)); break; case 23: value.setMaxCellVoltMv((double) u32(input, type)); break; case 31: value.setRechargeableCapacityKwh((double) f32(input, type)); break; case 32: value.setDischargeableCapacityKwh((double) f32(input, type)); break; case 33: value.setCumulativeDischargeKwh(f64(input, type)); break; case 34: value.setCumulativeChargeKwh(f64(input, type)); break; case 36: value.setAvgCellTemp((double) u32(input, type) - 40D); break; case 52: value.setCommStatus(enumName(input, type, "CommStatus")); break; case 48: value.setBatteryStatus(enumName(input, type, "BatteryStatus")); break; default: skip(input, type); } }); return value; } private ElectricMeterState parseElectricMeter(byte[] bytes) throws IOException { ElectricMeterState value = new ElectricMeterState(); read(bytes, (field, type, input) -> { switch (field) { case 2: value.setDeviceId(str(input, type)); break; case 48: value.setAPhaseVolt((double) f32(input, type)); break; case 49: value.setBPhaseVolt((double) f32(input, type)); break; case 50: value.setCPhaseVolt((double) f32(input, type)); break; case 51: value.setAPhaseCurrent((double) f32(input, type)); break; case 52: value.setBPhaseCurrent((double) f32(input, type)); break; case 53: value.setCPhaseCurrent((double) f32(input, type)); break; case 54: value.setAcTotalActivePower((double) f32(input, type)); break; case 69: value.setFrequency((double) f32(input, type)); break; case 74: value.setAcTotalPowerFactor((double) f32(input, type)); break; default: skip(input, type); } }); return value; } private TmsState parseTms(byte[] bytes) throws IOException { TmsState value = new TmsState(); read(bytes, (field, type, input) -> { switch (field) { case 1: value.setDeviceId(str(input, type)); break; case 2: value.setRun_status(enumName(input, type, "TmsRunStatus")); break; case 3: value.setError_status(enumName(input, type, "ErrorStatus")); break; case 4: value.setSupply_temp((double) f32(input, type)); break; case 5: value.setReturn_temp((double) f32(input, type)); break; case 8: value.setEnv_temp((double) f32(input, type)); break; case 21: value.setCool_start_temp((double) f32(input, type)); break; case 22: value.setCool_stop_temp((double) f32(input, type)); break; case 23: value.setHeat_start_temp((double) f32(input, type)); break; case 24: value.setHeat_stop_temp((double) f32(input, type)); break; case 6: value.setTms_mode(enumName(input, type, "TmsMode")); break; case 28: value.setControl_mode(enumName(input, type, "TmsControlMode")); break; case 25: value.setComm_status(enumName(input, type, "CommStatus")); break; default: skip(input, type); } }); return value; } private void read(byte[] bytes, Reader reader) throws IOException { CodedInputStream input = CodedInputStream.newInstance(bytes == null ? new byte[0] : bytes); while (!input.isAtEnd()) { int tag = input.readTag(); if (tag == 0) { break; } reader.read(WireFormat.getTagFieldNumber(tag), WireFormat.getTagWireType(tag), input); } } 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]; } if ("ErrorStatus".equals(prefix)) { return new String[]{"ERROR_STATUS_UNKNOWN", "ERROR_STATUS_OCCURRING", "ERROR_STATUS_NONE"}[number < 0 || number > 2 ? 0 : number]; } if ("AlarmStatus".equals(prefix)) { return new String[]{"ALARM_STATUS_UNKNOWN", "ALARM_STATUS_OCCURRING", "ALARM_STATUS_NONE"}[number < 0 || number > 2 ? 0 : number]; } if ("BatteryWorkStatus".equals(prefix)) { return new String[]{"BATTERY_WORK_STATUS_UNKNOWN", "BATTERY_WORK_STATUS_CHARGING", "BATTERY_WORK_STATUS_DISCHARGING", "BATTERY_WORK_STATUS_IDLE"}[number < 0 || number > 3 ? 0 : number]; } if ("TmsRunStatus".equals(prefix)) { return new String[]{"TMS_RUN_STATUS_UNKNOWN", "TMS_RUN_STATUS_IDLE", "TMS_RUN_STATUS_READY_TO_START", "TMS_RUN_STATUS_RUNNING", "TMS_RUN_STATUS_DELAYED_SHUTDOWN", "TMS_RUN_STATUS_SELF_TEST", "TMS_RUN_STATUS_ALARMING"}[number < 0 || number > 6 ? 0 : number]; } if ("PcsControlMode".equals(prefix)) { return new String[]{"PCS_CONTROL_MODE_UNKNOWN", "PCS_CONTROL_MODE_LOCAL_CONTROL", "PCS_CONTROL_MODE_REMOTE_CONTROL"}[number < 0 || number > 3 ? 0 : number]; } if ("BatteryStatus".equals(prefix)) { return new String[]{"Battery_Status_BAN_CHARGE_DISCHARGE", "Battery_Status_ONLY_CHARGE", "Battery_Status_ONLY_DISCHARGE","Battery_Status_NORMAL","Battery_Status_BAN_CHARGE_DISCHARGE_NEW","Battery_Status_IDLE"}[number < 0 || number > 6 ? 0 : number]; } if ("TmsMode".equals(prefix)) { return new String[]{"TMS_MODE_UNKNOWN", "TMS_MODE_REFRIGERATION", "TMS_MODE_HEATING","TMS_MODE_SELF_LOOP","TMS_MODE_AUTOMATIC","TMS_MODE_STOP"}[number < 0 || number > 6 ? 0 : number]; } if ("TmsControlMode".equals(prefix)) { return new String[]{"UNKNOWN", "OUTLET_WATER_TEMP", "INLET_WATER_TEMP","CELL_WEIGHT_TEMP","CELL_AVG_TEMP"}[number < 0 || number > 5 ? 0 : number]; } return prefix + "_" + number; } private static 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: 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 Reader { void read(int field, int type, CodedInputStream input) throws IOException; } }