完成暂态算法和调整部分逻辑

2025-03-17 20:13:04 +08:00
parent a7bf00e312
commit cd84bd00d4
29 changed files with 871 additions and 40 deletions
--- a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/executor/MeasurementExecutor.java
+++ b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/executor/MeasurementExecutor.java
@@ -1,9 +1,7 @@
 package com.njcn.algorithm.executor;

-import com.njcn.algorithm.service.line.IDataCleanService;
-import com.njcn.algorithm.service.line.IDataCrossingService;
-import com.njcn.algorithm.service.line.IDataIntegrityService;
-import com.njcn.algorithm.service.line.IDayDataService;
+import com.njcn.algorithm.pojo.bo.CalculatedParam;
+import com.njcn.algorithm.service.line.*;
 import com.yomahub.liteflow.annotation.LiteflowComponent;
 import com.yomahub.liteflow.annotation.LiteflowMethod;
 import com.yomahub.liteflow.core.NodeComponent;
@@ -33,17 +31,20 @@ public class MeasurementExecutor extends BaseExecutor {
    private IDataCrossingService dataCrossingService;
    @Resource
    private IDataIntegrityService dataIntegrityService;
-
+    @Resource
+    private IEventDetailService eventDetailService;

    /**
     * 数据清洗 电压表
     * dataV表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataVClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataVCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataVClean", nodeType = NodeTypeEnum.COMMON)
    public void dataVCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataVCleanHandler(bindCmp.getRequestData());
@@ -52,12 +53,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗 电流表
     * dataI表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataIClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataICleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataIClean", nodeType = NodeTypeEnum.COMMON)
    public void dataICleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataICleanHandler(bindCmp.getRequestData());
@@ -66,12 +69,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗 长闪表
     * dataPlt表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataPltClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataPltCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataPltClean", nodeType = NodeTypeEnum.COMMON)
    public void dataPltCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataPltCleanHandler(bindCmp.getRequestData());
@@ -80,12 +85,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗 间谐波电压表
     * DataInHarmV表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataInHarmVClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataInHarmVCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataInHarmVClean", nodeType = NodeTypeEnum.COMMON)
    public void dataInHarmVCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataInHarmVCleanHandler(bindCmp.getRequestData());
@@ -94,12 +101,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗 谐波电压含有率
     * DataHarmRateV表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataHarmRateVClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataHarmRateVCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmRateVClean", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmRateVCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataHarmRateVCleanHandler(bindCmp.getRequestData());
@@ -108,12 +117,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗 有功功率表
     * DataHarmPowerP表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataHarmPowerPClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataHarmPowerPCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPowerPClean", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPowerPCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataHarmPowerPCleanHandler(bindCmp.getRequestData());
@@ -122,12 +133,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗 谐波电压角度表
     * DataHarmPhasicV表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataHarmPhasicVClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataHarmPhasicVCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPhasicVClean", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPhasicVCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataHarmPhasicVCleanHandler(bindCmp.getRequestData());
@@ -136,12 +149,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗  电压波动表
     * DataFluc表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataFlucClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataFlucCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataFlucClean", nodeType = NodeTypeEnum.COMMON)
    public void dataFlucCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataFlucCleanHandler(bindCmp.getRequestData());
@@ -150,12 +165,14 @@ public class MeasurementExecutor extends BaseExecutor {
    /**
     * 数据清洗
     * DataFlicker表
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataFlickerClean", nodeType = NodeTypeEnum.COMMON)
    public boolean dataFlickerCleanAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataFlickerClean", nodeType = NodeTypeEnum.COMMON)
    public void dataFlickerCleanProcess(NodeComponent bindCmp) {
        dataCleanService.dataFlickerCleanHandler(bindCmp.getRequestData());
@@ -164,12 +181,14 @@ public class MeasurementExecutor extends BaseExecutor {

    /**
     * 监测点报表_日表(r_stat_data_*_d)
+     *
     * @author xy
     */
    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataV", nodeType = NodeTypeEnum.COMMON)
    public boolean dataVToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataV", nodeType = NodeTypeEnum.COMMON)
    public void dataVToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataVHandler(bindCmp.getRequestData());
@@ -179,6 +198,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataIToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataI", nodeType = NodeTypeEnum.COMMON)
    public void dataIToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataIHandler(bindCmp.getRequestData());
@@ -188,6 +208,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataFlickerToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataFlicker", nodeType = NodeTypeEnum.COMMON)
    public void dataFlickerToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataFlickerHandler(bindCmp.getRequestData());
@@ -197,6 +218,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataFlucToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataFluc", nodeType = NodeTypeEnum.COMMON)
    public void dataFlucToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataFlucHandler(bindCmp.getRequestData());
@@ -206,6 +228,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmPhasicIToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPhasicI", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPhasicIToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmPhasicIHandler(bindCmp.getRequestData());
@@ -215,6 +238,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmPhasicVToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPhasicV", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPhasicVToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmPhasicVHandler(bindCmp.getRequestData());
@@ -224,6 +248,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmPowerPToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPowerP", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPowerPToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmPowerPHandler(bindCmp.getRequestData());
@@ -233,6 +258,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmPowerQToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPowerQ", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPowerQToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmPowerQHandler(bindCmp.getRequestData());
@@ -242,6 +268,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmPowerSToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmPowerS", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmPowerSToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmPowerSHandler(bindCmp.getRequestData());
@@ -251,6 +278,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmRateIToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmRateI", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmRateIToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmRateIHandler(bindCmp.getRequestData());
@@ -260,6 +288,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataHarmRateVToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmRateV", nodeType = NodeTypeEnum.COMMON)
    public void dataHarmRateVToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataHarmRateVHandler(bindCmp.getRequestData());
@@ -269,6 +298,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataInHarmIToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataInHarmI", nodeType = NodeTypeEnum.COMMON)
    public void dataInHarmIToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataInHarmIHandler(bindCmp.getRequestData());
@@ -278,6 +308,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataInHarmVToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataInHarmV", nodeType = NodeTypeEnum.COMMON)
    public void dataInHarmVToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataInHarmVHandler(bindCmp.getRequestData());
@@ -287,17 +318,51 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataPltToDayAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataPlt", nodeType = NodeTypeEnum.COMMON)
    public void dataPltToDayProcess(NodeComponent bindCmp) {
        dayDataService.dataPltHandler(bindCmp.getRequestData());
    }


+    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "rMpEventDetail", nodeType = NodeTypeEnum.COMMON)
+    public boolean processRMpEventDetailAccess(NodeComponent bindCmp) {
+        return isAccess(bindCmp);
+    }
+
+    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "rMpEventDetail", nodeType = NodeTypeEnum.COMMON)
+    public void processRMpEventDetailProcess(NodeComponent bindCmp) {
+        String tag = bindCmp.getTag();
+        CalculatedParam calculatedParam = bindCmp.getRequestData();
+        if ("r_mp_event_detail_d".equalsIgnoreCase(tag)) {
+            //日表
+            eventDetailService.dataDayHandle(calculatedParam);
+        } else if ("r_mp_event_detail_m".equalsIgnoreCase(tag)) {
+            //数据补招不执行非日表算法
+            if (!calculatedParam.isRepair()) {
+                //月表
+                eventDetailService.dataMonthHandle(calculatedParam);
+            }
+        } else if ("r_mp_event_detail_q".equalsIgnoreCase(tag)) {
+            //数据补招不执行非日表算法
+            if (!calculatedParam.isRepair()) {
+                //季表
+                eventDetailService.dataQuarterHandle(calculatedParam);
+            }
+        } else if ("r_mp_event_detail_y".equalsIgnoreCase(tag)) {
+            //数据补招不执行非日表算法
+            if (!calculatedParam.isRepair()) {
+                //年表
+                eventDetailService.dataYearHandle(calculatedParam);
+            }
+        }
+    }

    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataLimitRate", nodeType = NodeTypeEnum.COMMON)
    public boolean dataLimitRateAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataLimitRate", nodeType = NodeTypeEnum.COMMON)
    public void dataLimitRateProcess(NodeComponent bindCmp) {
        dataCrossingService.limitRateHandler(bindCmp.getRequestData());
@@ -307,6 +372,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataLimitTargetAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataLimitTarget", nodeType = NodeTypeEnum.COMMON)
    public void dataLimitTargetProcess(NodeComponent bindCmp) {
        dataCrossingService.limitTargetHandler(bindCmp.getRequestData());
@@ -316,6 +382,7 @@ public class MeasurementExecutor extends BaseExecutor {
    public boolean dataIntegrityAccess(NodeComponent bindCmp) {
        return isAccess(bindCmp);
    }
+
    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataIntegrity", nodeType = NodeTypeEnum.COMMON)
    public void dataIntegrityProcess(NodeComponent bindCmp) {
        dataIntegrityService.dataIntegrity(bindCmp.getRequestData());
--- a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataCrossingServiceImpl.java
+++ b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataCrossingServiceImpl.java
@@ -24,7 +24,6 @@ import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.ListUtils;
-import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -34,7 +33,6 @@ import org.springframework.util.CollectionUtils;
 import javax.annotation.Resource;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
-import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.*;
 import java.util.function.Function;
--- a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IEventDetailServiceImpl.java
+++ b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IEventDetailServiceImpl.java
@@ -1,13 +1,39 @@
 package com.njcn.algorithm.serviceimpl.line;

+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.date.DatePattern;
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.date.LocalDateTimeUtil;
+import cn.hutool.core.util.ObjectUtil;
 import com.njcn.algorithm.pojo.bo.CalculatedParam;
 import com.njcn.algorithm.service.line.IEventDetailService;
+import com.njcn.dataProcess.api.EventDetailFeignClient;
+import com.njcn.dataProcess.api.RmpEventDetailFeignClient;
+import com.njcn.dataProcess.dto.RmpEventDetailDTO;
+import com.njcn.dataProcess.param.LineCountEvaluateParam;
+import com.njcn.dataProcess.pojo.dto.DataEventDetailDto;
+import com.njcn.dataProcess.pojo.dto.DataLimitDetailDto;
+import com.njcn.dataProcess.util.TimeUtils;
+import com.njcn.device.biz.pojo.po.Overlimit;
+import com.njcn.system.api.DicDataFeignClient;
+import com.njcn.system.enums.DicDataEnum;
+import com.njcn.system.pojo.po.DictData;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.ListUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;

+import javax.annotation.Resource;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
 /**
 * @author wr
 * @description
@@ -17,26 +43,163 @@ import org.springframework.stereotype.Component;
@Component
@RequiredArgsConstructor
 public class IEventDetailServiceImpl implements IEventDetailService {
-
    private static final Logger logger = LoggerFactory.getLogger(IDataOnlineRateServiceImpl.class);
+    @Value("${line.num}")
+    private Integer NUM = 100;
+    @Resource
+    private RmpEventDetailFeignClient eventDetailFeignClient;
+    @Resource
+    private DicDataFeignClient dicDataFeignClient;
+    @Resource
+    private EventDetailFeignClient eventDetailCountFeignClient;

    @Override
    public void dataDayHandle(CalculatedParam<String> calculatedParam) {
+        logger.info("{},r_mp_event_detail_d算法开始=====》", LocalDateTime.now());
+        List<DataEventDetailDto> result = new ArrayList<>();
+        //远程接口获取分钟数据
+        LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
+        lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
+        lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
+        List<String> lineIds = calculatedParam.getIdList();
+
+        DictData dip = dicDataFeignClient.getDicDataByCode(DicDataEnum.VOLTAGE_DIP.getCode()).getData();
+        DictData rise = dicDataFeignClient.getDicDataByCode(DicDataEnum.VOLTAGE_RISE.getCode()).getData();
+        DictData interruptions = dicDataFeignClient.getDicDataByCode(DicDataEnum.SHORT_INTERRUPTIONS.getCode()).getData();
+        //以100个监测点分片处理
+        List<List<String>> pendingIds = ListUtils.partition(lineIds, NUM);
+        for (List<String> pendingId : pendingIds) {
+            lineParam.setLineId(pendingId);
+            List<RmpEventDetailDTO> data = eventDetailFeignClient.getRawData(lineParam).getData();
+            if (CollUtil.isNotEmpty(data)) {
+                Map<String, List<RmpEventDetailDTO>> eventDetailMap =
+                        data.stream().collect(Collectors.groupingBy(RmpEventDetailDTO::getMeasurementPointId));
+                eventDetailMap.forEach((key, value) -> {
+                    DataEventDetailDto dto = new DataEventDetailDto();
+                    dto.setMeasurementPointId(key);
+                    dto.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate(), DatePattern.NORM_DATE_PATTERN));
+                    dto.setSwellTimes((int) value.stream().filter(x -> isData(x, dip)).count());
+                    dto.setSagTimes((int) value.stream().filter(x -> isData(x, rise)).count());
+                    dto.setInterruptTimes((int) value.stream().filter(x -> isData(x, interruptions)).count());
+                    result.add(dto);
+                });
+            }
+        }
+        if(CollUtil.isNotEmpty(result)){
+            eventDetailCountFeignClient.batchInsertionD(result);
+        }

    }

+    public Boolean isData(RmpEventDetailDTO dto, DictData dictData) {
+        if (ObjectUtil.isNull(dictData)) {
+            return false;
+        } else {
+            if (dictData.getId().trim().equals(dto.getEventType().trim())) {
+                return true;
+            } else {
+                return false;
+            }
+        }
+    }
+
+
    @Override
    public void dataMonthHandle(CalculatedParam<String> calculatedParam) {
-
+        logger.info("{},r_mp_event_detail_m算法开始=====》", LocalDateTime.now());
+        List<DataEventDetailDto> result = new ArrayList<>();
+        //远程接口获取分钟数据
+        LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
+        lineParam.setStartTime(TimeUtils.getBeginOfMonth(calculatedParam.getDataDate()));
+        lineParam.setEndTime(TimeUtils.getEndOfMonth(calculatedParam.getDataDate()));
+        List<String> lineIds = calculatedParam.getIdList();
+        //以100个监测点分片处理
+        List<List<String>> pendingIds = ListUtils.partition(lineIds, NUM);
+        for (List<String> pendingId : pendingIds) {
+            lineParam.setLineId(pendingId);
+            List<DataEventDetailDto> data = eventDetailCountFeignClient.getRawDataD(lineParam).getData();
+            if (CollUtil.isNotEmpty(data)) {
+                Map<String, List<DataEventDetailDto>> eventDetailMap =
+                        data.stream().collect(Collectors.groupingBy(DataEventDetailDto::getMeasurementPointId));
+                eventDetailMap.forEach((key, value) -> {
+                    DataEventDetailDto dto = new DataEventDetailDto();
+                    dto.setMeasurementPointId(key);
+                    dto.setDataDate(LocalDateTimeUtil.parseDate(lineParam.getStartTime(), DatePattern.NORM_DATETIME_PATTERN));
+                    dto.setSwellTimes(value.stream().mapToInt(DataEventDetailDto::getSwellTimes).sum());
+                    dto.setSagTimes(value.stream().mapToInt(DataEventDetailDto::getSagTimes).sum());
+                    dto.setInterruptTimes(value.stream().mapToInt(DataEventDetailDto::getInterruptTimes).sum());
+                    result.add(dto);
+                });
+            }
+        }
+        if(CollUtil.isNotEmpty(result)){
+            eventDetailCountFeignClient.batchInsertionM(result);
+        }
    }

    @Override
    public void dataQuarterHandle(CalculatedParam<String> calculatedParam) {
-
+        logger.info("{},r_mp_event_detail_q算法开始=====》", LocalDateTime.now());
+        List<DataEventDetailDto> result = new ArrayList<>();
+        //远程接口获取分钟数据
+        LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
+        lineParam.setStartTime(TimeUtils.getBeginOfQuarter(calculatedParam.getDataDate()));
+        lineParam.setEndTime(TimeUtils.getEndOfQuarter(calculatedParam.getDataDate()));
+        List<String> lineIds = calculatedParam.getIdList();
+        //以100个监测点分片处理
+        List<List<String>> pendingIds = ListUtils.partition(lineIds, NUM);
+        for (List<String> pendingId : pendingIds) {
+            lineParam.setLineId(pendingId);
+            List<DataEventDetailDto> data = eventDetailCountFeignClient.getRawDataM(lineParam).getData();
+            if (CollUtil.isNotEmpty(data)) {
+                Map<String, List<DataEventDetailDto>> eventDetailMap =
+                        data.stream().collect(Collectors.groupingBy(DataEventDetailDto::getMeasurementPointId));
+                eventDetailMap.forEach((key, value) -> {
+                    DataEventDetailDto dto = new DataEventDetailDto();
+                    dto.setMeasurementPointId(key);
+                    dto.setDataDate(LocalDateTimeUtil.parseDate(lineParam.getStartTime(), DatePattern.NORM_DATETIME_PATTERN));
+                    dto.setSwellTimes(value.stream().mapToInt(DataEventDetailDto::getSwellTimes).sum());
+                    dto.setSagTimes(value.stream().mapToInt(DataEventDetailDto::getSagTimes).sum());
+                    dto.setInterruptTimes(value.stream().mapToInt(DataEventDetailDto::getInterruptTimes).sum());
+                    result.add(dto);
+                });
+            }
+        }
+        if(CollUtil.isNotEmpty(result)){
+            eventDetailCountFeignClient.batchInsertionQ(result);
+        }
    }

    @Override
    public void dataYearHandle(CalculatedParam<String> calculatedParam) {
-
+        logger.info("{},r_mp_event_detail_y算法开始=====》", LocalDateTime.now());
+        List<DataEventDetailDto> result = new ArrayList<>();
+        //远程接口获取分钟数据
+        LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
+        lineParam.setStartTime(TimeUtils.getBeginOfYear(calculatedParam.getDataDate()));
+        lineParam.setEndTime(TimeUtils.getEndOfYear(calculatedParam.getDataDate()));
+        List<String> lineIds = calculatedParam.getIdList();
+        //以100个监测点分片处理
+        List<List<String>> pendingIds = ListUtils.partition(lineIds, NUM);
+        for (List<String> pendingId : pendingIds) {
+            lineParam.setLineId(pendingId);
+            List<DataEventDetailDto> data = eventDetailCountFeignClient.getRawDataQ(lineParam).getData();
+            if (CollUtil.isNotEmpty(data)) {
+                Map<String, List<DataEventDetailDto>> eventDetailMap =
+                        data.stream().collect(Collectors.groupingBy(DataEventDetailDto::getMeasurementPointId));
+                eventDetailMap.forEach((key, value) -> {
+                    DataEventDetailDto dto = new DataEventDetailDto();
+                    dto.setMeasurementPointId(key);
+                    dto.setDataDate(LocalDateTimeUtil.parseDate(lineParam.getStartTime(), DatePattern.NORM_DATETIME_PATTERN));
+                    dto.setSwellTimes(value.stream().mapToInt(DataEventDetailDto::getSwellTimes).sum());
+                    dto.setSagTimes(value.stream().mapToInt(DataEventDetailDto::getSagTimes).sum());
+                    dto.setInterruptTimes(value.stream().mapToInt(DataEventDetailDto::getInterruptTimes).sum());
+                    result.add(dto);
+                });
+            }
+        }
+        if(CollUtil.isNotEmpty(result)){
+            eventDetailCountFeignClient.batchInsertionY(result);
+        }
    }
 }