From 687f878b5f4ebb307d0b958e5954519bd82aa3ae Mon Sep 17 00:00:00 2001 From: wr <1754607820@qq.com> Date: Wed, 1 Jul 2026 11:27:46 +0800 Subject: [PATCH] =?UTF-8?q?1.=E8=A7=A3=E5=86=B3=E6=9C=89=E5=8A=9F=E5=8A=9F?= =?UTF-8?q?=E7=8E=87=E8=B6=8B=E5=8A=BF=E7=BB=9F=E8=AE=A1=E8=B0=83=E7=94=A8?= =?UTF-8?q?ataprocessboot=E6=8E=A5=E5=8F=A3=E6=94=B9=E9=80=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../mapper/influxdb/DataHarmRateIMapper.java | 11 + .../mapper/influxdb/DataHarmRateVMapper.java | 11 + .../harmonic/mapper/influxdb/DataIMapper.java | 16 ++ .../mapper/influxdb/DataInharmVMapper.java | 17 ++ .../mapper/influxdb/DataPltMapper.java | 17 ++ .../harmonic/mapper/influxdb/DataVMapper.java | 27 +++ .../impl/PowerStatisticsServiceImpl.java | 26 +-- .../service/influxdb/IDataHarmRateV.java | 26 +++ .../harmonic/service/influxdb/IDataI.java | 26 +++ .../service/influxdb/IDataInHarmV.java | 32 +++ .../harmonic/service/influxdb/IDataPlt.java | 31 +++ .../harmonic/service/influxdb/IDataV.java | 24 ++ .../impl/InfluxdbDataHarmRateVImpl.java | 158 +++++++++++++ .../influxdb/impl/InfluxdbDataIImpl.java | 135 +++++++++++ .../impl/InfluxdbDataInharmVImpl.java | 132 +++++++++++ .../influxdb/impl/InfluxdbDataPltImpl.java | 157 +++++++++++++ .../influxdb/impl/InfluxdbDataVImpl.java | 221 ++++++++++++++++++ 17 files changed, 1053 insertions(+), 14 deletions(-) create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateIMapper.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateVMapper.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataIMapper.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataInharmVMapper.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataPltMapper.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataVMapper.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataHarmRateV.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataI.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataInHarmV.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataPlt.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataV.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataHarmRateVImpl.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataIImpl.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataInharmVImpl.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataPltImpl.java create mode 100644 pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataVImpl.java diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateIMapper.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateIMapper.java new file mode 100644 index 000000000..5fb616382 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateIMapper.java @@ -0,0 +1,11 @@ +package com.njcn.harmonic.mapper.influxdb; + +import com.njcn.dataProcess.po.influx.DataHarmrateI; +import com.njcn.influx.base.InfluxDbBaseMapper; + +/** + * @author xy + */ +public interface DataHarmRateIMapper extends InfluxDbBaseMapper { + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateVMapper.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateVMapper.java new file mode 100644 index 000000000..4ad0a724d --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataHarmRateVMapper.java @@ -0,0 +1,11 @@ +package com.njcn.harmonic.mapper.influxdb; + +import com.njcn.dataProcess.po.influx.DataHarmrateV; +import com.njcn.influx.base.InfluxDbBaseMapper; + +/** + * @author xy + */ +public interface DataHarmRateVMapper extends InfluxDbBaseMapper { + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataIMapper.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataIMapper.java new file mode 100644 index 000000000..59d49e5ce --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataIMapper.java @@ -0,0 +1,16 @@ +package com.njcn.harmonic.mapper.influxdb; + +import com.njcn.dataProcess.po.influx.DataI; +import com.njcn.influx.base.InfluxDbBaseMapper; + + +/** + * @author hongawen + * @version 1.0 + * @data 2024/11/7 18:49 + */ +public interface DataIMapper extends InfluxDbBaseMapper { + + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataInharmVMapper.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataInharmVMapper.java new file mode 100644 index 000000000..dcc108da2 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataInharmVMapper.java @@ -0,0 +1,17 @@ +package com.njcn.harmonic.mapper.influxdb; + + +import com.njcn.dataProcess.po.influx.DataInharmV; +import com.njcn.influx.base.InfluxDbBaseMapper; + +/** + *

+ * Mapper 接口 + *

+ * + * @author hongawen + * @since 2023-12-28 + */ +public interface DataInharmVMapper extends InfluxDbBaseMapper { + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataPltMapper.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataPltMapper.java new file mode 100644 index 000000000..5aefda9e2 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataPltMapper.java @@ -0,0 +1,17 @@ +package com.njcn.harmonic.mapper.influxdb; + + +import com.njcn.dataProcess.po.influx.DataPlt; +import com.njcn.influx.base.InfluxDbBaseMapper; + +/** + *

+ * Mapper 接口 + *

+ * + * @author hongawen + * @since 2023-12-28 + */ +public interface DataPltMapper extends InfluxDbBaseMapper { + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataVMapper.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataVMapper.java new file mode 100644 index 000000000..ad77f6c29 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/mapper/influxdb/DataVMapper.java @@ -0,0 +1,27 @@ +package com.njcn.harmonic.mapper.influxdb; + +import com.njcn.dataProcess.dto.LineDataVFiveItemDTO; +import com.njcn.dataProcess.dto.MeasurementCountDTO; +import com.njcn.dataProcess.po.influx.DataV; +import com.njcn.influx.base.InfluxDbBaseMapper; +import com.njcn.influx.query.InfluxQueryWrapper; + +import java.util.List; + +/** + * @author hongawen + * @version 1.0 + * @data 2024/11/7 18:49 + */ +public interface DataVMapper extends InfluxDbBaseMapper { + + + List queryDataValue(InfluxQueryWrapper dataVQueryWrapper); + + + List getMeasurementCount(InfluxQueryWrapper influxQueryWrapper); + + + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/activepowerrange/impl/PowerStatisticsServiceImpl.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/activepowerrange/impl/PowerStatisticsServiceImpl.java index 696c8d870..fdd442f6f 100644 --- a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/activepowerrange/impl/PowerStatisticsServiceImpl.java +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/activepowerrange/impl/PowerStatisticsServiceImpl.java @@ -10,7 +10,6 @@ import com.alibaba.excel.EasyExcel; import com.alibaba.excel.ExcelWriter; import com.alibaba.excel.write.metadata.WriteSheet; import com.alibaba.fastjson.JSONArray; -import com.njcn.dataProcess.api.*; import com.njcn.dataProcess.param.LineCountEvaluateParam; import com.njcn.dataProcess.pojo.dto.*; import com.njcn.device.biz.pojo.po.Overlimit; @@ -21,9 +20,8 @@ import com.njcn.harmonic.pojo.vo.*; import com.njcn.harmonic.service.IDataLimitRateDetailService; import com.njcn.harmonic.service.activepowerrange.PowerStatisticsService; import com.njcn.harmonic.service.activepowerrange.RActivePowerRangeService; -import com.njcn.influx.service.CommonService; +import com.njcn.harmonic.service.influxdb.*; import com.njcn.poi.util.PoiUtil; -import com.njcn.system.api.EpdFeignClient; import lombok.RequiredArgsConstructor; import org.apache.poi.ss.usermodel.*; import org.apache.poi.xssf.usermodel.XSSFClientAnchor; @@ -56,11 +54,11 @@ public class PowerStatisticsServiceImpl implements PowerStatisticsService { private final RActivePowerRangeService rActivePowerRangeService; private final DecimalFormat dftwo = new DecimalFormat(Param.DECIMAL_FORMATTWOSTR); - private final DataVFeignClient dataVFeignClient; - private final DataIFeignClient dataIFeignClient; - private final DataPltFeignClient dataPltFeignClient; - private final DataInharmVFeignClient dataInharmVFeignClient; - private final DataHarmRateVFeignClient dataHarmRateVFeignClient; + private final IDataV dataV; + private final IDataI dataI; + private final IDataPlt dataPlt; + private final IDataInHarmV dataInHarmV; + private final IDataHarmRateV dataHarmRateV; private final OverLimitClient overLimitClient; private final IDataLimitRateDetailService iDataLimitRateDetailService; @@ -156,11 +154,11 @@ public class PowerStatisticsServiceImpl implements PowerStatisticsService { String time = param.getSearchBeginTime(); if ("1".equals(param.getStatisticalId())) { //电压数据 - List dataVAllTime = dataVFeignClient.getRawData(evaluateParam).getData(); + List dataVAllTime = dataV.getRawData(evaluateParam); //闪变数据 - List dataFlickerAllTime = dataPltFeignClient.getRawData(evaluateParam).getData(); + List dataFlickerAllTime = dataPlt.getRawData(evaluateParam); //电流数据 - List dataIList = dataIFeignClient.getRawData(evaluateParam).getData(); + List dataIList = dataI.getRawData(evaluateParam); //电压偏差 if ("Dev".equals(param.getCode()) || StrUtil.isBlank(param.getCode())) { addThdData(info, overlimit.getVoltageDev(), "vuDev", dataVAllTime, "电压上偏差","%",time); @@ -190,17 +188,17 @@ public class PowerStatisticsServiceImpl implements PowerStatisticsService { } if ("2".equals(param.getStatisticalId())) { //谐波数据 - List dataVHarmList = dataHarmRateVFeignClient.getRawData(evaluateParam).getData(); + List dataVHarmList = dataHarmRateV.getRawData(evaluateParam); addThdData(info, overlimit, "getUharm", "v", 2, 26, dataVHarmList, "谐波电压","%",time); } if ("3".equals(param.getStatisticalId())) { //电流数据 - List dataIList = dataIFeignClient.getRawData(evaluateParam).getData(); + List dataIList = dataI.getRawData(evaluateParam); addThdData(info, overlimit, "getUharm", "i", 2, 26, dataIList, "谐波电流","A",time); } if ("4".equals(param.getStatisticalId())) { //间谐波数据 - List dataVInHarmList = dataInharmVFeignClient.getRawData(evaluateParam).getData(); + List dataVInHarmList = dataInHarmV.getRawData(evaluateParam); addThdData(info, overlimit, "getInuharm", "v", 1, 17, dataVInHarmList, "间谐波电压","%",time); } return info; diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataHarmRateV.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataHarmRateV.java new file mode 100644 index 000000000..8f87a0d00 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataHarmRateV.java @@ -0,0 +1,26 @@ +package com.njcn.harmonic.service.influxdb; + +import com.github.jeffreyning.mybatisplus.service.IMppService; +import com.njcn.dataProcess.dto.DataHarmrateVDTO; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.pojo.dto.CommonMinuteDto; +import com.njcn.dataProcess.pojo.dto.DataHarmDto; +import com.njcn.dataProcess.pojo.dto.DataHarmRateVDto; +import com.njcn.dataProcess.pojo.po.RStatDataHarmRateVD; + +import java.util.List; + +/** + * @author xy + */ +public interface IDataHarmRateV { + + /** + * 获取原始数据 + * @param lineParam + * @return + */ + List getRawData(LineCountEvaluateParam lineParam); + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataI.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataI.java new file mode 100644 index 000000000..37fa93f51 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataI.java @@ -0,0 +1,26 @@ +package com.njcn.harmonic.service.influxdb; + +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.pojo.dto.DataIDto; + +import java.util.List; + +/** + * Description: + * Date: 2024/11/18 11:17【需求编号】 + * + * @author clam + * @version V1.0.0 + */ +public interface IDataI { + + + /** + * 获取原始数据 + * @param lineParam + * @return + */ + List getRawData(LineCountEvaluateParam lineParam); + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataInHarmV.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataInHarmV.java new file mode 100644 index 000000000..d979a2a57 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataInHarmV.java @@ -0,0 +1,32 @@ +package com.njcn.harmonic.service.influxdb; + +import com.github.jeffreyning.mybatisplus.service.IMppService; +import com.njcn.dataProcess.dto.DataInharmVDTO; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.pojo.dto.CommonMinuteDto; +import com.njcn.dataProcess.pojo.dto.DataHarmDto; +import com.njcn.dataProcess.pojo.dto.DataInHarmVDto; +import com.njcn.dataProcess.pojo.dto.DataVDto; +import com.njcn.dataProcess.pojo.po.RStatDataInHarmVD; + +import java.util.List; + +/** + * Description: + * Date: 2024/11/18 13:27【需求编号】 + * + * @author clam + * @version V1.0.0 + */ +public interface IDataInHarmV { + + + /** + * 获取原始数据 + * @param lineParam + * @return + */ + List getRawData(LineCountEvaluateParam lineParam); + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataPlt.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataPlt.java new file mode 100644 index 000000000..4782845a7 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataPlt.java @@ -0,0 +1,31 @@ +package com.njcn.harmonic.service.influxdb; + +import com.github.jeffreyning.mybatisplus.service.IMppService; +import com.njcn.dataProcess.dto.DataPltDTO; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.pojo.dto.CommonMinuteDto; +import com.njcn.dataProcess.pojo.dto.DataPltDto; +import com.njcn.dataProcess.pojo.dto.DataVDto; +import com.njcn.dataProcess.pojo.po.RStatDataPltD; + +import java.util.List; + +/** + * Description: + * Date: 2024/11/18 11:17【需求编号】 + * + * @author clam + * @version V1.0.0 + */ +public interface IDataPlt { + + + /** + * 获取原始数据 + * @param lineParam + * @return + */ + List getRawData(LineCountEvaluateParam lineParam); + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataV.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataV.java new file mode 100644 index 000000000..6595eb59c --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/IDataV.java @@ -0,0 +1,24 @@ +package com.njcn.harmonic.service.influxdb; + +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.pojo.dto.DataVDto; + +import java.util.List; + +/** + * @author hongawen + * @version 1.0 + * @data 2024/11/7 10:54 + */ +public interface IDataV { + + + /** + * 获取原始数据 + * @param lineParam + * @return + */ + List getRawData(LineCountEvaluateParam lineParam); + + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataHarmRateVImpl.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataHarmRateVImpl.java new file mode 100644 index 000000000..21b1ee82c --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataHarmRateVImpl.java @@ -0,0 +1,158 @@ +package com.njcn.harmonic.service.influxdb.impl; + +import cn.hutool.core.bean.BeanUtil; +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.collection.CollectionUtil; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; +import com.njcn.common.utils.HarmonicTimesUtil; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.po.influx.DataHarmrateV; +import com.njcn.dataProcess.pojo.dto.DataHarmDto; +import com.njcn.harmonic.mapper.influxdb.DataHarmRateVMapper; +import com.njcn.harmonic.service.influxdb.IDataHarmRateV; +import com.njcn.influx.constant.InfluxDbSqlConstant; +import com.njcn.influx.query.InfluxQueryWrapper; +import com.njcn.redis.utils.RedisUtil; +import lombok.RequiredArgsConstructor; +import org.springframework.beans.BeanUtils; +import org.springframework.stereotype.Service; + +import java.lang.reflect.Type; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.stream.Collectors; + +/** + * @author xy + */ +@Service +@RequiredArgsConstructor +public class InfluxdbDataHarmRateVImpl implements IDataHarmRateV { + private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault()); + private final DataHarmRateVMapper dataHarmRateVMapper; + private static final Map PHASE_MAPPING = new HashMap() {{ + put("AB", "A"); + put("BC", "B"); + put("CA", "C"); + put("M", "T"); + }}; + + private final RedisUtil redisUtil; + private static final Set LINE_VOLTAGE_TYPES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("AB", "BC", "CA", "T"))); + private static final Set PHASE_VOLTAGE_TYPES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("A", "B", "C", "T"))); + + @Override + public List getRawData(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + List list = getMinuteData(lineParam); + list.forEach(item->{ + DataHarmDto dto = new DataHarmDto(); + BeanUtils.copyProperties(item,dto); + dto.setMinTime(DATE_TIME_FORMATTER.format(item.getTime())); + result.add(dto); + }); + return result; + } + + + + /** + * 按监测点集合、时间条件获取分钟数据 + * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 + * 需要进行剔除异常数据时,这里会有三种情况判断 + * 1.无异常数据，则直接返回集合； + * 2.异常数据和无异常数据参杂，剔除异常数据，只计算正常数据； + * 3.全是异常数据，则使用异常数据进行计算，但是日表中需要标记出来，此数据有异常 + */ + public List getMinuteData(LineCountEvaluateParam lineParam) { + List dataList; + List result = new ArrayList<>(); + List data = new ArrayList<>(); + //获取监测点、接线方式数据 + Type type = new TypeToken>(){}.getType(); + Map map = new Gson().fromJson( + String.valueOf(redisUtil.getObjectByKey("wlLineDetail")), + type + ); + InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmrateV.class); + influxQueryWrapper.samePrefixAndSuffix(InfluxDbSqlConstant.V, InfluxDbSqlConstant.V, HarmonicTimesUtil.harmonicTimesList(1, 50, 1)); + influxQueryWrapper.regular(DataHarmrateV::getLineId, lineParam.getLineId()) + .select(DataHarmrateV::getLineId) + .select(DataHarmrateV::getPhasicType) + .select(DataHarmrateV::getValueType) + .select(DataHarmrateV::getQualityFlag) + .select(DataHarmrateV::getAbnormalFlag) + .between(DataHarmrateV::getTime, lineParam.getStartTime(), lineParam.getEndTime()) + .eq(DataHarmrateV::getQualityFlag,"0"); + if(CollUtil.isNotEmpty(lineParam.getPhasicType())){ + influxQueryWrapper.regular(DataHarmrateV::getPhasicType,lineParam.getPhasicType()); + } + List list = dataHarmRateVMapper.selectByQueryWrapper(influxQueryWrapper); + if(CollUtil.isNotEmpty(list)){ + //过滤掉暂态事件影响的数据 true过滤 false不过滤 + if (lineParam.getDataType()) { + dataList = list.stream().filter(item -> Objects.isNull(item.getAbnormalFlag())).collect(Collectors.toList()); + } else { + dataList = list; + } + Map> lineMap = dataList.stream().collect(Collectors.groupingBy(DataHarmrateV::getLineId)); + //有异常数据 + if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) { + lineMap.forEach((k,v)->{ + List timeList = lineParam.getAbnormalTime().get(k); + //有异常数据,当前监测点自身的异常数据 + if (CollectionUtil.isNotEmpty(timeList)) { + List filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()); + //1.过滤掉异常数据后还有正常数据，则用正常数据计算 + if (CollectionUtil.isNotEmpty(filterList)) { + result.addAll(filterList); + } + //2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的 + else { + v.parallelStream().forEach(item -> item.setQualityFlag("1")); + result.addAll(v); + } + } + //没有异常数据，则使用原数据 + else { + result.addAll(v); + } + }); + } + //没有异常数据，则使用原数据 + else { + result.addAll(dataList); + } + } + if (CollectionUtil.isNotEmpty(result)) { + if (!Objects.isNull(map)) { + //现根据监测点分组，然后根据接线方式排除多于数据，在修改相别 + Map> lineMap = result.stream().collect(Collectors.groupingBy(DataHarmrateV::getLineId)); + lineMap.forEach((k,v)->{ + if (Objects.isNull(map.get(k))) { + return; + } + Integer conType = map.get(k); + Set validPhasicTypes = (conType != 0) ? LINE_VOLTAGE_TYPES : PHASE_VOLTAGE_TYPES; + List result2 = v.stream().filter(item -> validPhasicTypes.contains(item.getPhasicType())).collect(Collectors.toList()); + data.addAll(result2); + }); + } else { + data.addAll(result); + } + if (CollectionUtil.isNotEmpty(data)) { + data.forEach(item -> { + String newType = PHASE_MAPPING.get(item.getPhasicType()); + if (newType != null) { + item.setPhasicType(newType); + } + }); + } + } + return data; + } +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataIImpl.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataIImpl.java new file mode 100644 index 000000000..8ccbd9f23 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataIImpl.java @@ -0,0 +1,135 @@ +package com.njcn.harmonic.service.influxdb.impl; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.collection.CollectionUtil; +import com.njcn.common.utils.HarmonicTimesUtil; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.po.influx.DataI; +import com.njcn.dataProcess.pojo.dto.DataIDto; +import com.njcn.harmonic.mapper.influxdb.DataIMapper; +import com.njcn.harmonic.service.influxdb.IDataI; +import com.njcn.influx.constant.InfluxDbSqlConstant; +import com.njcn.influx.query.InfluxQueryWrapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.BeanUtils; +import org.springframework.stereotype.Service; + +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.stream.Collectors; + + +/** + * @author wr + * @description + * @date 2026/7/1 10:49 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class InfluxdbDataIImpl implements IDataI { + private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault()); + + private final DataIMapper dataIMapper; + + private static final Map PHASE_MAPPING = new HashMap() {{ + put("AB", "A"); + put("BC", "B"); + put("CA", "C"); + put("M", "T"); + }}; + + @Override + public List getRawData(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + List list = getMinuteDataI(lineParam);; + list.forEach(item->{ + DataIDto dto = new DataIDto(); + BeanUtils.copyProperties(item,dto); + dto.setMinTime(DATE_TIME_FORMATTER.format(item.getTime())); + result.add(dto); + }); + return result; + } + + /** + * 按监测点集合、时间条件获取dataI分钟数据 + * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 + * 需要进行剔除异常数据时,这里会有三种情况判断 + * 1.无异常数据，则直接返回集合； + * 2.异常数据和无异常数据参杂，剔除异常数据，只计算正常数据； + * 3.全是异常数据，则使用异常数据进行计算，但是日表中需要标记出来，此数据有异常 + */ + public List getMinuteDataI(LineCountEvaluateParam lineParam) { + List dataList; + List result = new ArrayList<>(); + InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataI.class); + influxQueryWrapper.samePrefixAndSuffix(InfluxDbSqlConstant.I, InfluxDbSqlConstant.I, HarmonicTimesUtil.harmonicTimesList(1, 50, 1)); + influxQueryWrapper.regular(DataI::getLineId, lineParam.getLineId()) + .select(DataI::getLineId) + .select(DataI::getPhasicType) + .select(DataI::getValueType) + .select(DataI::getINeg) + .select(DataI::getIPos) + .select(DataI::getIThd) + .select(DataI::getIUnbalance) + .select(DataI::getIZero) + .select(DataI::getRms) + .select(DataI::getQualityFlag) + .select(DataI::getAbnormalFlag) + .between(DataI::getTime, lineParam.getStartTime(), lineParam.getEndTime()) + .eq(DataI::getQualityFlag,"0"); + if(CollUtil.isNotEmpty(lineParam.getPhasicType())){ + influxQueryWrapper.regular(DataI::getPhasicType,lineParam.getPhasicType()); + } + + List list = dataIMapper.selectByQueryWrapper(influxQueryWrapper); + if(CollUtil.isNotEmpty(list)){ + //过滤掉暂态事件影响的数据 true过滤 false不过滤 + if (lineParam.getDataType()) { + dataList = list.stream().filter(item -> Objects.isNull(item.getAbnormalFlag())).collect(Collectors.toList()); + } else { + dataList = list; + } + Map> lineMap = dataList.stream().collect(Collectors.groupingBy(DataI::getLineId)); + //有异常数据 + if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) { + lineMap.forEach((k,v)->{ + List timeList = lineParam.getAbnormalTime().get(k); + //有异常数据,当前监测点自身的异常数据 + if (CollectionUtil.isNotEmpty(timeList)) { + List filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()); + //1.过滤掉异常数据后还有正常数据，则用正常数据计算 + if (CollectionUtil.isNotEmpty(filterList)) { + result.addAll(filterList); + } + //2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的 + else { + v.parallelStream().forEach(item -> item.setQualityFlag("1")); + result.addAll(v); + } + } + //没有异常数据，则使用原数据 + else { + result.addAll(v); + } + }); + } + //没有异常数据，则使用原数据 + else { + result.addAll(dataList); + } + } + if (CollectionUtil.isNotEmpty(result)) { + result.forEach(item -> { + String newType = PHASE_MAPPING.get(item.getPhasicType()); + if (newType != null) { + item.setPhasicType(newType); + } + }); + } + return result; + } +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataInharmVImpl.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataInharmVImpl.java new file mode 100644 index 000000000..a86568896 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataInharmVImpl.java @@ -0,0 +1,132 @@ +package com.njcn.harmonic.service.influxdb.impl; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.collection.CollectionUtil; +import com.njcn.common.utils.HarmonicTimesUtil; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.po.influx.DataInharmV; +import com.njcn.dataProcess.pojo.dto.DataHarmDto; +import com.njcn.harmonic.mapper.influxdb.DataInharmVMapper; +import com.njcn.harmonic.service.influxdb.IDataInHarmV; +import com.njcn.influx.constant.InfluxDbSqlConstant; +import com.njcn.influx.query.InfluxQueryWrapper; +import lombok.RequiredArgsConstructor; +import org.springframework.beans.BeanUtils; +import org.springframework.stereotype.Service; + +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.stream.Collectors; + + +/** + * Description: + * Date: 2024/11/18 14:33【需求编号】 + * + * @author clam + * @version V1.0.0 + */ +@Service +@RequiredArgsConstructor +public class InfluxdbDataInharmVImpl implements IDataInHarmV { + + private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault()); + private final DataInharmVMapper dataInharmVMapper; + + private static final Map PHASE_MAPPING = new HashMap() {{ + put("AB", "A"); + put("BC", "B"); + put("CA", "C"); + put("M", "T"); + }}; + + + + @Override + public List getRawData(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + List list = getMinuteData(lineParam); + list.forEach(item->{ + DataHarmDto dto = new DataHarmDto(); + BeanUtils.copyProperties(item,dto); + dto.setMinTime(DATE_TIME_FORMATTER.format(item.getTime())); + result.add(dto); + }); + return result; + } + + + + /** + * 按监测点集合、时间条件获取dataI分钟数据 + * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 + * 需要进行剔除异常数据时,这里会有三种情况判断 + * 1.无异常数据，则直接返回集合； + * 2.异常数据和无异常数据参杂，剔除异常数据，只计算正常数据； + * 3.全是异常数据，则使用异常数据进行计算，但是日表中需要标记出来，此数据有异常 + */ + public List getMinuteData(LineCountEvaluateParam lineParam) { + List dataList; + List result = new ArrayList<>(); + InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataInharmV.class); + influxQueryWrapper.samePrefixAndSuffix(InfluxDbSqlConstant.V, InfluxDbSqlConstant.V, HarmonicTimesUtil.harmonicTimesList(1, 50, 1)); + influxQueryWrapper.regular(DataInharmV::getLineId, lineParam.getLineId()) + .select(DataInharmV::getLineId) + .select(DataInharmV::getPhasicType) + .select(DataInharmV::getValueType) + .select(DataInharmV::getQualityFlag) + .select(DataInharmV::getAbnormalFlag) + .between(DataInharmV::getTime, lineParam.getStartTime(), lineParam.getEndTime()) + .eq(DataInharmV::getQualityFlag,"0"); + if(CollUtil.isNotEmpty(lineParam.getPhasicType())){ + influxQueryWrapper.regular(DataInharmV::getPhasicType,lineParam.getPhasicType()); + } + List list = dataInharmVMapper.selectByQueryWrapper(influxQueryWrapper); + if(CollUtil.isNotEmpty(list)){ + //过滤掉暂态事件影响的数据 true过滤 false不过滤 + if (lineParam.getDataType()) { + dataList = list.stream().filter(item -> Objects.isNull(item.getAbnormalFlag())).collect(Collectors.toList()); + } else { + dataList = list; + } + Map> lineMap = dataList.stream().collect(Collectors.groupingBy(DataInharmV::getLineId)); + //有异常数据 + if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) { + lineMap.forEach((k,v)->{ + List timeList = lineParam.getAbnormalTime().get(k); + //有异常数据,当前监测点自身的异常数据 + if (CollectionUtil.isNotEmpty(timeList)) { + List filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()); + //1.过滤掉异常数据后还有正常数据，则用正常数据计算 + if (CollectionUtil.isNotEmpty(filterList)) { + result.addAll(filterList); + } + //2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的 + else { + v.parallelStream().forEach(item -> item.setQualityFlag("1")); + result.addAll(v); + } + } + //没有异常数据，则使用原数据 + else { + result.addAll(v); + } + }); + } + //没有异常数据，则使用原数据 + else { + result.addAll(dataList); + } + } + if (CollectionUtil.isNotEmpty(result)) { + result.forEach(item -> { + String newType = PHASE_MAPPING.get(item.getPhasicType()); + if (newType != null) { + item.setPhasicType(newType); + } + }); + } + return result; + } +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataPltImpl.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataPltImpl.java new file mode 100644 index 000000000..d9b43d536 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataPltImpl.java @@ -0,0 +1,157 @@ +package com.njcn.harmonic.service.influxdb.impl; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.collection.CollectionUtil; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.po.influx.DataPlt; +import com.njcn.dataProcess.po.influx.DataV; +import com.njcn.dataProcess.pojo.dto.DataPltDto; +import com.njcn.harmonic.mapper.influxdb.DataPltMapper; +import com.njcn.harmonic.service.influxdb.IDataPlt; +import com.njcn.influx.query.InfluxQueryWrapper; +import com.njcn.redis.utils.RedisUtil; +import lombok.RequiredArgsConstructor; +import org.springframework.beans.BeanUtils; +import org.springframework.stereotype.Service; + +import java.lang.reflect.Type; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.stream.Collectors; + +/** + * Description: + * Date: 2024/11/18 14:33【需求编号】 + * + * @author clam + * @version V1.0.0 + */ +@Service +@RequiredArgsConstructor +public class InfluxdbDataPltImpl implements IDataPlt { + + private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault()); + private final DataPltMapper dataPltMapper; + private final RedisUtil redisUtil; + private static final Map PHASE_MAPPING = new HashMap() {{ + put("AB", "A"); + put("BC", "B"); + put("CA", "C"); + put("M", "T"); + }}; + private static final Set LINE_VOLTAGE_TYPES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("AB", "BC", "CA", "T"))); + private static final Set PHASE_VOLTAGE_TYPES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("A", "B", "C", "T"))); + + @Override + public List getRawData(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + List list = getMinuteDataPlt(lineParam); + list.forEach(item->{ + DataPltDto dto = new DataPltDto(); + BeanUtils.copyProperties(item,dto); + dto.setMinTime(DATE_TIME_FORMATTER.format(item.getTime())); + result.add(dto); + }); + return result; + } + /** + * 按监测点集合、时间条件获取dataI分钟数据 + * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 + * 需要进行剔除异常数据时,这里会有三种情况判断 + * 1.无异常数据，则直接返回集合； + * 2.异常数据和无异常数据参杂，剔除异常数据，只计算正常数据； + * 3.全是异常数据，则使用异常数据进行计算，但是日表中需要标记出来，此数据有异常 + */ + public List getMinuteDataPlt(LineCountEvaluateParam lineParam) { + List dataList; + List result = new ArrayList<>(); + List data = new ArrayList<>(); + //获取监测点、接线方式数据 + Type type = new TypeToken>(){}.getType(); + Map map = new Gson().fromJson( + String.valueOf(redisUtil.getObjectByKey("wlLineDetail")), + type + ); + InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataPlt.class); + influxQueryWrapper.regular(DataPlt::getLineId, lineParam.getLineId()) + .select(DataPlt::getLineId) + .select(DataPlt::getPhasicType) + .select(DataPlt::getPlt) + .select(DataPlt::getQualityFlag) + .select(DataPlt::getAbnormalFlag) + .between(DataPlt::getTime, lineParam.getStartTime(), lineParam.getEndTime()) + .eq(DataPlt::getQualityFlag,"0"); + if(CollUtil.isNotEmpty(lineParam.getPhasicType())){ + influxQueryWrapper.regular(DataV::getPhasicType,lineParam.getPhasicType()); + } + List list = dataPltMapper.selectByQueryWrapper(influxQueryWrapper); + if(CollUtil.isNotEmpty(list)){ + //过滤掉暂态事件影响的数据 true过滤 false不过滤 + if (lineParam.getDataType()) { + dataList = list.stream().filter(item -> Objects.isNull(item.getAbnormalFlag())).collect(Collectors.toList()); + } else { + dataList = list; + } + Map> lineMap = dataList.stream().collect(Collectors.groupingBy(DataPlt::getLineId)); + //有异常数据 + if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) { + lineMap.forEach((k,v)->{ + List timeList = lineParam.getAbnormalTime().get(k); + //有异常数据,当前监测点自身的异常数据 + if (CollectionUtil.isNotEmpty(timeList)) { + List filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()); + //1.过滤掉异常数据后还有正常数据，则用正常数据计算 + if (CollectionUtil.isNotEmpty(filterList)) { + result.addAll(filterList); + } + //2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的 + else { + v.parallelStream().forEach(item -> item.setQualityFlag("1")); + result.addAll(v); + } + } + //没有异常数据，则使用原数据 + else { + result.addAll(v); + } + }); + } + //没有异常数据，则使用原数据 + else { + result.addAll(dataList); + } + } + if (CollectionUtil.isNotEmpty(result)) { + if (!Objects.isNull(map)) { + //现根据监测点分组，然后根据接线方式排除多于数据，在修改相别 + Map> lineMap = result.stream().collect(Collectors.groupingBy(DataPlt::getLineId)); + lineMap.forEach((k,v)->{ + if (Objects.isNull(map.get(k))) { + return; + } + Integer conType = map.get(k); + Set validPhasicTypes = (conType != 0) ? LINE_VOLTAGE_TYPES : PHASE_VOLTAGE_TYPES; + List result2 = v.stream().filter(item -> validPhasicTypes.contains(item.getPhasicType())).collect(Collectors.toList()); + data.addAll(result2); + }); + } else { + data.addAll(result); + } + if (CollectionUtil.isNotEmpty(data)) { + data.forEach(item -> { + String newType = PHASE_MAPPING.get(item.getPhasicType()); + if (newType != null) { + item.setPhasicType(newType); + } + }); + } + } + return data; + } + +} diff --git a/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataVImpl.java b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataVImpl.java new file mode 100644 index 000000000..037213150 --- /dev/null +++ b/pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/influxdb/impl/InfluxdbDataVImpl.java @@ -0,0 +1,221 @@ +package com.njcn.harmonic.service.influxdb.impl; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.collection.CollectionUtil; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; +import com.njcn.common.utils.HarmonicTimesUtil; +import com.njcn.dataProcess.param.LineCountEvaluateParam; +import com.njcn.dataProcess.pojo.dto.DataVDto; +import com.njcn.harmonic.mapper.influxdb.DataVMapper; +import com.njcn.harmonic.service.influxdb.IDataV; +import com.njcn.influx.constant.InfluxDbSqlConstant; +import com.njcn.dataProcess.po.influx.DataV; +import com.njcn.influx.query.InfluxQueryWrapper; +import com.njcn.redis.utils.RedisUtil; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.BeanUtils; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.lang.reflect.Type; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * @author wr + * @description + * @date 2026/7/1 10:49 + */ +@Slf4j +@Service +public class InfluxdbDataVImpl implements IDataV { + + private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault()); + private static final Map PHASE_MAPPING = new HashMap() {{ + put("AB", "A"); + put("BC", "B"); + put("CA", "C"); + put("M", "T"); + }}; + @Resource + private DataVMapper dataVMapper; + @Resource + private RedisUtil redisUtil; + private static final Set LINE_VOLTAGE_TYPES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("AB", "BC", "CA", "T"))); + private static final Set PHASE_VOLTAGE_TYPES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("A", "B", "C", "T"))); + + @Override + public List getRawData(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + List list = getMinuteDataV(lineParam); + if (CollectionUtil.isNotEmpty(list)) { + list.forEach(item -> { + DataVDto dto = new DataVDto(); + BeanUtils.copyProperties(item, dto); + dto.setMinTime(DATE_TIME_FORMATTER.format(item.getTime())); + result.add(dto); + }); + } + return result; + } + + /** + * 按监测点集合、时间条件获取dataV分钟数据 + * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 + * 剔除异常数据,这里会有三种情况判断 + * 1.无异常数据，则直接返回集合； + * 2.异常数据和无异常数据参杂，剔除异常数据，只计算正常数据； + * 3.全是异常数据，则使用异常数据进行计算，但是日表中需要标记出来，此数据有异常 + */ + public List getMinuteDataV(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + List data = new ArrayList<>(); + //获取监测点、接线方式数据 + Type type = new TypeToken>(){}.getType(); + Map map = new Gson().fromJson( + String.valueOf(redisUtil.getObjectByKey("wlLineDetail")), + type + ); + InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class); + influxQueryWrapper.samePrefixAndSuffix(InfluxDbSqlConstant.V, InfluxDbSqlConstant.V, HarmonicTimesUtil.harmonicTimesList(1, 50, 1)); + influxQueryWrapper.regular(DataV::getLineId, lineParam.getLineId()) + .select(DataV::getLineId) + .select(DataV::getPhasicType) + .select(DataV::getValueType) + .select(DataV::getFreq) + .select(DataV::getFreqDev) + .select(DataV::getRms) + .select(DataV::getRmsLvr) + .select(DataV::getVNeg) + .select(DataV::getVPos) + .select(DataV::getVThd) + .select(DataV::getVUnbalance) + .select(DataV::getVZero) + .select(DataV::getVlDev) + .select(DataV::getVuDev) + .select(DataV::getQualityFlag) + .select(DataV::getAbnormalFlag) + .between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime()) + .eq(DataV::getQualityFlag, "0"); + if (CollUtil.isNotEmpty(lineParam.getPhasicType())) { + influxQueryWrapper.regular(DataV::getPhasicType, lineParam.getPhasicType()); + } + quality(result, influxQueryWrapper, lineParam); + if (CollectionUtil.isNotEmpty(result)) { + if (!Objects.isNull(map)) { + //现根据监测点分组，然后根据接线方式排除多于数据，在修改相别 + Map> lineMap = result.stream().collect(Collectors.groupingBy(DataV::getLineId)); + lineMap.forEach((k,v)->{ + if (Objects.isNull(map.get(k))) { + return; + } + //这边需要特殊处理下，将线电压数据赋值 + Map lineVoltageIndex = v.stream() + .filter(d -> PHASE_MAPPING.containsKey(d.getPhasicType())) + .filter(d -> d.getRmsLvr() != null) + .collect(Collectors.toMap( + d -> buildKey(d.getTime(), d.getValueType(), d.getPhasicType()), + Function.identity(), + (existing, replacement) -> existing + )); + v.stream() + .filter(d -> PHASE_VOLTAGE_TYPES.contains(d.getPhasicType())) + .forEach(phaseData -> { + // 根据当前相电压反查对应的线电压相别 + String targetLinePhasic = getReverseLinePhasic(phaseData.getPhasicType()); + if (targetLinePhasic == null) { + return; + } + String key = buildKey(phaseData.getTime(), phaseData.getValueType(), targetLinePhasic); + DataV matchedLineData = lineVoltageIndex.get(key); + if (matchedLineData != null && matchedLineData.getRmsLvr() != null) { + phaseData.setRmsLvr(matchedLineData.getRmsLvr()); + } + }); + Integer conType = map.get(k); + Set validPhasicTypes = (conType != 0) ? LINE_VOLTAGE_TYPES : PHASE_VOLTAGE_TYPES; + List result2 = v.stream().filter(item -> validPhasicTypes.contains(item.getPhasicType())).collect(Collectors.toList()); + data.addAll(result2); + }); + } else { + data.addAll(result); + } + if (CollectionUtil.isNotEmpty(data)) { + data.forEach(item -> { + String newType = PHASE_MAPPING.get(item.getPhasicType()); + if (newType != null) { + item.setPhasicType(newType); + } + }); + } + } + return data; + } + private void quality(List result, InfluxQueryWrapper influxQueryWrapper, LineCountEvaluateParam lineParam) { + List dataList; + List list = dataVMapper.selectByQueryWrapper(influxQueryWrapper); + if (CollUtil.isNotEmpty(list)) { + //过滤掉暂态事件影响的数据 true过滤 false不过滤 + if (lineParam.getDataType()) { + dataList = list.stream().filter(item -> Objects.isNull(item.getAbnormalFlag())).collect(Collectors.toList()); + } else { + dataList = list; + } + Map> lineMap = dataList.stream().collect(Collectors.groupingBy(DataV::getLineId)); + //有异常数据 + Map> timeMap = lineParam.getAbnormalTime(); + if (CollectionUtil.isNotEmpty(timeMap)) { + lineMap.forEach((k, v) -> { + List timeList = timeMap.get(k); + //有异常数据,当前监测点自身的异常数据 + if (CollectionUtil.isNotEmpty(timeList)) { + List filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()); + //1.过滤掉异常数据后还有正常数据，则用正常数据计算 + if (CollectionUtil.isNotEmpty(filterList)) { + result.addAll(filterList); + } + //2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的 + else { + v.parallelStream().forEach(item -> item.setQualityFlag("1")); + result.addAll(v); + } + } + //没有异常数据，则使用原数据 + else { + result.addAll(v); + } + }); + } + //没有异常数据，则使用原数据 + else { + result.addAll(dataList); + } + } + } + + private static String buildKey(Object time, Object valueType, Object phasicType) { + return time + "|" + valueType + "|" + phasicType; + } + + private static String getReverseLinePhasic(String phaseType) { + if (phaseType == null) { + return null; + } + switch (phaseType) { + case "A": + return "AB"; + case "B": + return "BC"; + case "C": + return "CA"; + default: + return null; + } + } +}