增加 dds 数据解析

ems-admin/src/main/java/com/xzzn/web/controller/ems/MqttMessageController.java

@@ -1,6 +1,7 @@
 package com.xzzn.web.controller.ems;
 
 import com.xzzn.ems.domain.EmsMqttMessage;
+import com.xzzn.ems.service.IDDSDataProcessService;
 import com.xzzn.ems.service.IEmsMqttMessageService;
 import com.xzzn.ems.service.IFXXDataProcessService;
 import com.xzzn.framework.manager.MqttLifecycleManager;
@@ -31,6 +32,8 @@ public class MqttMessageController implements MqttPublisher, MqttSubscriber {
     @Autowired
     private IFXXDataProcessService fXXDataProcessService;
 
+    @Autowired
+    private IDDSDataProcessService dDSDataProcessService;
     @Autowired
     public MqttMessageController(MqttLifecycleManager mqttLifecycleManager) {
         this.mqttLifecycleManager = mqttLifecycleManager;
@@ -38,12 +41,18 @@ public class MqttMessageController implements MqttPublisher, MqttSubscriber {
 
     @PostConstruct
     public void init() {
-        // 订阅系统状态主题
+        // 订阅奉贤系统状态主题
         subscribe("021_FXX_01_UP", 1, this::handleDeviceData);
         subscribe("021_FXX_01_RECALL", 1, this::handleDeviceData);
         subscribe("021_FXX_01_DOWN", 1, this::handleDeviceData);
         subscribe("021_FXX_01", 1, this::handleSystemStatus);
 
+        // 订阅电动所系统状态主题
+        subscribe("021_DDS_01_UP", 1, this::handleDeviceData);
+        subscribe("021_DDS_01_RECALL", 1, this::handleDeviceData);
+        subscribe("021_DDS_01_DOWN", 1, this::handleDeviceData);
+        subscribe("021_DDS_01", 1, this::handleSystemStatus);
+
     }
 
     // 处理系统状态消息
@@ -72,7 +81,11 @@ public class MqttMessageController implements MqttPublisher, MqttSubscriber {
         System.out.println("[DEVICE] data: " + payload);
         try {
             // 业务处理逻辑
-            fXXDataProcessService.handleFxData(payload);
+            if (topic.startsWith("021_DDS")) {
+                dDSDataProcessService.handleDdsData(payload);
+            } else if (topic.startsWith("021_FXX")) {
+                fXXDataProcessService.handleFxData(payload);
+            }
 
 
             EmsMqttMessage mqttMessage = new EmsMqttMessage();

ems-system/src/main/java/com/xzzn/ems/service/IDDSDataProcessService.java

@@ -0,0 +1,7 @@
+package com.xzzn.ems.service;
+
+public interface IDDSDataProcessService {
+
+    public void handleDdsData(String message);
+
+}

ems-system/src/main/java/com/xzzn/ems/service/impl/DDSDataProcessServiceImpl.java

@@ -0,0 +1,646 @@
+package com.xzzn.ems.service.impl;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import com.alibaba.fastjson2.TypeReference;
+import com.xzzn.common.constant.RedisKeyConstants;
+import com.xzzn.common.core.redis.RedisCache;
+import com.xzzn.common.enums.*;
+import com.xzzn.common.utils.DateUtils;
+import com.xzzn.common.utils.StringUtils;
+import com.xzzn.ems.domain.*;
+import com.xzzn.ems.mapper.*;
+import com.xzzn.ems.service.IDDSDataProcessService;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.*;
+
+@Service
+public class DDSDataProcessServiceImpl implements IDDSDataProcessService {
+    private static final Log log = LogFactory.getLog(DDSDataProcessServiceImpl.class);
+    private static final String SITE_ID = "021_DDS_01";
+    @Autowired
+    private EmsBatteryClusterMapper emsBatteryClusterMapper;
+
+    @Autowired
+    private EmsBatteryStackMapper emsBatteryStackMapper;
+
+    @Autowired
+    private EmsBatteryDataMapper emsBatteryDataMapper;
+
+    @Autowired
+    private EmsPcsDataMapper emsPcsDataMapper;
+
+    @Autowired
+    private EmsPcsBranchDataMapper emsPcsBranchDataMapper;
+
+    @Autowired
+    private RedisCache redisCache;
+
+    @Autowired
+    private EmsDevicesSettingMapper emsDevicesSettingMapper;
+
+    @Autowired
+    private EmsAmmeterDataMapper emsAmmeterDataMapper;
+
+    @Override
+    public void handleDdsData(String message) {
+        JSONArray arraylist = JSONArray.parseArray(message);
+
+        for (int i = 0; i < arraylist.size(); i++) {
+            JSONObject obj = JSONObject.parseObject(arraylist.get(i).toString());
+
+            String deviceId = obj.get("Device").toString();
+            String jsonData = obj.get("Data").toString();
+
+            log.info("deviceId:" + deviceId);
+            if (deviceId.contains("BMSD")) {
+                batteryStackDataProcess(deviceId, jsonData);
+
+            } else if (deviceId.contains("BMSC")) {
+                log.info("BMSC data:"+ jsonData);
+                batteryClusterDataProcess(deviceId, jsonData);
+                batteryDataProcess(deviceId, jsonData);
+
+            } else if (deviceId.contains("PCS")) {
+                pcsDataProcess(deviceId, jsonData);
+                pcsBranchDataProcess(deviceId, jsonData);
+            } else if (deviceId.contains("LOAD")) {
+                loadDataProcess(deviceId, jsonData);
+            }   else if (deviceId.contains("METE")) {
+                meteDataProcess(deviceId, jsonData);
+            }
+        }
+    }
+
+    private void batteryStackDataProcess(String deviceId, String dataJson) {
+
+        //电池堆
+        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
+        });
+
+        //BMS 电池簇
+        EmsBatteryStack dataStack = new EmsBatteryStack();
+
+        // 其他非 BigDecimal 字段
+        dataStack.setWorkStatus(WorkStatus.NORMAL.getCode()); // 或其他默认值
+        dataStack.setPcsCommunicationStatus(CommunicationStatus.OK.getCode());
+        dataStack.setEmsCommunicationStatus(CommunicationStatus.OK.getCode());
+
+        // 电池堆状态数据设置
+        dataStack.setOperationStatus(StringUtils.getString(obj.get("DCZT")));
+        dataStack.setStackVoltage(StringUtils.getBigDecimal(obj.get("DCDDY")));
+        dataStack.setStackCurrent(StringUtils.getBigDecimal(obj.get("DCDDL")));
+        dataStack.setStackSoc(StringUtils.getBigDecimal(obj.get("DCDSOC")));
+        dataStack.setStackSoh(StringUtils.getBigDecimal(obj.get("DCDSOH")));
+
+        // 电压极值信息
+        dataStack.setMaxCellVoltage(StringUtils.getBigDecimal(obj.get("ZGDCDY")));
+        dataStack.setMaxVoltageGroupId(StringUtils.getLong(obj.get("ZGDCDYZH")));
+        dataStack.setMaxVoltageCellId(StringUtils.getLong(obj.get("ZGDCDYZHDH")));
+        dataStack.setMinCellVoltage(StringUtils.getBigDecimal(obj.get("ZDDCDY")));
+        dataStack.setMinVoltageGroupId(StringUtils.getLong(obj.get("ZDDCDYZH")));
+        dataStack.setMinVoltageCellId(StringUtils.getLong(obj.get("ZDDCDYZHDH")));
+
+        // 温度极值信息
+        dataStack.setMaxCellTemp(StringUtils.getBigDecimal(obj.get("ZGDCWD")));
+        dataStack.setMaxTempGroupId(StringUtils.getLong(obj.get("ZGDCWDZH")));
+        dataStack.setMaxTempCellId(StringUtils.getLong(obj.get("ZGDCWDZHDH")));
+        dataStack.setMinCellTemp(StringUtils.getBigDecimal(obj.get("ZDDCWD")));
+        dataStack.setMinTempGroupId(StringUtils.getLong(obj.get("ZDDCWDZH")));
+        dataStack.setMinTempCellId(StringUtils.getLong(obj.get("ZDDCWDZHDH")));
+
+        // 电量统计信息
+        dataStack.setTotalChargeCapacity(StringUtils.getBigDecimal(obj.get("DLJCDDL")));
+        dataStack.setTotalDischargeCapacity(StringUtils.getBigDecimal(obj.get("DLCFDDL")));
+        dataStack.setSessionChargeCapacity(StringUtils.getBigDecimal(obj.get("DDCLJCDDL")));
+        dataStack.setSessionDischargeCapacity(StringUtils.getBigDecimal(obj.get("DDCLJFDDL")));
+        dataStack.setAvailableChargeCapacity(StringUtils.getBigDecimal(obj.get("DKCDL")));
+        dataStack.setAvailableDischargeCapacity(StringUtils.getBigDecimal(obj.get("DKFDL")));
+
+// 时间信息
+        dataStack.setRemainingDischargeTime(StringUtils.getLong(obj.get("KYFDSJ")));
+        dataStack.setRemainingChargeTime(StringUtils.getLong(obj.get("KYCDSJ")));
+
+// 功率/电流限制
+        dataStack.setMaxDischargePower(StringUtils.getBigDecimal(obj.get("YXZDFDGL")));
+        dataStack.setMaxChargePower(StringUtils.getBigDecimal(obj.get("YXZDCDGL")));
+        dataStack.setMaxDischargeCurrent(StringUtils.getBigDecimal(obj.get("YXZDFDDL")));
+        dataStack.setMaxChargeCurrent(StringUtils.getBigDecimal(obj.get("YXZDCDDL")));
+
+        // 当日统计
+        dataStack.setDailyDischargeCycles(StringUtils.getLong(obj.get("DTFDCS")));
+        dataStack.setDailyChargeCycles(StringUtils.getLong(obj.get("DTCDCS")));
+        dataStack.setDailyDischargeCapacity(StringUtils.getBigDecimal(obj.get("DTFDDL")));
+        dataStack.setDailyChargeCapacity(StringUtils.getBigDecimal(obj.get("DTCDDL")));
+
+        // 系统状态
+        dataStack.setOperatingTemp(StringUtils.getBigDecimal(obj.get("YXWD")));
+        dataStack.setBmsStatus(StringUtils.getString(obj.get("BMSDDQZT")));
+        dataStack.setBmsChargeStatus(StringUtils.getString(obj.get("BMSCFDZT")));
+        dataStack.setStackInsulationResistance(StringUtils.getBigDecimal(obj.get("DCDJYDZ")));
+
+
+        dataStack.setCreateBy("system");
+        dataStack.setCreateTime(DateUtils.getNowDate());
+        dataStack.setUpdateBy("system");
+        dataStack.setUpdateTime(DateUtils.getNowDate());
+        dataStack.setSiteId(SITE_ID);
+        dataStack.setDeviceId(deviceId);
+
+        emsBatteryStackMapper.insertEmsBatteryStack(dataStack);
+
+        redisCache.setCacheObject(RedisKeyConstants.STACK + SITE_ID + "_" +deviceId, dataStack);
+    }
+
+
+    private void batteryClusterDataProcess(String deviceId, String dataJson) {
+
+        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
+        });
+        EmsDevicesSetting joken = new EmsDevicesSetting();
+        joken.setDeviceId(deviceId);
+        List<EmsDevicesSetting> up = emsDevicesSettingMapper.selectEmsDevicesSettingList(joken);
+        String stackDeviceId = "";
+        if (up != null && up.size() >0) {
+            stackDeviceId = up.get(0).getParentId();
+            if (stackDeviceId == null || stackDeviceId.isEmpty()) {
+                stackDeviceId = "1";
+            }
+        }
+        //BMS 电池簇
+        EmsBatteryCluster data = new EmsBatteryCluster();
+        // 设置所有 BigDecimal 类型字段为 ZERO
+        data.setChargeableCapacity(StringUtils.getBigDecimal(obj.get("KCDL")));
+        data.setClusterVoltage(StringUtils.getBigDecimal(obj.get("ZDY")));
+        data.setClusterCurrent(StringUtils.getBigDecimal(obj.get("ZDL")));
+        data.setTotalChargedCapacity(StringUtils.getBigDecimal(obj.get("LJCDDL")));
+        data.setDischargeableCapacity(StringUtils.getBigDecimal(obj.get("KFDL")));
+        data.setTotalDischargedCapacity(StringUtils.getBigDecimal(obj.get("LJFDDL")));
+        data.setSoh(StringUtils.getBigDecimal(obj.get("ZSOH")));
+        data.setAverageTemperature(StringUtils.getBigDecimal(obj.get("MKWD")));
+        data.setInsulationResistance(StringUtils.getBigDecimal(obj.get("ZJYDZ")));
+        data.setCurrentSoc(StringUtils.getBigDecimal(obj.get("ZSOC")));
+        data.setMaxAllowedChargePower(StringUtils.getBigDecimal(obj.get("YXCDZDGL")));
+        data.setMaxAllowedDischargePower(StringUtils.getBigDecimal(obj.get("YXFDZDGL")));
+        data.setMaxAllowedChargeVoltage(StringUtils.getBigDecimal(obj.get("YXCDZDDY")));
+        data.setMaxAllowedDischargeVoltage(StringUtils.getBigDecimal(obj.get("YXFDZDDY")));
+        data.setMaxAllowedChargeCurrent(StringUtils.getBigDecimal(obj.get("YXCDZDDL")));
+        data.setMaxAllowedDischargeCurrent(StringUtils.getBigDecimal(obj.get("YXFDZDDL")));
+        data.setBatteryPackVoltage(StringUtils.getBigDecimal(obj.get("ZDY")));
+        data.setBatteryPackCurrent(StringUtils.getBigDecimal(obj.get("ZDL")));
+        data.setBatteryPackTemp(StringUtils.getBigDecimal(obj.get("MKWD")));
+        data.setBatteryPackSoc(StringUtils.getBigDecimal(obj.get("ZSOC")));
+        data.setBatteryPackSoh(StringUtils.getBigDecimal(obj.get("ZSOH")));
+        data.setBatteryPackInsulationResistance(StringUtils.getBigDecimal(obj.get("ZJYDZ")));
+        data.setAvgCellVoltage(StringUtils.getBigDecimal(obj.get("PJDTDY")));
+        data.setAvgCellTemp(StringUtils.getBigDecimal(obj.get("PJDTWD")));
+        data.setMaxCellVoltage(StringUtils.getBigDecimal(obj.get("ZGDTDY")));
+        data.setMinCellVoltage(StringUtils.getBigDecimal(obj.get("ZDDTDY")));
+        data.setMaxCellTemp(StringUtils.getBigDecimal(obj.get("ZGDTWD")));
+        data.setMinCellTemp(StringUtils.getBigDecimal(obj.get("ZDDTWD")));
+        data.setMaxCellSoc(StringUtils.getBigDecimal(obj.get("ZGDTSOC")));
+        data.setMinCellSoc(StringUtils.getBigDecimal(obj.get("ZDDTSOC")));
+        data.setMaxCellSoh(StringUtils.getBigDecimal(obj.get("ZGDTSOH")));
+        data.setMinCellSoh(StringUtils.getBigDecimal(obj.get("ZDDTSOH")));
+        data.setTotalChargeEnergy(StringUtils.getBigDecimal(obj.get("DCLJCDDL")));
+        data.setTotalDischargeEnergy(StringUtils.getBigDecimal(obj.get("DCLJFDDL")));
+
+        // 其他非 BigDecimal 字段
+        data.setWorkStatus(WorkStatus.NORMAL.getCode()); // 或其他默认值
+        data.setPcsCommunicationStatus(CommunicationStatus.OK.getCode());
+        data.setEmsCommunicationStatus(CommunicationStatus.OK.getCode());
+        data.setCreateBy("system");
+        data.setCreateTime(DateUtils.getNowDate());
+        data.setUpdateBy("system");
+        data.setUpdateTime(DateUtils.getNowDate());
+        data.setSiteId(SITE_ID);
+        data.setDeviceId(deviceId);
+        data.setMaxCellVoltageId(StringUtils.getLong(obj.get("ZGDTDYDYD")));
+        data.setMinCellVoltageId(StringUtils.getLong(obj.get("ZDDTDYDYD")));
+        data.setMaxCellTempId(StringUtils.getLong(obj.get("ZGDTWDDYD")));
+        data.setMinCellTempId(StringUtils.getLong(obj.get("ZDDTWDDYD")));
+        data.setMaxCellSocId(StringUtils.getLong(obj.get("ZGDTSOCDYD")));
+        data.setMinCellSocId(StringUtils.getLong(obj.get("ZDDTSOCDYD")));
+        data.setMaxCellSohId(StringUtils.getLong(obj.get("ZGDTSOHDYD")));
+        data.setMinCellSohId(StringUtils.getLong(obj.get("ZDDTSOHDYD")));
+        if (StringUtils.isNotBlank(stackDeviceId)) {
+            data.setStackDeviceId(stackDeviceId);
+        } else {
+            data.setStackDeviceId("1");
+        }
+        emsBatteryClusterMapper.insertEmsBatteryCluster(data);
+
+        redisCache.setCacheObject(RedisKeyConstants.CLUSTER + SITE_ID + "_" +deviceId, data);
+
+    }
+
+    private void batteryDataProcess(String deviceId, String dataJson) {
+        EmsDevicesSetting joken = new EmsDevicesSetting();
+        joken.setDeviceId(deviceId);
+        List<EmsDevicesSetting> up = emsDevicesSettingMapper.selectEmsDevicesSettingList(joken);
+        String stackDeviceId = "";
+        if (up != null && up.size() >0) {
+            stackDeviceId = up.get(0).getParentId();
+            if (stackDeviceId == null || stackDeviceId.isEmpty()) {
+                stackDeviceId = "1";
+            }
+        }
+        //单体电池
+        Map<String, Map<String, Object>> records = processData(JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {}));
+        List<EmsBatteryData> list = new ArrayList<>();
+        //单体电池
+        for (Map.Entry<String, Map<String, Object>> record : records.entrySet()) {
+            String recordId = record.getKey();
+            Map<String, Object> fields = record.getValue();
+
+            EmsBatteryData batteryData = new EmsBatteryData();
+            batteryData.setDeviceId(recordId);
+            batteryData.setBatteryCellId(recordId);
+            batteryData.setSoc(StringUtils.getBigDecimal(fields.get("DTSOC")));
+            batteryData.setSoh(StringUtils.getBigDecimal(fields.get("DTSOH")));
+            batteryData.setTemperature(StringUtils.getBigDecimal(fields.get("DTWD")));
+            batteryData.setVoltage(StringUtils.getBigDecimal(fields.get("DTDY")));
+
+            batteryData.setBatteryCluster(deviceId);
+            batteryData.setBatteryPack(stackDeviceId);
+
+            // 时间戳
+            batteryData.setDataTimestamp(new Date());
+
+            // ID字段
+            batteryData.setSiteId(SITE_ID);
+            batteryData.setClusterDeviceId(deviceId);
+
+            list.add(batteryData);
+        }
+        if (list.size() > 0 ) {
+            emsBatteryDataMapper.insertEmsBatteryDataList(list);
+
+            redisCache.deleteList(RedisKeyConstants.BATTERY + SITE_ID + "_" +deviceId);
+            redisCache.setCacheList(RedisKeyConstants.BATTERY + SITE_ID + "_" +deviceId, list);
+        }
+    }
+
+
+    private void pcsDataProcess(String deviceId, String dataJson) {
+        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
+        });
+        //pcs
+        EmsPcsData pcsData = new EmsPcsData();
+        // 时间与状态类字段
+        pcsData.setDataUpdateTime(new Date());
+        pcsData.setWorkStatus(WorkStatus.NORMAL.getCode());
+        pcsData.setGridStatus(GridStatus.GRID.getCode());
+        pcsData.setDeviceStatus(DeviceStatus.ONLINE.getCode());
+        pcsData.setControlMode(ControlModeStatus.REMOTE.getCode());
+
+        // 功率与能量类字段
+        pcsData.setTotalActivePower(StringUtils.getBigDecimal(obj.get("YGGL")));
+        pcsData.setDailyAcChargeEnergy(StringUtils.getBigDecimal(obj.get("RCDL")));
+        pcsData.setTotalReactivePower(StringUtils.getBigDecimal(obj.get("XTWGGL")));
+        pcsData.setDailyAcDischargeEnergy(StringUtils.getBigDecimal(obj.get("RFDL")));
+        pcsData.setTotalApparentPower(StringUtils.getBigDecimal(obj.get("XTSZGL")));
+        pcsData.setTotalPowerFactor(StringUtils.getBigDecimal(obj.get("GLYS")));
+        pcsData.setDcPower(StringUtils.getBigDecimal(obj.get("XTSZGL")));
+        pcsData.setTotalAcChargeEnergy(StringUtils.getBigDecimal(obj.get("ZCDL")));
+        pcsData.setTotalAcDischargeEnergy(StringUtils.getBigDecimal(obj.get("ZFDL")));
+        pcsData.setAcChargeActivePower(StringUtils.getBigDecimal(obj.get("JLCCDYGGL")));
+        pcsData.setAcCapacitiveReactivePower(StringUtils.getBigDecimal(obj.get("JLCRXWGGL")));
+        pcsData.setAcDischargeActivePower(StringUtils.getBigDecimal(obj.get("JLCFDYGGL")));
+        pcsData.setAcInductiveReactivePower(StringUtils.getBigDecimal(obj.get("JLCGXWGGL")));
+        pcsData.setMaxCapacitivePowerCapacity(StringUtils.getBigDecimal(obj.get("ZDRXWGNL")));
+        pcsData.setMaxInductivePowerCapacity(StringUtils.getBigDecimal(obj.get("ZDGXWGNL")));
+        pcsData.setMaxChargePowerCapacity(StringUtils.getBigDecimal(obj.get("ZDKCGL")));
+        pcsData.setMaxDischargePowerCapacity(StringUtils.getBigDecimal(obj.get("ZDKFGL")));
+
+
+        // 温度与环境参数
+//        pcsData.setPcsModuleTemperature(StringUtils.getBigDecimal(obj.get("ChargeableCapacity")));
+//        pcsData.setPcsEnvironmentTemperature(StringUtils.getBigDecimal(obj.get("ChargeableCapacity")));
+//        pcsData.setAcFrequency(StringUtils.getBigDecimal(obj.get("ChargeableCapacity")));
+
+        // 状态指示类
+        pcsData.setBranchStatus(BranchStatus.NORMAL.getCode());
+        pcsData.setDischargeStatus(ChargeStatus.CHARGING.getCode());
+        String acSwitchStatus = StringUtils.getString(obj.get("JLKGZT"));
+        pcsData.setAcSwitchStatus(SwitchStatus.CLOSED.getCode());
+        String dcSwitchStatus = StringUtils.getString(obj.get("ZLKGZT"));
+        pcsData.setDcSwitchStatus(SwitchStatus.CLOSED.getCode());
+        String controlMode = StringUtils.getString(obj.get("YCTT"));
+        pcsData.setRemoteControlStatus(ControlModeStatus.REMOTE.getCode());
+
+        // 电流参数
+        pcsData.setSysUCurrent(StringUtils.getBigDecimal(obj.get("XTSCUXDL")));
+        pcsData.setSysVCurrent(StringUtils.getBigDecimal(obj.get("XTSCVXDL")));
+        pcsData.setSysWCurrent(StringUtils.getBigDecimal(obj.get("XTSCWXDL")));
+
+        // 直流参数
+//        pcsData.setDcVoltage(StringUtils.getBigDecimal(obj.get("ChargeableCapacity")));
+//        pcsData.setDcCurrent(StringUtils.getBigDecimal(obj.get("ChargeableCapacity")));
+
+        // 系统管理字段
+        pcsData.setCreateBy("system");
+        pcsData.setCreateTime(DateUtils.getNowDate());
+        pcsData.setUpdateBy("system");
+        pcsData.setUpdateTime(DateUtils.getNowDate());
+        pcsData.setSiteId(SITE_ID);
+        pcsData.setDeviceId(deviceId);
+        pcsData.setDateMonth(DateUtils.getNowMonthLong());
+        pcsData.setDateDay(DateUtils.getNowDayLong());
+
+        emsPcsDataMapper.insertEmsPcsData(pcsData);
+        redisCache.setCacheObject(RedisKeyConstants.PCS + SITE_ID + "_" +deviceId, pcsData);
+
+    }
+
+    private void pcsBranchDataProcess(String deviceId, String dataJson) {
+
+        Map<String, Map<String, Object>> records = processDataPrefix(JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {}));
+        List<EmsPcsBranchData> list = new ArrayList<>();
+
+        //PCS支路
+        for (Map.Entry<String, Map<String, Object>> record : records.entrySet()) {
+            String recordId = record.getKey();
+            Map<String, Object> fields = record.getValue();
+
+            EmsPcsBranchData data = new EmsPcsBranchData();
+            data.setDeviceId(deviceId);
+            data.setGridStatus(GridStatus.GRID.getCode());
+            data.setDcPower(StringUtils.getBigDecimal(fields.get("ZLGL")));
+            data.setDcVoltage(StringUtils.getBigDecimal(fields.get("ZLDY")));
+            data.setDcCurrent(StringUtils.getBigDecimal(fields.get("ZLDL")));
+            data.setGridUVoltage(StringUtils.getBigDecimal(fields.get("DWVXDY")));
+            data.setGridVVoltage(StringUtils.getBigDecimal(fields.get("DWUXDY")));
+            data.setGridWVoltage(StringUtils.getBigDecimal(fields.get("DWWXDY")));
+            data.setOutputUCurrent(StringUtils.getBigDecimal(fields.get("SCUXDL")));
+            data.setOutputVCurrent(StringUtils.getBigDecimal(fields.get("SCVXDL")));
+            data.setOutputWCurrent(StringUtils.getBigDecimal(fields.get("SCWXDL")));
+            data.setApparentPower(StringUtils.getBigDecimal(fields.get("SZGL")));
+            data.setActivePower(StringUtils.getBigDecimal(fields.get("YGGL")));
+            data.setReactivePower(StringUtils.getBigDecimal(fields.get("WGGL")));
+            data.setPowerFactor(StringUtils.getBigDecimal(fields.get("GLYS")));
+            data.setFrequency(StringUtils.getBigDecimal(fields.get("PL")));
+            data.setInternalTemp(StringUtils.getBigDecimal(fields.get("DY1WD")));
+            data.setuIgbtTemp(StringUtils.getBigDecimal(fields.get("UXIGBTWD")));
+            data.setvIgbtTemp(StringUtils.getBigDecimal(fields.get("VXIGBTWD")));
+            data.setwIgbtTemp(StringUtils.getBigDecimal(fields.get("WXIGBTWD")));
+            data.setAvailablePower(StringUtils.getBigDecimal(fields.get("KYGL")));
+            data.setTotalLoadRatio(StringUtils.getBigDecimal(fields.get("ZFZB")));
+            data.setAcLeakageCurrent(StringUtils.getBigDecimal(fields.get("JLLDL")));
+            data.setInsulationResistance(StringUtils.getBigDecimal(fields.get("JYZK")));
+            data.setBranchId(recordId);
+            list.add(data);
+        }
+        if (list.size() > 0 ) {
+            emsPcsBranchDataMapper.insertPcsBranchDataList(list);
+
+            redisCache.setCacheObject(RedisKeyConstants.BRANCH + SITE_ID + "_" +deviceId, list);
+
+        }
+
+    }
+
+    private void loadDataProcess(String deviceId, String dataJson) {
+
+        //总表
+        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
+        });
+
+        EmsAmmeterData dataLoad = new EmsAmmeterData();
+        // 更新时间
+        dataLoad.setDataUpdateTime(new Date());
+
+        // 电能设置-组合有功
+        dataLoad.setCurrentCombActiveTotal(StringUtils.getBigDecimal(obj.get("DQZHYGZDN")));
+        dataLoad.setCurrentCombActivePeak(StringUtils.getBigDecimal(obj.get("DQZHYGJDN")));
+        dataLoad.setCurrentCombActiveHigh(StringUtils.getBigDecimal(obj.get("DQZHYGFDN")));
+        dataLoad.setCurrentCombActiveFlat(StringUtils.getBigDecimal(obj.get("DQZHYGPDN")));
+        dataLoad.setCurrentCombActiveValley(StringUtils.getBigDecimal(obj.get("DQZHYGGDN")));
+        // 电能设置-正向有功
+        dataLoad.setCurrentForwardActiveTotal(StringUtils.getBigDecimal(obj.get("DQZXYGZDN")));
+        dataLoad.setCurrentForwardActivePeak(StringUtils.getBigDecimal(obj.get("DQZXYGJDN")));
+        dataLoad.setCurrentForwardActiveHigh(StringUtils.getBigDecimal(obj.get("DQZXYGFDN")));
+        dataLoad.setCurrentForwardActiveFlat(StringUtils.getBigDecimal(obj.get("DQZXYGPDN")));
+        dataLoad.setCurrentForwardActiveValley(StringUtils.getBigDecimal(obj.get("DQZXYGGDN")));
+        // 电能设置-反向有功
+        dataLoad.setCurrentReverseActiveTotal(StringUtils.getBigDecimal(obj.get("DQFXYGZDN")));
+        dataLoad.setCurrentReverseActivePeak(StringUtils.getBigDecimal(obj.get("DQFXYGJDN")));
+        dataLoad.setCurrentReverseActiveHigh(StringUtils.getBigDecimal(obj.get("DQFXYGFDN")));
+        dataLoad.setCurrentReverseActiveFlat(StringUtils.getBigDecimal(obj.get("DQFXYGPDN")));
+        dataLoad.setCurrentReverseActiveValley(StringUtils.getBigDecimal(obj.get("DQFXYGGDN")));
+        // 电能设置-组合无功
+        dataLoad.setCurrentCombReactiveTotal(StringUtils.getBigDecimal(obj.get("DQZHWGZDN")));
+        dataLoad.setCurrentCombReactivePeak(StringUtils.getBigDecimal(obj.get("DQZHWGJDN")));
+        dataLoad.setCurrentCombReactiveHigh(StringUtils.getBigDecimal(obj.get("DQZHWGFDN")));
+        dataLoad.setCurrentCombReactiveFlat(StringUtils.getBigDecimal(obj.get("DQZHWGPDN")));
+        dataLoad.setCurrentCombReactiveValley(StringUtils.getBigDecimal(obj.get("DQZHWGGDN")));
+        // 电能设置-正向无功
+        dataLoad.setCurrentForwardReactiveTotal(StringUtils.getBigDecimal(obj.get("DQZXWGZDN")));
+        dataLoad.setCurrentForwardReactivePeak(StringUtils.getBigDecimal(obj.get("DQZXWGJDN")));
+        dataLoad.setCurrentForwardReactiveHigh(StringUtils.getBigDecimal(obj.get("DQZXWGFDN")));
+        dataLoad.setCurrentForwardReactiveFlat(StringUtils.getBigDecimal(obj.get("DQZXWGPDN")));
+        dataLoad.setCurrentForwardReactiveValley(StringUtils.getBigDecimal(obj.get("DQZXWGGDN")));
+        // 电能设置-反向无功
+        dataLoad.setCurrentReverseReactiveTotal(StringUtils.getBigDecimal(obj.get("DQFXWGZDN")));
+        dataLoad.setCurrentReverseReactivePeak(StringUtils.getBigDecimal(obj.get("DQFXWGJDN")));
+        dataLoad.setCurrentReverseReactiveHigh(StringUtils.getBigDecimal(obj.get("DQFXWGFDN")));
+        dataLoad.setCurrentReverseReactiveFlat(StringUtils.getBigDecimal(obj.get("DQFXWGPDN")));
+        dataLoad.setCurrentReverseReactiveValley(StringUtils.getBigDecimal(obj.get("DQFXWGGDN")));
+
+        // 电压+电流
+        dataLoad.setPhaseAVoltage(StringUtils.getBigDecimal(obj.get("AXDY")));
+        dataLoad.setPhaseBVoltage(StringUtils.getBigDecimal(obj.get("BXDY")));
+        dataLoad.setPhaseCVoltage(StringUtils.getBigDecimal(obj.get("CXDY")));
+        dataLoad.setPhaseACurrent(StringUtils.getBigDecimal(obj.get("AXDL")));
+        dataLoad.setPhaseBCurrent(StringUtils.getBigDecimal(obj.get("BXDL")));
+        dataLoad.setPhaseCCurrent(StringUtils.getBigDecimal(obj.get("CXDL")));
+
+        dataLoad.setAbLineVoltage(StringUtils.getBigDecimal(obj.get("ABXDY")));
+        dataLoad.setCbLineVoltage(StringUtils.getBigDecimal(obj.get("CBXDY")));
+        dataLoad.setAcLineVoltage(StringUtils.getBigDecimal(obj.get("ACXDY")));
+
+        // 频率
+        dataLoad.setFrequency(StringUtils.getBigDecimal(obj.get("PL")));
+
+        // 功率 有功+总+无功+无总+视在
+        dataLoad.setPhaseAActivePower(StringUtils.getBigDecimal(obj.get("AXYGGL")));
+        dataLoad.setPhaseBActivePower(StringUtils.getBigDecimal(obj.get("BXYGGL")));
+        dataLoad.setPhaseCActivePower(StringUtils.getBigDecimal(obj.get("CXYGGL")));
+        dataLoad.setTotalActivePower(StringUtils.getBigDecimal(obj.get("ZYGGL")));
+        dataLoad.setPhaseAReactivePower(StringUtils.getBigDecimal(obj.get("AXWGGL")));
+        dataLoad.setPhaseBReactivePower(StringUtils.getBigDecimal(obj.get("BXWGGL")));
+        dataLoad.setPhaseCReactivePower(StringUtils.getBigDecimal(obj.get("CXWGGL")));
+        dataLoad.setTotalReactivePower(StringUtils.getBigDecimal(obj.get("ZWGGL")));
+        dataLoad.setPhaseAApparentPower(StringUtils.getBigDecimal(obj.get("AXSZGL")));
+        dataLoad.setPhaseBApparentPower(StringUtils.getBigDecimal(obj.get("BXSZGL")));
+        dataLoad.setPhaseCApparentPower(StringUtils.getBigDecimal(obj.get("CXSZGL")));
+        dataLoad.setTotalApparentPower(StringUtils.getBigDecimal(obj.get("ZSZGL")));
+
+        // 功率因数
+        dataLoad.setPhaseAPowerFactor(StringUtils.getBigDecimal(obj.get("AXGLYS")));
+        dataLoad.setPhaseBPowerFactor(StringUtils.getBigDecimal(obj.get("BXGLYS")));
+        dataLoad.setPhaseCPowerFactor(StringUtils.getBigDecimal(obj.get("CXGLYS")));
+        dataLoad.setTotalPowerFactor(StringUtils.getBigDecimal(obj.get("ZGLYS")));
+
+        // 需量
+        dataLoad.setForwardAcMaxDemand(StringUtils.getBigDecimal(obj.get("ZXYGZDXL")));
+        dataLoad.setReverseAcMaxDemand(StringUtils.getBigDecimal(obj.get("FXYGZDXL")));
+        dataLoad.setDailyForwardMaxDemand(StringUtils.getBigDecimal(obj.get("DRZXYGZDXL")));
+
+        dataLoad.setCreateBy("system");
+        dataLoad.setCreateTime(DateUtils.getNowDate());
+        dataLoad.setUpdateBy("system");
+        dataLoad.setUpdateTime(DateUtils.getNowDate());
+        dataLoad.setSiteId(SITE_ID);
+        dataLoad.setDeviceId(deviceId);
+
+        emsAmmeterDataMapper.insertEmsAmmeterData(dataLoad);
+
+        redisCache.setCacheObject(RedisKeyConstants.AMMETER + SITE_ID + "_" +deviceId, dataLoad);
+    }
+
+    private void meteDataProcess(String deviceId, String dataJson) {
+
+        //总表
+        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
+        });
+
+        EmsAmmeterData dataLoad = new EmsAmmeterData();
+        // 更新时间
+        dataLoad.setDataUpdateTime(new Date());
+
+        // 电压+电流
+        dataLoad.setPhaseAVoltage(StringUtils.getBigDecimal(obj.get("AXDY")));
+        dataLoad.setPhaseBVoltage(StringUtils.getBigDecimal(obj.get("BXDY")));
+        dataLoad.setPhaseCVoltage(StringUtils.getBigDecimal(obj.get("CXDY")));
+        dataLoad.setPhaseACurrent(StringUtils.getBigDecimal(obj.get("AXDL")));
+        dataLoad.setPhaseBCurrent(StringUtils.getBigDecimal(obj.get("BXDL")));
+        dataLoad.setPhaseCCurrent(StringUtils.getBigDecimal(obj.get("CXDL")));
+
+        dataLoad.setAbLineVoltage(StringUtils.getBigDecimal(obj.get("ABXDY")));
+        dataLoad.setCbLineVoltage(StringUtils.getBigDecimal(obj.get("BCXDY")));
+        dataLoad.setAcLineVoltage(StringUtils.getBigDecimal(obj.get("CAXDY")));
+
+        // 频率
+        dataLoad.setFrequency(StringUtils.getBigDecimal(obj.get("DWPL")));
+
+        // 功率
+        dataLoad.setPhaseAActivePower(StringUtils.getBigDecimal(obj.get("AXYGGL")));
+        dataLoad.setPhaseBActivePower(StringUtils.getBigDecimal(obj.get("BXYGGL")));
+        dataLoad.setPhaseCActivePower(StringUtils.getBigDecimal(obj.get("CXYGGL")));
+        dataLoad.setTotalActivePower(StringUtils.getBigDecimal(obj.get("ZYGGL")));
+        dataLoad.setPhaseAReactivePower(StringUtils.getBigDecimal(obj.get("AXWGGL")));
+        dataLoad.setPhaseBReactivePower(StringUtils.getBigDecimal(obj.get("BXWGGL")));
+        dataLoad.setPhaseCReactivePower(StringUtils.getBigDecimal(obj.get("CXWGGL")));
+        dataLoad.setTotalReactivePower(StringUtils.getBigDecimal(obj.get("ZWGGL")));
+
+        dataLoad.setTotalApparentPower(StringUtils.getBigDecimal(obj.get("ZSZGL")));
+        dataLoad.setTotalPowerFactor(StringUtils.getBigDecimal(obj.get("ZGLYS")));
+
+        // 二次相关数据
+        dataLoad.setSecondaryAbLineVoltage(StringUtils.getBigDecimal(obj.get("ECABXDY")));
+        dataLoad.setSecondaryAPhaseCurrent(StringUtils.getBigDecimal(obj.get("ECAXDL")));
+        dataLoad.setSecondaryAPhaseVoltage(StringUtils.getBigDecimal(obj.get("ECAXDY")));
+        dataLoad.setSecondaryAPowerFactor(StringUtils.getBigDecimal(obj.get("ECAXGLYS")));
+        dataLoad.setSecondaryAApparentPower(StringUtils.getBigDecimal(obj.get("ECAXSZGL")));
+        dataLoad.setSecondaryAReactivePower(StringUtils.getBigDecimal(obj.get("ECAXWGGL")));
+        dataLoad.setSecondaryAActivePower(StringUtils.getBigDecimal(obj.get("ECAXYGGL")));
+        dataLoad.setSecondaryBcLineVoltage(StringUtils.getBigDecimal(obj.get("ECBCXDY")));
+        dataLoad.setSecondaryBPhaseCurrent(StringUtils.getBigDecimal(obj.get("ECBXDL")));
+        dataLoad.setSecondaryBPhaseVoltage(StringUtils.getBigDecimal(obj.get("ECBXDY")));
+        dataLoad.setSecondaryBPowerFactor(StringUtils.getBigDecimal(obj.get("ECBXGLYS")));
+        dataLoad.setSecondaryBApparentPower(StringUtils.getBigDecimal(obj.get("ECBXSZGL")));
+        dataLoad.setSecondaryBReactivePower(StringUtils.getBigDecimal(obj.get("ECBXWGGL")));
+        dataLoad.setSecondaryBActivePower(StringUtils.getBigDecimal(obj.get("ECBXYGGL")));
+        dataLoad.setSecondaryCaLineVoltage(StringUtils.getBigDecimal(obj.get("ECCAXDY")));
+        dataLoad.setSecondaryCPhaseCurrent(StringUtils.getBigDecimal(obj.get("ECCXDL")));
+        dataLoad.setSecondaryCPhaseVoltage(StringUtils.getBigDecimal(obj.get("ECCXDY")));
+        dataLoad.setSecondaryCPowerFactor(StringUtils.getBigDecimal(obj.get("ECCXGLYS")));
+        dataLoad.setSecondaryCApparentPower(StringUtils.getBigDecimal(obj.get("ECCXSZGL")));
+        dataLoad.setSecondaryCReactivePower(StringUtils.getBigDecimal(obj.get("ECCXWGGL")));
+        dataLoad.setSecondaryCActivePower(StringUtils.getBigDecimal(obj.get("ECCXYGGL")));
+        dataLoad.setSecondaryGridFrequency(StringUtils.getBigDecimal(obj.get("ECDWPL")));
+        dataLoad.setSecondaryReverseReactiveEnergy(StringUtils.getBigDecimal(obj.get("ECFXWGDN")));
+        dataLoad.setSecondaryNegativeActiveEnergy(StringUtils.getBigDecimal(obj.get("ECFXYGDN")));
+        dataLoad.setSecondaryTotalPowerFactor(StringUtils.getBigDecimal(obj.get("ECZGLYS")));
+        dataLoad.setSecondaryTotalApparentPower(StringUtils.getBigDecimal(obj.get("ECZSZFL")));
+        dataLoad.setSecondaryTotalReactivePower(StringUtils.getBigDecimal(obj.get("ECZWGGL")));
+        dataLoad.setSecondaryPositiveReactiveEnergy(StringUtils.getBigDecimal(obj.get("ECZXWGDN")));
+        dataLoad.setSecondaryPositiveActiveEnergy(StringUtils.getBigDecimal(obj.get("ECZXYGDN")));
+        dataLoad.setSecondaryTotalActivePower(StringUtils.getBigDecimal(obj.get("ECZYGGL")));
+
+        // 需量
+        dataLoad.setReverseReactiveEnergyEqMinus(StringUtils.getBigDecimal(obj.get("FXWGDN")));
+        dataLoad.setReverseActiveEnergyEpMinus(StringUtils.getBigDecimal(obj.get("FXYGDN")));
+        dataLoad.setPositiveReactiveEnergyEqPlus(StringUtils.getBigDecimal(obj.get("ZXWGDN")));
+        dataLoad.setPositiveActiveEnergyEpPlus(StringUtils.getBigDecimal(obj.get("ZXYGDN")));
+
+        dataLoad.setCreateBy("system");
+        dataLoad.setCreateTime(DateUtils.getNowDate());
+        dataLoad.setUpdateBy("system");
+        dataLoad.setUpdateTime(DateUtils.getNowDate());
+        dataLoad.setSiteId(SITE_ID);
+        dataLoad.setDeviceId(deviceId);
+
+        emsAmmeterDataMapper.insertEmsAmmeterData(dataLoad);
+
+        redisCache.setCacheObject(RedisKeyConstants.AMMETER + SITE_ID + "_" +deviceId, dataLoad);
+    }
+    // 数据分组处理
+    private static Map<String, Map<String, Object>> processData(Map<String, Object> rawData) {
+        Map<String, Map<String, Object>> records = new HashMap<>();
+
+        for (Map.Entry<String, Object> entry : rawData.entrySet()) {
+            String key = entry.getKey();
+            // 提取记录ID（最后3位）
+            String recordId = key.substring(key.length() - 3);
+            try {
+                Long.parseLong(recordId);
+            } catch (Exception e) {
+                continue;
+            }
+
+            // 提取字段类型（前缀）
+            String fieldType = key.substring(0, key.length() - 3);
+
+            // 初始化记录
+            records.putIfAbsent(recordId, new HashMap<>());
+            // 存入字段值
+            records.get(recordId).put(fieldType, entry.getValue());
+        }
+        return records;
+    }
+
+    private static Map<String, Map<String, Object>> processDataPrefix(Map<String, Object> rawData) {
+        Map<String, Map<String, Object>> records = new HashMap<>();
+
+        for (Map.Entry<String, Object> entry : rawData.entrySet()) {
+            String key = entry.getKey();
+            // 确保键长度足够
+            if (key.length() < 3) {
+                continue;
+            }
+
+            // 提取记录ID（前3位）
+            String recordId = key.substring(0, 3);
+            if (!recordId.startsWith("DY")) {
+                continue;
+            }
+
+            // 提取字段类型（剩余部分）
+            String fieldType = key.substring(3);
+
+            // 初始化记录
+            records.putIfAbsent(recordId, new HashMap<>());
+            // 存入字段值
+            records.get(recordId).put(fieldType, entry.getValue());
+        }
+        return records;
+    }
+
+}