ruoyi-api/ruoyi-api-other/src/main/java/com/ruoyi/other/api/domain/TScreenContent.java
@@ -94,4 +94,9 @@ @ApiModelProperty("更新时间") @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") private Date updateTime; @TableField("screen_name") @ApiModelProperty("大屏名称") private String screenName; } ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/builder/IotBuilder.java
@@ -35,7 +35,7 @@ return IoTDAClient.newBuilder() .withCredential(auth) // 标准版/企业版：需自行创建Region对象，基础版：请使用IoTDARegion的region对象，如"withRegion(IoTDARegion.CN_NORTH_4)" .withRegion(new Region("cn-south-1", IotConstant.URL)) .withRegion(new Region("cn-south-1", config.getEndpoint())) .build(); } /** ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/constant/IotConstant.java
@@ -6,15 +6,15 @@ /** * 华为云iot token */ public static final String IOT_TOKEN = "iot_token"; public static final String IOT_TOKEN = ""; /** * 模式 */ public static final String PASSWORD = "password"; public static final String PASSWORD = ""; /** * URL */ public static final String URL = "https://94cb3eb5f9.st1.iotda-app.cn-south-1.myhuaweicloud.com"; public static final String URL = ""; /** * 创建产品 */ ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java
@@ -373,6 +373,8 @@ platformConfirmationCharging.setAuthentication(0); platformConfirmationCharging.setFailure_cause(startVO == null ? null : startVO.getFailureCause()); } //先默认写死一个余额，后续再补充功能 platformConfirmationCharging.setAccount_balance(new BigDecimal(100)); iotMessageProduce.sendMessage(platformConfirmationCharging.getCharging_pile_code(), ServiceIdMenu.PLATFORM_CONFIRMATION_CHARGING.getKey(),messageUtil.platformConfirmationCharging(platformConfirmationCharging)); // 鉴权/建单失败时删除幂等键，放行设备重试（避免60秒内重新刷卡/VIN启动被误拦）；成功则保留键至TTL过期 if (!Integer.valueOf(1).equals(platformConfirmationCharging.getAuthentication())) { ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/tools/ChargingMessageUtil.java
@@ -181,6 +181,9 @@ // 持久化消息 UploadRealTimeMonitoringData uploadRealTimeMonitoringData = new UploadRealTimeMonitoringData(); BeanUtils.copyProperties(uploadRealTimeMonitoringDataMessage,uploadRealTimeMonitoringData); if(StringUtils.isEmpty(uploadRealTimeMonitoringDataMessage.getTransaction_serial_number())){ return; } // 查询mogondb上一条数据 UploadRealTimeMonitoringData data = uploadRealTimeMonitoringDataService.getLastDataById(uploadRealTimeMonitoringDataMessage.getTransaction_serial_number()); // 查询订单 ruoyi-service/ruoyi-integration/src/main/resources/logback.xml
@@ -7,8 +7,7 @@ <contextName>logback</contextName> <!-- name的值是变量的名称，value的值时变量定义的值。通过定义的值会被插入到logger上下文中。定义后，可以使“${}”来使用变量。 --> <!-- <property name="log.path" value="/data/log/app/ruoyi-integration"/>--> <property name="log.path" value="/Users/mitao/Documents/log/mx_charging_pile/log/app/ruoyi-integration"/> <property name="log.path" value="/data/log/app/ruoyi-integration"/> <!--0. 日志格式和颜色渲染 --> <!-- 彩色日志依赖的渲染类 --> ruoyi-service/ruoyi-jianguan/src/main/resources/logback.xml
@@ -8,7 +8,7 @@ <!-- name的值是变量的名称，value的值时变量定义的值。通过定义的值会被插入到logger上下文中。定义后，可以使“${}”来使用变量。 --> <!-- <property name="log.path" value="/data/log/app/ruoyi-jianguan"/>--> <property name="log.path" value="/Users/mitao/Documents/log/mx_charging_pile/log/app/ruoyi-jianguan"/> <property name="log.path" value="/data/log/app/ruoyi-jianguan"/> <!--0. 日志格式和颜色渲染 --> <!-- 彩色日志依赖的渲染类 --> ruoyi-service/ruoyi-other/pom.xml
@@ -15,6 +15,17 @@ </description> <dependencies> <!-- UDAN-EMS platform-to-platform Kafka protobuf payloads --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.6.1</version> </dependency> <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>3.25.3</version> </dependency> <!-- SpringCloud Alibaba Nacos --> <dependency> <groupId>com.alibaba.cloud</groupId> ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/controller/TStationInfoController.java
@@ -66,5 +66,7 @@ @PostMapping("/syncKsolarDataTask") @ApiOperation("手动同步数据") public void syncKsolarDataTask() { stationInfoService.syncKsolarDataTask(); stationInfoService.syncHuaWeiDataTask(); } } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/service/TStationCollectService.java
@@ -15,6 +15,9 @@ boolean saveCollectBaseList(JSONArray collectBaseList); /** 保存华为 FusionSolar 设备列表。 */ boolean saveCollectHuaWeiList(JSONArray devList); PageInfo<TStationCollect> pageList(StationCollectQuery query); boolean updateStationCollect(TStationCollect stationCollect); ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/service/TStationEarnService.java
@@ -22,6 +22,9 @@ boolean saveStationEarnList(JSONArray stationEarnList); /** 保存华为 FusionSolar 电站日统计数据。 */ boolean saveStationHuaWeiList(JSONArray stationDayList); LocalDate getMaxSaveTime(); PageInfo<TStationEarn> pageList(StationEarnQuery query); ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/service/TStationInfoService.java
@@ -37,4 +37,7 @@ * @return true可以删除，false不可以删除 */ Boolean validateDelete(Long[] ids); /** 同步华为 FusionSolar 电站数据。 */ void syncHuaWeiDataTask(); } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/service/impl/TStationCollectServiceImpl.java
@@ -262,6 +262,17 @@ } } private Integer parseInteger(String value) { if (StringUtils.isBlank(value)) { return 0; } try { return Integer.valueOf(value); } catch (NumberFormatException exception) { return parsePowerNum(value); } } private LocalDateTime parseDateTime(String value) { if (StringUtils.isBlank(value)) { return null; @@ -293,4 +304,51 @@ } return true; } /** * 将华为 FusionSolar 设备列表转换为逆变器记录并批量保存。 * devDn 作为设备唯一标识（接口未提供时使用 esnCode）。 */ @Override @Transactional(rollbackFor = Exception.class) public boolean saveCollectHuaWeiList(JSONArray devList) { if (devList == null || devList.isEmpty()) { return false; } Set<String> existingKeys = getExistingCollectKeys(); Set<String> pendingKeys = new HashSet<>(); List<TStationCollect> insertList = new ArrayList<>(devList.size()); for (Object item : devList) { if (item == null) { continue; } JSONObject json = JSONObject.parseObject(item.toString()); String collectId = firstNotBlank(json.getString("devDn"), json.getString("esnCode")); String stationId = json.getString("stationCode"); if (StringUtils.isBlank(collectId) || StringUtils.isBlank(stationId) || existingKeys.contains(collectId) || !pendingKeys.add(collectId)) { continue; } TStationCollect collect = new TStationCollect(); collect.setPlatformId(1L); collect.setStationId(stationId); collect.setStationName(firstNotBlank(json.getString("stationName"), "")); collect.setCollectId(collectId); collect.setCollectName(firstNotBlank(json.getString("devName"), collectId)); collect.setDeviceModel(firstNotBlank(json.getString("invType"), json.getString("model"))); collect.setInverterId(json.getString("esnCode")); collect.setInstalledCapacity(firstNotBlank(json.getString("capacity"), "0")); collect.setPowerNum(parseInteger(json.getString("optimizerNumber"))); collect.setLatitude(json.getString("latitude")); collect.setLongitude(json.getString("longitude")); collect.setStatus("1"); collect.setDataSource(2); collect.setSyncTime(LocalDateTime.now()); collect.setCreateTime(LocalDateTime.now()); collect.setUpdateTime(LocalDateTime.now()); fillDefaults(collect); insertList.add(collect); } return CollectionUtils.isNotEmpty(insertList) && saveBatch(insertList); } } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/service/impl/TStationEarnServiceImpl.java
@@ -29,6 +29,8 @@ import java.math.RoundingMode; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.Instant; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; import java.util.*; @@ -77,6 +79,80 @@ } /** * 将华为 FusionSolar 电站日指标转换为收益记录并批量保存。 */ @Override @Transactional(rollbackFor = Exception.class) public boolean saveStationHuaWeiList(JSONArray stationDayList) { if (stationDayList == null || stationDayList.isEmpty()) { return false; } Map<String, TStationEarn> existingByKey = getExistingEarnByKey(); Map<String, TStationEarn> recordsByKey = new LinkedHashMap<>(); for (Object item : stationDayList) { if (item == null) { continue; } JSONObject json = JSONObject.parseObject(item.toString()); String stationId = json.getString("stationCode"); LocalDate saveTime = parseHuaweiCollectTime(json.getString("collectTime")); if (StringUtils.isBlank(stationId) || saveTime == null) { continue; } JSONObject dataItemMap = json.getJSONObject("dataItemMap"); if (dataItemMap == null) { dataItemMap = new JSONObject(); } LocalDateTime now = LocalDateTime.now(); TStationEarn earn = new TStationEarn(); earn.setPlatformId(1L); earn.setStationId(stationId); earn.setStationName(json.getString("stationName")); earn.setSaveTime(saveTime); earn.setElecTotal(firstDecimal(dataItemMap, "PVYield", "inverterYield", "inverter_power")); earn.setEarnTotal(parseBigDecimal(dataItemMap.getString("power_profit"))); earn.setDataSource(2); earn.setCreateBy(""); earn.setUpdateBy(""); earn.setCreateTime(now); earn.setUpdateTime(now); earn.setDelFlag("0"); TStationEarn existing = existingByKey.get(buildEarnKey(earn)); if (existing != null) { earn.setId(existing.getId()); earn.setCreateTime(null); } recordsByKey.put(buildEarnKey(earn), earn); } if (recordsByKey.isEmpty()) { return false; } List<List<TStationEarn>> partitions = Lists.partition(new ArrayList<>(recordsByKey.values()), 500); partitions.forEach(this::saveOrUpdateBatch); return true; } private static LocalDate parseHuaweiCollectTime(String value) { if (StringUtils.isBlank(value)) { return null; } try { return Instant.ofEpochMilli(Long.parseLong(value)).atZone(ZoneId.systemDefault()).toLocalDate(); } catch (NumberFormatException exception) { return null; } } private static BigDecimal firstDecimal(JSONObject data, String... keys) { for (String key : keys) { if (data.get(key) != null && StringUtils.isNotBlank(data.getString(key))) { return parseBigDecimal(data.getString(key)); } } return BigDecimal.ZERO; } /** * 查询发电量统计记录中的最大统计日期。 * * @return 最大统计日期，无记录时返回 null ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/service/impl/TStationInfoServiceImpl.java
@@ -13,6 +13,7 @@ import com.ruoyi.other.service.TStationInfoService; import com.ruoyi.other.query.StationInfoQuery; import com.ruoyi.other.util.KsolarUtils; import com.ruoyi.other.util.HuWeiUtils; import com.ruoyi.other.vo.StationInfoStatusVO; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections4.CollectionUtils; @@ -23,6 +24,8 @@ import java.math.BigDecimal; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.OffsetDateTime; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; import java.util.*; @@ -43,6 +46,8 @@ @Autowired private TStationEarnService stationEarnService; @Autowired private HuWeiUtils huWeiUtils; /** * 每天同步科士达数据到数据库 @@ -311,6 +316,21 @@ } } private LocalDateTime parseHuaweiDateTime(String value) { if (StringUtils.isBlank(value)) { return null; } try { return OffsetDateTime.parse(value).toLocalDateTime(); } catch (DateTimeParseException ignored) { try { return ZonedDateTime.parse(value).toLocalDateTime(); } catch (DateTimeParseException ignoredAgain) { return parseDataTime(value); } } } private String firstNotBlank(String preferred, String fallback) { return StringUtils.isNotBlank(preferred) ? preferred : fallback; } @@ -364,4 +384,81 @@ return true; } /** * 同步华为 FusionSolar 电站数据。 * 电站编号 plantCode 作为系统中的 stationId，仅插入数据库中不存在的电站。 */ @Transactional(rollbackFor = Exception.class) public void syncHuaWeiDataTask() { log.info("-------1、开始同步华为电站列表信息----"); String token = huWeiUtils.getXsrfToken(); JSONArray stationArray = huWeiUtils.getStations(1, token); Set<String> stationIds = new HashSet<>(); if (CollectionUtils.isNotEmpty(stationArray)) { List<TStationInfo> insertList = new ArrayList<>(); huaWeiBuild(stationArray, stationIds, insertList); if (CollectionUtils.isNotEmpty(insertList)) { List<TStationInfo> existing = lambdaQuery().eq(TStationInfo::getPlatformId, 1L).in(TStationInfo::getStationId, stationIds).list(); Set<String> existingIds = existing.stream().map(TStationInfo::getStationId).collect(Collectors.toSet()); insertList.removeIf(item -> existingIds.contains(item.getStationId())); if (CollectionUtils.isNotEmpty(insertList)) { saveBatch(insertList); } } log.info("-------1、结束同步华为电站列表信息，本次新增{}条----", insertList.size()); } log.info("-------2、开始同步华为设备列表信息----"); if(CollectionUtils.isNotEmpty(stationIds)) { String codes = String.join(",", stationIds); JSONArray devList = huWeiUtils.getDevList(codes, token); collectService.saveCollectHuaWeiList(devList); } log.info("-------3、开始同步华为电站日数据信息----"); List<TStationInfo> existInfos = lambdaQuery().eq(TStationInfo::getPlatformId, 1L).list(); Set<String> existIds = existInfos.stream().map(TStationInfo::getStationId).collect(Collectors.toSet()); if(CollectionUtils.isNotEmpty(existIds)) { String codes = String.join(",", existIds); JSONArray stationDayList = huWeiUtils.getKpiStationDay(codes, System.currentTimeMillis(), token); stationEarnService.saveStationHuaWeiList(stationDayList); } log.info("-------3、结束同步华为电站日数据信息----"); } public void huaWeiBuild(JSONArray stationArray, Set<String> stationIds, List<TStationInfo> insertList) { for (Object item : stationArray) { JSONObject json = JSONObject.parseObject(item.toString()); String stationId = json.getString("plantCode"); if (StringUtils.isBlank(stationId) || !stationIds.add(stationId)) { continue; } TStationInfo info = new TStationInfo(); info.setPlatformId(1L); info.setStationId(stationId); info.setStationName(json.getString("plantName")); info.setStationType("1"); info.setRunStatus("0"); info.setInstalledCapacity(toBigDecimal(json, "capacity")); info.setContactName(json.getString("contactPerson")); info.setPhone(json.getString("contactMethod")); info.setLatitude(json.getString("latitude")); info.setLongitude(json.getString("longitude")); info.setAddress(json.getString("plantAddress")); LocalDateTime gridTime = parseHuaweiDateTime(json.getString("gridConnectionDate")); info.setDataTime(gridTime); info.setLoanStatus(gridTime != null && gridTime.isBefore(LocalDateTime.now()) ? 3 : 1); info.setGridStatus(info.getLoanStatus() == 3 ? "1" : "0"); info.setDataSource(2); info.setSyncTime(LocalDateTime.now()); info.setCreateTime(LocalDateTime.now()); info.setUpdateTime(LocalDateTime.now()); fillDefaults(info); insertList.add(info); } } } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/task/ExchangeCodeExpireTask.java
@@ -28,4 +28,14 @@ public void syncKsolarDataTask() { tStationInfoService.syncKsolarDataTask(); } /** * 每6小时执行科士达的数据同步 */ @Scheduled(cron = "0 0 */6 * * ?") public void syncHuaWeiDataTask() { tStationInfoService.syncHuaWeiDataTask(); } } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/config/UdanEmsKafkaProperties.java
New file @@ -0,0 +1,20 @@ package com.ruoyi.other.udan.config; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; /** Kafka connection settings issued by UDAN after integration approval. */ @Data @Component @ConfigurationProperties(prefix = "udan.ems.kafka") public class UdanEmsKafkaProperties { private boolean enabled = false; private String bootstrapServers; private String username; private String password; private String topic; private String groupId = "group-device-state"; private String autoOffsetReset = "latest"; private int pollSeconds = 3600; } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/PowerMeterState.java
New file @@ -0,0 +1,209 @@ package com.ruoyi.other.udan.model; import lombok.Data; /** * @author Lux * @version 1.0.0 * @since 2026/9/10 17:55 */ @Data public class PowerMeterState { /** 峰用电总价，单位：元 */ private double cumulativeImportAmountPeak; /** 平用电总价，单位：元 */ private double cumulativeImportAmountShoulder; /** 谷用电总价，单位：元 */ private double cumulativeImportAmountValley; /** 累计用电总价，单位：元 */ private double cumulativeImportAmount; /** 日用电总价，单位：元 */ private double dailyImportAmount; /** 月用电总价，单位：元 */ private double monthlyImportAmount; /** 当日收益，单位：元 */ private double dailyProfit; /** 昨日收益，单位：元 */ private double yesterdayProfit; /** 当月收益，单位：元 */ private double monthlyProfit; /** 上月收益，单位：元 */ private double lastMonthlyProfit; /** 累计收益，单位：元 */ private double cumulativeProfit; /** 累计尖用电总价，单位：元 */ private double cumulativeImportAmountSharp; /** 累计用电电量，单位：kWh */ private double cumulativeImportKwh; /** 累计尖用电电量，单位：kWh */ private double cumulativeImportKwhSharp; /** 累计峰用电电量，单位：kWh */ private double cumulativeImportKwhPeak; /** 累计平用电电量，单位：kWh */ private double cumulativeImportKwhShoulder; /** 累计谷用电电量，单位：kWh */ private double cumulativeImportKwhValley; /** 累计馈电电量，单位：kWh */ private double cumulativeExportKwh; /** 累计尖馈电电量，单位：kWh */ private double cumulativeExportKwhSharp; /** 累计峰馈电电量，单位：kWh */ private double cumulativeExportKwhPeak; /** 累计平馈电电量，单位：kWh */ private double cumulativeExportKwhShoulder; /** 累计谷馈电电量，单位：kWh */ private double cumulativeExportKwhValley; /** 月用电电量，单位：kWh */ private double monthlyImportKwh; /** 月尖用电电量，单位：kWh */ private double monthlyImportKwhSharp; /** 月峰用电电量，单位：kWh */ private double monthlyImportKwhPeak; /** 月平用电电量，单位：kWh */ private double monthlyImportKwhShoulder; /** 月谷用电电量，单位：kWh */ private double monthlyImportKwhValley; /** 月馈电电量，单位：kWh */ private double monthlyExportKwh; /** 月尖馈电电量，单位：kWh */ private double monthlyExportKwhSharp; /** 月峰馈电电量，单位：kWh */ private double monthlyExportKwhPeak; /** 月平馈电电量，单位：kWh */ private double monthlyExportKwhShoulder; /** 月谷馈电电量，单位：kWh */ private double monthlyExportKwhValley; /** 日用电电量，单位：kWh */ private float dailyImportKwh; /** 日尖用电电量，单位：kWh */ private float dailyImportKwhSharp; /** 日峰用电电量，单位：kWh */ private float dailyImportKwhPeak; /** 日平用电电量，单位：kWh */ private float dailyImportKwhShoulder; /** 日谷用电电量，单位：kWh */ private float dailyImportKwhValley; /** 日馈电电量，单位：kWh */ private float dailyExportKwh; /** 日尖馈电电量，单位：kWh */ private float dailyExportKwhSharp; /** 日峰馈电电量，单位：kWh */ private float dailyExportKwhPeak; /** 日平馈电电量，单位：kWh */ private float dailyExportKwhShoulder; /** 日谷馈电电量，单位：kWh */ private float dailyExportKwhValley; /** 累计馈电总价，单位：元 */ private double cumulativeExportAmount; /** 累计尖馈电总价，单位：元 */ private double cumulativeExportAmountSharp; /** 累计峰馈电总价，单位：元 */ private double cumulativeExportAmountPeak; /** 累计平馈电总价，单位：元 */ private double cumulativeExportAmountShoulder; /** 累计谷馈电总价，单位：元 */ private double cumulativeExportAmountValley; /** 月尖用电总价，单位：元 */ private double monthlyImportAmountSharp; /** 月峰用电总价，单位：元 */ private double monthlyImportAmountPeak; /** 月平用电总价，单位：元 */ private double monthlyImportAmountShoulder; /** 月谷用电总价，单位：元 */ private double monthlyImportAmountValley; /** 月馈电总价，单位：元 */ private double monthlyExportAmount; /** 月尖馈电总价，单位：元 */ private double monthlyExportAmountSharp; /** 月峰馈电总价，单位：元 */ private double monthlyExportAmountPeak; /** 月平馈电总价，单位：元 */ private double monthlyExportAmountShoulder; /** 月谷馈电总价，单位：元 */ private double monthlyExportAmountValley; /** 日尖用电总价，单位：元 */ private float dailyImportAmountSharp; /** 日峰用电总价，单位：元 */ private float dailyImportAmountPeak; /** 日平用电总价，单位：元 */ private float dailyImportAmountShoulder; /** 日谷用电总价，单位：元 */ private float dailyImportAmountValley; /** 日馈电总价，单位：元 */ private float dailyExportAmount; /** 日尖馈电总价，单位：元 */ private float dailyExportAmountSharp; /** 日峰馈电总价，单位：元 */ private float dailyExportAmountPeak; /** 日平馈电总价，单位：元 */ private float dailyExportAmountShoulder; /** 日谷馈电总价，单位：元 */ private float dailyExportAmountValley; /** 时间戳，1970-01-01 00:00:00 */ private long time; } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDatabaseRecord.java
New file @@ -0,0 +1,26 @@ 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; } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/model/UdanEmsDeviceState.java
New file @@ -0,0 +1,35 @@ package com.ruoyi.other.udan.model; import lombok.Data; import java.util.ArrayList; import java.util.List; /** 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; @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
New file @@ -0,0 +1,114 @@ 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.udan.model.UdanEmsDeviceState; import com.ruoyi.other.udan.util.UdanEmsKafkaUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import java.io.IOException; /** 将 iems-open-v1.0.5.proto 消息解析为业务对象，并映射为待入库对象。 */ @Slf4j @Component @ConditionalOnProperty(prefix = "udan.ems.kafka", name = "enabled", havingValue = "true") public class UdanEmsMessageProcessor { private final UdanEmsKafkaUtil kafkaUtil; /** 注入 Kafka 消费工具。 */ public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil) { this.kafkaUtil = kafkaUtil; } /** 注册消息回调，避免消费者只消费但没有业务处理。 */ //@PostConstruct public void register() { kafkaUtil.setMessageHandler(this::process); } /** 解析消息、转换待入库对象并输出日志；此处暂不调用数据库。 */ public void process(String emsSystemId, byte[] protobufBytes) { try { UdanEmsDeviceState state = parse(emsSystemId, protobufBytes); log.info("2、UDAN EMS实际对象: {}", JSON.toJSONString(state)); } catch (IOException e) { log.error("UDAN EMS protobuf解析失败: emsSystemId={}, bytes={}", emsSystemId, protobufBytes == null ? 0 : protobufBytes.length, e); } } /** 按顶层 IEmsAllDeviceState 字段号读取并构造真实业务对象。 */ 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; } /** 解析顶层各子消息，字段号来自 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); }); 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 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; } /** 遍历嵌套 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 { 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; } } @FunctionalInterface private interface FieldReader { void read(int field, int type, CodedInputStream input) throws IOException; } } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/udan/util/UdanEmsKafkaUtil.java
New file @@ -0,0 +1,321 @@ package com.ruoyi.other.udan.util; import com.google.protobuf.CodedInputStream; import com.google.protobuf.WireFormat; import com.ruoyi.other.udan.config.UdanEmsKafkaProperties; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.consumer.ConsumerConfig; 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.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; 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.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.io.IOException; /** * 优旦 EMS 平台 Kafka 消费工具。 * * <p>按照《UDAN-EMS云平台对平台协议》订阅设备监控 Topic：消息 Key 为 EMS 系统 ID， * 消息 Value 为 protobuf3 二进制数据，不是 JSON。业务方应注册消息处理器，并使用 * 客户最终确认版本的 iems-open proto 生成类反序列化消息。</p> */ @Component @ConditionalOnProperty(prefix = "udan.ems.kafka", name = "enabled", havingValue = "true") public class UdanEmsKafkaUtil { private static final Logger log = LoggerFactory.getLogger(UdanEmsKafkaUtil.class); private final UdanEmsKafkaProperties properties; private final ExecutorService executor = Executors.newSingleThreadExecutor(r -> { Thread t = new Thread(r, "udan-ems-kafka-consumer"); t.setDaemon(true); return t; }); private volatile KafkaConsumer<String, byte[]> consumer; private volatile boolean running; private volatile MessageHandler messageHandler = (key, payload) -> { }; private volatile ErrorHandler errorHandler = Throwable::printStackTrace; /** * 创建优旦 EMS Kafka 消费工具。 * * @param properties Kafka 连接、认证、Topic 和消费组配置 */ public UdanEmsKafkaUtil(UdanEmsKafkaProperties properties) { this.properties = properties; } /** * 校验配置并在独立守护线程中启动持续消费。 * 当 udan.ems.kafka.enabled=true 时由 Spring 在 Bean 初始化后自动调用。 * PostConstruct 创建并完成依赖注入后，自动执行这个方法一次 */ //@PostConstruct public void start() { validate(); running = true; executor.submit(this::consumeLoop); } /** * 注册设备监控消息处理器。 * * @param handler 接收 EMS 系统 ID 和 protobuf 原始字节的处理器；传入 null 时忽略消息 */ public void setMessageHandler(MessageHandler handler) { this.messageHandler = handler == null ? (key, payload) -> { } : handler; } /** * 注册消费或业务处理异常的处理器。 * * @param handler 异常处理器；传入 null 时恢复为打印异常堆栈 */ public void setErrorHandler(ErrorHandler handler) { this.errorHandler = handler == null ? Throwable::printStackTrace : handler; } /** * 按配置的超时时间轮询一次 Kafka，并将取得的消息交给消息处理器。 * * @return 本次轮询取得的消息数量 * @throws IllegalStateException 消费者尚未启动时抛出 */ public int 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(); } /** * 唤醒并关闭 Kafka 消费者，同时停止后台消费线程。 * 由 Spring 在 Bean 销毁前自动调用。 */ @PreDestroy public void stop() { running = false; if (consumer != null) consumer.wakeup(); executor.shutdownNow(); } /** * 创建消费者、订阅配置的设备监控 Topic，并持续执行轮询。 * WakeupException 在正常关闭时忽略，运行期间出现的异常交给异常处理器。 */ private void consumeLoop() { try (KafkaConsumer<String, byte[]> c = new KafkaConsumer<>(consumerConfig())) { 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 { consumer = null; } } /** * 逐条分发 Kafka 消息；单条消息处理失败不会阻断同批次后续消息。 * * @param records Kafka 本次轮询返回的消息集合 */ private void dispatch(ConsumerRecords<String, byte[]> records) { for (ConsumerRecord<String, byte[]> record : records) { byte[] payload = record.value() == null ? new byte[0] : record.value(); try { // 先输出可核验信息，再交给业务回调；线上 Value 是 protobuf 原始字节，不能当 JSON 打印。 log.info("1、UDAN EMS可读字段: emsSystemId={}, data={}", record.key(), toReadableProtobuf(payload)); // 此处即为业务接入点：解析 protobuf 后调用保存服务，当前仅由回调方决定如何处理。 messageHandler.onMessage(record.key(), payload); } catch (Throwable e) { errorHandler.onError(e); } } } /** * 将消息前若干字节转为十六进制，便于核对消息是否为稳定的二进制 protobuf 内容。 * * @param bytes 原始消息字节 * @param maxBytes 最多输出的字节数 * @return 大写十六进制字符串，截断时追加省略号 */ private static String toHex(byte[] bytes, int maxBytes) { StringBuilder result = new StringBuilder(); int length = Math.min(bytes.length, Math.max(0, maxBytes)); for (int i = 0; i < length; i++) { if (i > 0) result.append(' '); result.append(String.format("%02X", bytes[i] & 0xFF)); } if (bytes.length > length) result.append(" ..."); return result.toString(); } /** * 将 protobuf 二进制按 wire type 转为可读文本。由于未绑定生成的 proto 类，字段名显示为 fieldN； * 字符串、整数、浮点数和嵌套消息内容会被直接展开，便于联调确认实际收到的数据。 */ private static String toReadableProtobuf(byte[] bytes) { if (bytes == null || bytes.length == 0) return "{}"; try { return readReadableFields(CodedInputStream.newInstance(bytes), 0); } catch (IOException e) { return "<protobuf解析失败: " + e.getMessage() + ">"; } } /** 递归读取 protobuf 字段，限制递归深度和字段数量避免日志过大。 */ private static String readReadableFields(CodedInputStream input, int depth) throws IOException { if (depth > 3) return "<nested>"; StringBuilder text = new StringBuilder("{"); int count = 0; while (!input.isAtEnd() && count++ < 80) { int tag = input.readTag(); if (tag == 0) break; int field = WireFormat.getTagFieldNumber(tag); int type = WireFormat.getTagWireType(tag); if (text.length() > 1) text.append(", "); text.append("field").append(field).append('='); switch (type) { case WireFormat.WIRETYPE_VARINT: text.append(input.readInt64()); break; case WireFormat.WIRETYPE_FIXED64: text.append(input.readDouble()); break; case WireFormat.WIRETYPE_LENGTH_DELIMITED: byte[] data = input.readBytes().toByteArray(); if (isReadableString(data)) { text.append('"').append(escape(new String(data, java.nio.charset.StandardCharsets.UTF_8))).append('"'); } else { text.append(readNestedOrHex(data, depth)); } break; case WireFormat.WIRETYPE_FIXED32: text.append(input.readFloat()); break; default: text.append("<wireType=").append(type).append('>'); return text.append('}').toString(); } } if (count > 80) text.append(", ..."); return text.append('}').toString(); } /** 尝试将长度字段识别为嵌套 protobuf，否则显示前 32 字节十六进制。 */ private static String readNestedOrHex(byte[] data, int depth) { try { String nested = readReadableFields(CodedInputStream.newInstance(data), depth + 1); if (!nested.equals("{}")) return nested; } catch (IOException ignored) { // 不是嵌套 protobuf 时按原始二进制展示。 } return "0x" + toHex(data, 32); } /** 判断长度字段是否是可直接阅读的 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; } return true; } /** 转义日志中的双引号和反斜杠。 */ private static String escape(String value) { return value.replace("\\", "\\\\").replace("\"", "\\\""); } /** * 按协议示例构造 Kafka Consumer 配置。 * 固定使用 SASL_SSL、SCRAM-SHA-512、字符串 Key 和字节数组 Value。 * * @return 可直接用于创建 KafkaConsumer 的配置项 */ private Map<String, Object> consumerConfig() { Map<String, Object> config = new HashMap<>(); config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers()); config.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getGroupId()); config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, properties.getAutoOffsetReset()); config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL"); config.put("sasl.mechanism", "SCRAM-SHA-512"); config.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"" + properties.getUsername() + "\" password=\"" + properties.getPassword() + "\";"); return config; } /** * 解码协议文档第 4.2 节提供的 Base64 测试报文。 * 线上 Kafka Value 已经是 protobuf byte[]，不得调用此方法二次解码。 * * @param base64 文档中的 Base64 报文，可以包含换行 * @return protobuf 原始字节；输入为空时返回空数组 */ public static byte[] decodeDocumentSample(String base64) { if (!StringUtils.hasText(base64)) { return new byte[0]; } return Base64.getMimeDecoder().decode(base64); } /** * 校验连接和订阅所需的必填参数，防止以不完整配置启动消费者。 * * @throws IllegalStateException 缺少 Kafka 地址、用户名、密码或 Topic 时抛出 */ private void validate() { if (!StringUtils.hasText(properties.getBootstrapServers()) || !StringUtils.hasText(properties.getUsername()) || !StringUtils.hasText(properties.getPassword()) || !StringUtils.hasText(properties.getTopic())) { throw new IllegalStateException("udan.ems.kafka requires bootstrap-servers, username, password and topic"); } } /** 业务消息处理回调。 */ @FunctionalInterface public interface MessageHandler { /** * 处理一条优旦 EMS 设备监控消息。 * * @param emsSystemId Kafka Key，即 EMS 系统 ID * @param protobufBytes Kafka Value，即 protobuf3 原始字节 */ void onMessage(String emsSystemId, byte[] protobufBytes); } /** 消费和消息处理异常回调。 */ @FunctionalInterface public interface ErrorHandler { /** * 处理 Kafka 消费或业务消息处理过程中发生的异常。 * * @param error 捕获到的异常 */ void onError(Throwable error); } } ruoyi-service/ruoyi-other/src/main/java/com/ruoyi/other/util/HuWeiUtils.java
New file @@ -0,0 +1,151 @@ package com.ruoyi.other.util; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import lombok.extern.slf4j.Slf4j; import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.ResponseEntity; import org.springframework.web.client.RestTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.Collections; import java.util.List; /** * 华为：光伏数据对接 * * @author Lux * @version 1.0.0 * @since 2026/9/7 17:38 */ @Slf4j @Service public class HuWeiUtils { private static final String API_HOST = "https://cn.fusionsolar.huawei.com"; private static final String DEFAULT_USER_NAME = "mxxnycd"; private static final String DEFAULT_SYSTEM_CODE = "mxxny888888"; private static final String LOGIN_PATH = "/thirdData/login"; private static final String STATIONS_PATH = "/thirdData/stations"; private static final String DEV_LIST_PATH = "/thirdData/getDevList"; private static final String Kpi_STATION_DAY_PATH = "/thirdData/getKpiStationDay"; @Resource private RestTemplate restTemplate; /** 登录华为 FusionSolar 并取得 XSRF-TOKEN。 */ public String getXsrfToken() { return getXsrfToken(DEFAULT_USER_NAME, DEFAULT_SYSTEM_CODE); } /** 使用指定账号登录并取得 XSRF-TOKEN。 */ public String getXsrfToken(String userName, String systemCode) { JSONObject body = new JSONObject(); body.put("userName", userName); body.put("systemCode", systemCode); HttpHeaders headers = new HttpHeaders(); headers.set(HttpHeaders.CONTENT_TYPE, "application/json"); ResponseEntity<String> response = restTemplate.exchange(API_HOST + LOGIN_PATH, HttpMethod.POST, new HttpEntity<>(body.toJSONString(), headers), String.class); String token = response.getHeaders().getFirst("XSRF-TOKEN"); if (token == null) { List<String> cookies = response.getHeaders().get(HttpHeaders.SET_COOKIE); if (cookies != null) { for (String cookie : cookies) { if (cookie.startsWith("XSRF-TOKEN=")) { token = cookie.substring("XSRF-TOKEN=".length()).split(";", 2)[0]; break; } } } } if (token == null || token.trim().isEmpty()) { throw new IllegalStateException("华为登录未返回XSRF-TOKEN"); } return token; } /** * 获取电站列表 * * @param pageNo 当前页 * @param token 登陆后的token * @return 列表数据 */ public JSONArray getStations(Integer pageNo, String token) { HttpHeaders headers = new HttpHeaders(); headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON); headers.setAccept(Collections.singletonList(org.springframework.http.MediaType.APPLICATION_JSON)); headers.set("XSRF-TOKEN", token); headers.set(HttpHeaders.COOKIE, "XSRF-TOKEN=" + token); JSONObject body = new JSONObject(); body.put("pageNo", pageNo == null || pageNo < 1 ? 1 : pageNo); ResponseEntity<String> response = restTemplate.exchange(API_HOST + STATIONS_PATH, HttpMethod.POST, new HttpEntity<>(body.toJSONString(), headers), String.class); JSONObject result = JSON.parseObject(response.getBody()); if (result == null || !Boolean.TRUE.equals(result.getBoolean("success"))) { throw new IllegalStateException("华为电站列表接口调用失败：" + (result == null ? "空响应" : result.getString("message"))); } JSONObject data = result.getJSONObject("data"); return data == null || data.getJSONArray("list") == null ? new JSONArray() : data.getJSONArray("list"); } /** * 获取设备列表 * * @param stationCodes 电站ID,多个用逗号分隔 * @param token 登陆后的token * @return 列表数据 */ public JSONArray getDevList(String stationCodes, String token) { HttpHeaders headers = new HttpHeaders(); headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON); headers.setAccept(Collections.singletonList(org.springframework.http.MediaType.APPLICATION_JSON)); headers.set("XSRF-TOKEN", token); headers.set(HttpHeaders.COOKIE, "XSRF-TOKEN=" + token); JSONObject body = new JSONObject(); body.put("stationCodes", stationCodes); ResponseEntity<String> response = restTemplate.exchange(API_HOST + DEV_LIST_PATH, HttpMethod.POST, new HttpEntity<>(body.toJSONString(), headers), String.class); JSONObject result = JSON.parseObject(response.getBody()); if (result == null || !Boolean.TRUE.equals(result.getBoolean("success"))) { throw new IllegalStateException("华为设备列表接口调用失败：" + (result == null ? "空响应" : result.getString("message"))); } JSONArray dataList = result.getJSONArray("data"); return dataList == null ? new JSONArray() : dataList; } /** * 获取设备列表 * * @param stationCodes 电站ID,多个用逗号分隔 * @param token 登陆后的token * @return 列表数据 */ public JSONArray getKpiStationDay(String stationCodes, Long collectTime, String token) { HttpHeaders headers = new HttpHeaders(); headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON); headers.setAccept(Collections.singletonList(org.springframework.http.MediaType.APPLICATION_JSON)); headers.set("XSRF-TOKEN", token); headers.set(HttpHeaders.COOKIE, "XSRF-TOKEN=" + token); JSONObject body = new JSONObject(); body.put("stationCodes", stationCodes); body.put("collectTime", collectTime); ResponseEntity<String> response = restTemplate.exchange(API_HOST + Kpi_STATION_DAY_PATH, HttpMethod.POST, new HttpEntity<>(body.toJSONString(), headers), String.class); JSONObject result = JSON.parseObject(response.getBody()); if (result == null || !Boolean.TRUE.equals(result.getBoolean("success"))) { throw new IllegalStateException("华为电站列表接口调用失败：" + (result == null ? "空响应" : result.getString("message"))); } JSONArray dataList = result.getJSONArray("data"); return dataList == null ? new JSONArray() : dataList; } } ruoyi-service/ruoyi-other/src/main/resources/sql/t_screen_content_alter.sql
New file @@ -0,0 +1,4 @@ -- ---------------------------- -- 增加大屏名称配置字段 alter table t_screen_content add column screen_name varchar(255) default 'NULL' null comment '大屏名称';
