1.解决有功功率趋势统计调用ataprocessboot接口改造
This commit is contained in:
@@ -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<DataHarmrateI> {
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataHarmrateV> {
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataI> {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
package com.njcn.harmonic.mapper.influxdb;
|
||||||
|
|
||||||
|
|
||||||
|
import com.njcn.dataProcess.po.influx.DataInharmV;
|
||||||
|
import com.njcn.influx.base.InfluxDbBaseMapper;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* <p>
|
||||||
|
* Mapper 接口
|
||||||
|
* </p>
|
||||||
|
*
|
||||||
|
* @author hongawen
|
||||||
|
* @since 2023-12-28
|
||||||
|
*/
|
||||||
|
public interface DataInharmVMapper extends InfluxDbBaseMapper<DataInharmV> {
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
package com.njcn.harmonic.mapper.influxdb;
|
||||||
|
|
||||||
|
|
||||||
|
import com.njcn.dataProcess.po.influx.DataPlt;
|
||||||
|
import com.njcn.influx.base.InfluxDbBaseMapper;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* <p>
|
||||||
|
* Mapper 接口
|
||||||
|
* </p>
|
||||||
|
*
|
||||||
|
* @author hongawen
|
||||||
|
* @since 2023-12-28
|
||||||
|
*/
|
||||||
|
public interface DataPltMapper extends InfluxDbBaseMapper<DataPlt> {
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataV> {
|
||||||
|
|
||||||
|
|
||||||
|
List<LineDataVFiveItemDTO> queryDataValue(InfluxQueryWrapper dataVQueryWrapper);
|
||||||
|
|
||||||
|
|
||||||
|
List<MeasurementCountDTO> getMeasurementCount(InfluxQueryWrapper influxQueryWrapper);
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -10,7 +10,6 @@ import com.alibaba.excel.EasyExcel;
|
|||||||
import com.alibaba.excel.ExcelWriter;
|
import com.alibaba.excel.ExcelWriter;
|
||||||
import com.alibaba.excel.write.metadata.WriteSheet;
|
import com.alibaba.excel.write.metadata.WriteSheet;
|
||||||
import com.alibaba.fastjson.JSONArray;
|
import com.alibaba.fastjson.JSONArray;
|
||||||
import com.njcn.dataProcess.api.*;
|
|
||||||
import com.njcn.dataProcess.param.LineCountEvaluateParam;
|
import com.njcn.dataProcess.param.LineCountEvaluateParam;
|
||||||
import com.njcn.dataProcess.pojo.dto.*;
|
import com.njcn.dataProcess.pojo.dto.*;
|
||||||
import com.njcn.device.biz.pojo.po.Overlimit;
|
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.IDataLimitRateDetailService;
|
||||||
import com.njcn.harmonic.service.activepowerrange.PowerStatisticsService;
|
import com.njcn.harmonic.service.activepowerrange.PowerStatisticsService;
|
||||||
import com.njcn.harmonic.service.activepowerrange.RActivePowerRangeService;
|
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.poi.util.PoiUtil;
|
||||||
import com.njcn.system.api.EpdFeignClient;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import org.apache.poi.ss.usermodel.*;
|
import org.apache.poi.ss.usermodel.*;
|
||||||
import org.apache.poi.xssf.usermodel.XSSFClientAnchor;
|
import org.apache.poi.xssf.usermodel.XSSFClientAnchor;
|
||||||
@@ -56,11 +54,11 @@ public class PowerStatisticsServiceImpl implements PowerStatisticsService {
|
|||||||
|
|
||||||
private final RActivePowerRangeService rActivePowerRangeService;
|
private final RActivePowerRangeService rActivePowerRangeService;
|
||||||
private final DecimalFormat dftwo = new DecimalFormat(Param.DECIMAL_FORMATTWOSTR);
|
private final DecimalFormat dftwo = new DecimalFormat(Param.DECIMAL_FORMATTWOSTR);
|
||||||
private final DataVFeignClient dataVFeignClient;
|
private final IDataV dataV;
|
||||||
private final DataIFeignClient dataIFeignClient;
|
private final IDataI dataI;
|
||||||
private final DataPltFeignClient dataPltFeignClient;
|
private final IDataPlt dataPlt;
|
||||||
private final DataInharmVFeignClient dataInharmVFeignClient;
|
private final IDataInHarmV dataInHarmV;
|
||||||
private final DataHarmRateVFeignClient dataHarmRateVFeignClient;
|
private final IDataHarmRateV dataHarmRateV;
|
||||||
private final OverLimitClient overLimitClient;
|
private final OverLimitClient overLimitClient;
|
||||||
private final IDataLimitRateDetailService iDataLimitRateDetailService;
|
private final IDataLimitRateDetailService iDataLimitRateDetailService;
|
||||||
|
|
||||||
@@ -156,11 +154,11 @@ public class PowerStatisticsServiceImpl implements PowerStatisticsService {
|
|||||||
String time = param.getSearchBeginTime();
|
String time = param.getSearchBeginTime();
|
||||||
if ("1".equals(param.getStatisticalId())) {
|
if ("1".equals(param.getStatisticalId())) {
|
||||||
//电压数据
|
//电压数据
|
||||||
List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(evaluateParam).getData();
|
List<DataVDto> dataVAllTime = dataV.getRawData(evaluateParam);
|
||||||
//闪变数据
|
//闪变数据
|
||||||
List<DataPltDto> dataFlickerAllTime = dataPltFeignClient.getRawData(evaluateParam).getData();
|
List<DataPltDto> dataFlickerAllTime = dataPlt.getRawData(evaluateParam);
|
||||||
//电流数据
|
//电流数据
|
||||||
List<DataIDto> dataIList = dataIFeignClient.getRawData(evaluateParam).getData();
|
List<DataIDto> dataIList = dataI.getRawData(evaluateParam);
|
||||||
//电压偏差
|
//电压偏差
|
||||||
if ("Dev".equals(param.getCode()) || StrUtil.isBlank(param.getCode())) {
|
if ("Dev".equals(param.getCode()) || StrUtil.isBlank(param.getCode())) {
|
||||||
addThdData(info, overlimit.getVoltageDev(), "vuDev", dataVAllTime, "电压上偏差","%",time);
|
addThdData(info, overlimit.getVoltageDev(), "vuDev", dataVAllTime, "电压上偏差","%",time);
|
||||||
@@ -190,17 +188,17 @@ public class PowerStatisticsServiceImpl implements PowerStatisticsService {
|
|||||||
}
|
}
|
||||||
if ("2".equals(param.getStatisticalId())) {
|
if ("2".equals(param.getStatisticalId())) {
|
||||||
//谐波数据
|
//谐波数据
|
||||||
List<DataHarmDto> dataVHarmList = dataHarmRateVFeignClient.getRawData(evaluateParam).getData();
|
List<DataHarmDto> dataVHarmList = dataHarmRateV.getRawData(evaluateParam);
|
||||||
addThdData(info, overlimit, "getUharm", "v", 2, 26, dataVHarmList, "谐波电压","%",time);
|
addThdData(info, overlimit, "getUharm", "v", 2, 26, dataVHarmList, "谐波电压","%",time);
|
||||||
}
|
}
|
||||||
if ("3".equals(param.getStatisticalId())) {
|
if ("3".equals(param.getStatisticalId())) {
|
||||||
//电流数据
|
//电流数据
|
||||||
List<DataIDto> dataIList = dataIFeignClient.getRawData(evaluateParam).getData();
|
List<DataIDto> dataIList = dataI.getRawData(evaluateParam);
|
||||||
addThdData(info, overlimit, "getUharm", "i", 2, 26, dataIList, "谐波电流","A",time);
|
addThdData(info, overlimit, "getUharm", "i", 2, 26, dataIList, "谐波电流","A",time);
|
||||||
}
|
}
|
||||||
if ("4".equals(param.getStatisticalId())) {
|
if ("4".equals(param.getStatisticalId())) {
|
||||||
//间谐波数据
|
//间谐波数据
|
||||||
List<DataHarmDto> dataVInHarmList = dataInharmVFeignClient.getRawData(evaluateParam).getData();
|
List<DataHarmDto> dataVInHarmList = dataInHarmV.getRawData(evaluateParam);
|
||||||
addThdData(info, overlimit, "getInuharm", "v", 1, 17, dataVInHarmList, "间谐波电压","%",time);
|
addThdData(info, overlimit, "getInuharm", "v", 1, 17, dataVInHarmList, "间谐波电压","%",time);
|
||||||
}
|
}
|
||||||
return info;
|
return info;
|
||||||
|
|||||||
@@ -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<DataHarmDto> getRawData(LineCountEvaluateParam lineParam);
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataIDto> getRawData(LineCountEvaluateParam lineParam);
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataHarmDto> getRawData(LineCountEvaluateParam lineParam);
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataPltDto> getRawData(LineCountEvaluateParam lineParam);
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<DataVDto> getRawData(LineCountEvaluateParam lineParam);
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<String, String> PHASE_MAPPING = new HashMap<String, String>() {{
|
||||||
|
put("AB", "A");
|
||||||
|
put("BC", "B");
|
||||||
|
put("CA", "C");
|
||||||
|
put("M", "T");
|
||||||
|
}};
|
||||||
|
|
||||||
|
private final RedisUtil redisUtil;
|
||||||
|
private static final Set<String> LINE_VOLTAGE_TYPES =
|
||||||
|
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("AB", "BC", "CA", "T")));
|
||||||
|
private static final Set<String> PHASE_VOLTAGE_TYPES =
|
||||||
|
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("A", "B", "C", "T")));
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<DataHarmDto> getRawData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataHarmDto> result = new ArrayList<>();
|
||||||
|
List<DataHarmrateV> 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<DataHarmrateV> getMinuteData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataHarmrateV> dataList;
|
||||||
|
List<DataHarmrateV> result = new ArrayList<>();
|
||||||
|
List<DataHarmrateV> data = new ArrayList<>();
|
||||||
|
//获取监测点、接线方式数据
|
||||||
|
Type type = new TypeToken<Map<String, Integer>>(){}.getType();
|
||||||
|
Map<String, Integer> 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<DataHarmrateV> 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<String,List<DataHarmrateV>> lineMap = dataList.stream().collect(Collectors.groupingBy(DataHarmrateV::getLineId));
|
||||||
|
//有异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) {
|
||||||
|
lineMap.forEach((k,v)->{
|
||||||
|
List<String> timeList = lineParam.getAbnormalTime().get(k);
|
||||||
|
//有异常数据,当前监测点自身的异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(timeList)) {
|
||||||
|
List<DataHarmrateV> 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<String, List<DataHarmrateV>> 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<String> validPhasicTypes = (conType != 0) ? LINE_VOLTAGE_TYPES : PHASE_VOLTAGE_TYPES;
|
||||||
|
List<DataHarmrateV> 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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<String, String> PHASE_MAPPING = new HashMap<String, String>() {{
|
||||||
|
put("AB", "A");
|
||||||
|
put("BC", "B");
|
||||||
|
put("CA", "C");
|
||||||
|
put("M", "T");
|
||||||
|
}};
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<DataIDto> getRawData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataIDto> result = new ArrayList<>();
|
||||||
|
List<DataI> 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<DataI> getMinuteDataI(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataI> dataList;
|
||||||
|
List<DataI> 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<DataI> 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<String,List<DataI>> lineMap = dataList.stream().collect(Collectors.groupingBy(DataI::getLineId));
|
||||||
|
//有异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) {
|
||||||
|
lineMap.forEach((k,v)->{
|
||||||
|
List<String> timeList = lineParam.getAbnormalTime().get(k);
|
||||||
|
//有异常数据,当前监测点自身的异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(timeList)) {
|
||||||
|
List<DataI> 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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<String, String> PHASE_MAPPING = new HashMap<String, String>() {{
|
||||||
|
put("AB", "A");
|
||||||
|
put("BC", "B");
|
||||||
|
put("CA", "C");
|
||||||
|
put("M", "T");
|
||||||
|
}};
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<DataHarmDto> getRawData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataHarmDto> result = new ArrayList<>();
|
||||||
|
List<DataInharmV> 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<DataInharmV> getMinuteData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataInharmV> dataList;
|
||||||
|
List<DataInharmV> 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<DataInharmV> 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<String,List<DataInharmV>> lineMap = dataList.stream().collect(Collectors.groupingBy(DataInharmV::getLineId));
|
||||||
|
//有异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) {
|
||||||
|
lineMap.forEach((k,v)->{
|
||||||
|
List<String> timeList = lineParam.getAbnormalTime().get(k);
|
||||||
|
//有异常数据,当前监测点自身的异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(timeList)) {
|
||||||
|
List<DataInharmV> 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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<String, String> PHASE_MAPPING = new HashMap<String, String>() {{
|
||||||
|
put("AB", "A");
|
||||||
|
put("BC", "B");
|
||||||
|
put("CA", "C");
|
||||||
|
put("M", "T");
|
||||||
|
}};
|
||||||
|
private static final Set<String> LINE_VOLTAGE_TYPES =
|
||||||
|
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("AB", "BC", "CA", "T")));
|
||||||
|
private static final Set<String> PHASE_VOLTAGE_TYPES =
|
||||||
|
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("A", "B", "C", "T")));
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<DataPltDto> getRawData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataPltDto> result = new ArrayList<>();
|
||||||
|
List<DataPlt> 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<DataPlt> getMinuteDataPlt(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataPlt> dataList;
|
||||||
|
List<DataPlt> result = new ArrayList<>();
|
||||||
|
List<DataPlt> data = new ArrayList<>();
|
||||||
|
//获取监测点、接线方式数据
|
||||||
|
Type type = new TypeToken<Map<String, Integer>>(){}.getType();
|
||||||
|
Map<String, Integer> 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<DataPlt> 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<String,List<DataPlt>> lineMap = dataList.stream().collect(Collectors.groupingBy(DataPlt::getLineId));
|
||||||
|
//有异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(lineParam.getAbnormalTime())) {
|
||||||
|
lineMap.forEach((k,v)->{
|
||||||
|
List<String> timeList = lineParam.getAbnormalTime().get(k);
|
||||||
|
//有异常数据,当前监测点自身的异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(timeList)) {
|
||||||
|
List<DataPlt> 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<String, List<DataPlt>> 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<String> validPhasicTypes = (conType != 0) ? LINE_VOLTAGE_TYPES : PHASE_VOLTAGE_TYPES;
|
||||||
|
List<DataPlt> 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;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@@ -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<String, String> PHASE_MAPPING = new HashMap<String, String>() {{
|
||||||
|
put("AB", "A");
|
||||||
|
put("BC", "B");
|
||||||
|
put("CA", "C");
|
||||||
|
put("M", "T");
|
||||||
|
}};
|
||||||
|
@Resource
|
||||||
|
private DataVMapper dataVMapper;
|
||||||
|
@Resource
|
||||||
|
private RedisUtil redisUtil;
|
||||||
|
private static final Set<String> LINE_VOLTAGE_TYPES =
|
||||||
|
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("AB", "BC", "CA", "T")));
|
||||||
|
private static final Set<String> PHASE_VOLTAGE_TYPES =
|
||||||
|
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("A", "B", "C", "T")));
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<DataVDto> getRawData(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataVDto> result = new ArrayList<>();
|
||||||
|
List<DataV> 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<DataV> getMinuteDataV(LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataV> result = new ArrayList<>();
|
||||||
|
List<DataV> data = new ArrayList<>();
|
||||||
|
//获取监测点、接线方式数据
|
||||||
|
Type type = new TypeToken<Map<String, Integer>>(){}.getType();
|
||||||
|
Map<String, Integer> 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<String, List<DataV>> lineMap = result.stream().collect(Collectors.groupingBy(DataV::getLineId));
|
||||||
|
lineMap.forEach((k,v)->{
|
||||||
|
if (Objects.isNull(map.get(k))) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
//这边需要特殊处理下,将线电压数据赋值
|
||||||
|
Map<String, DataV> 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<String> validPhasicTypes = (conType != 0) ? LINE_VOLTAGE_TYPES : PHASE_VOLTAGE_TYPES;
|
||||||
|
List<DataV> 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<DataV> result, InfluxQueryWrapper influxQueryWrapper, LineCountEvaluateParam lineParam) {
|
||||||
|
List<DataV> dataList;
|
||||||
|
List<DataV> 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<String, List<DataV>> lineMap = dataList.stream().collect(Collectors.groupingBy(DataV::getLineId));
|
||||||
|
//有异常数据
|
||||||
|
Map<String, List<String>> timeMap = lineParam.getAbnormalTime();
|
||||||
|
if (CollectionUtil.isNotEmpty(timeMap)) {
|
||||||
|
lineMap.forEach((k, v) -> {
|
||||||
|
List<String> timeList = timeMap.get(k);
|
||||||
|
//有异常数据,当前监测点自身的异常数据
|
||||||
|
if (CollectionUtil.isNotEmpty(timeList)) {
|
||||||
|
List<DataV> 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;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user