Middleware/cn-data-platform
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.增加暂态数据清洗

Browse Source
This commit is contained in:
wr
2025-11-25 17:02:31 +08:00
parent 89df5f3039
commit 438f618d8a
6 changed files with 103 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java
View File

@@ -333,6 +333,8 @@ public class DataCleanServiceImpl implements IDataCleanService {
List<DataFlucDto> data8 = dataFlucFeignClient.getRawData(lineParam).getData();
//dataFlicker
List<DataFlickerDto> data9 = dataFlickerFeignClient.getRawData(lineParam).getData();
//获取监测点的暂态事件
List<RmpEventDetailDTO> eventList = rmpEventDetailFeignClient.getRawData(lineParam).getData();
if (CollUtil.isNotEmpty(data1)) {
logger.info("{}数据清洗dataV集合大小为>>>>>>>>>>>>{}",lineParam.getStartTime(), MemorySizeUtil.getObjectSize(data1));
@@ -514,6 +516,26 @@ public class DataCleanServiceImpl implements IDataCleanService {
}
}
if (CollUtil.isNotEmpty(eventList)) {
logger.info("{}数据清洗eventDetail集合大小为>>>>>>>>>>>>{}",lineParam.getStartTime(), MemorySizeUtil.getObjectSize(eventList));
List<PqDataVerify> result = new ArrayList<>();
List<PqReasonableRangeDto> list = map.get(DataCleanEnum.RMpEventDetail.getCode());
Map<String, PqReasonableRangeDto> pqReasonableRangeDtoMap = list.stream().collect(Collectors.toMap(PqReasonableRangeDto::getIndexCode, Function.identity()));
eventList.forEach(data->{
List<PqDataVerify> pqDataVerifies = judgeDataEvent(pqReasonableRangeDtoMap,data);
result.addAll(pqDataVerifies);
});
if (CollUtil.isNotEmpty(result)) {
Map<String,Object> dtoMap = handleDataEvent(result,bak,item.getTimeInterval());
Set<String> timeSet = new HashSet<>(result.stream().map(dt -> dt.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_TIME_PATTERN))).collect(Collectors.toSet()));
dtoMap.put("errorTimes",timeSet.size() * item.getTimeInterval());
allTimeSet.addAll(timeSet);
if (CollUtil.isNotEmpty(dtoMap)) {
resultData.add(dtoMap);
}
}
}
if (CollUtil.isNotEmpty(allTimeSet)) {
Map<String,Object> dtoMap = new HashMap<>();
dtoMap.put("lineErrorTimes",allTimeSet.size() * item.getTimeInterval());
@@ -716,6 +738,24 @@ public class DataCleanServiceImpl implements IDataCleanService {
return map;
}
public Map<String,Object> handleDataEvent(List<PqDataVerify> result, PqDataVerifyBak bak,Integer timeInterval) {
Map<String,Object> map = new HashMap<>();
Map<String,List<PqDataVerify>> codeMap = result.stream().collect(Collectors.groupingBy(PqDataVerify::getIndexCode));
//暂降事件
List<DataCleanJsonDTO.DataHarmCleanJsonDTO> list = handleData(codeMap.get(DataCleanEnum.VoltageDip.getCode()),timeInterval);
if (CollUtil.isNotEmpty(list)) {
bak.setEvent(1);
map.put(DataCleanEnum.RMpEventDetail.getCode() + "-" + DataCleanEnum.VoltageDip.getCode(),list);
}
//暂降事件
List<DataCleanJsonDTO.DataHarmCleanJsonDTO> list2 = handleData(codeMap.get(DataCleanEnum.VoltageRise.getCode()),timeInterval);
if (CollUtil.isNotEmpty(list2)) {
bak.setEvent(1);
map.put(DataCleanEnum.RMpEventDetail.getCode() + "-" + DataCleanEnum.VoltageRise.getCode(),list2);
}
return map;
}
public List<DataCleanJsonDTO.DataHarmCleanJsonDTO> handleData(List<PqDataVerify> list, Integer timeInterval) {
List<DataCleanJsonDTO.DataHarmCleanJsonDTO> dataList = new ArrayList<>();
if (CollUtil.isNotEmpty(list)) {
@@ -754,6 +794,8 @@ public class DataCleanServiceImpl implements IDataCleanService {
});
dto.setErrorCounts(allCounts.get());
dto.setErrorTimes(timeSet.size()*timeInterval);
dto.setMinValue(list.get(0).getMinValue());
dto.setMaxValue(list.get(0).getMaxValue());
dto.setList(list1);
});
dataList.add(dto);
@@ -1144,7 +1186,7 @@ public class DataCleanServiceImpl implements IDataCleanService {
Map<String, List<PqReasonableRangeDto>> pqReasonableRangeDtoMap = new HashMap<>();
DataCleanParam param = new DataCleanParam();
param.setSystemType(DataCleanEnum.Pqs.getCode());
param.setDataSource(DataCleanEnum.InfluxDB.getCode());
// param.setDataSource(DataCleanEnum.InfluxDB.getCode());
List<PqReasonableRangeDto> list = pqReasonableRangeFeignClient.getData(param).getData();
if (CollUtil.isNotEmpty(list)) {
pqReasonableRangeDtoMap = list.stream().collect(Collectors.groupingBy(PqReasonableRangeDto::getInfluxdbTableName));
@@ -1769,11 +1811,51 @@ public class DataCleanServiceImpl implements IDataCleanService {
return list;
}
/**
* 获取数据的标准范围，进行数据比对
*/
public List<PqDataVerify> judgeDataEvent(Map<String, PqReasonableRangeDto> map, RmpEventDetailDTO dto) {
List<PqDataVerify> list = new ArrayList<>();
PqReasonableRangeDto pqReasonableRangeDto = map.get(DataCleanEnum.VoltageDip.getCode());
if(ObjectUtil.isNotNull(dto.getFeatureAmplitude())){
if (dto.getFeatureAmplitude() < pqReasonableRangeDto.getMinValue() || dto.getFeatureAmplitude() > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(dto.getMeasurementPointId()
,dto.getStartTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN))
,null
,"/"
,pqReasonableRangeDto.getIndexCode()
,pqReasonableRangeDto.getIndexName()
,pqReasonableRangeDto.getInfluxdbTableName()
,dto.getFeatureAmplitude()
,pqReasonableRangeDto.getMinValue()
,pqReasonableRangeDto.getMaxValue());
list.add(pqDataVerify);
}
}
PqReasonableRangeDto pqReasonableRangeDto2 = map.get(DataCleanEnum.VoltageRise.getCode());
if(ObjectUtil.isNotNull(dto.getFeatureAmplitude())){
if (dto.getFeatureAmplitude() < pqReasonableRangeDto2.getMinValue() || dto.getFeatureAmplitude() > pqReasonableRangeDto2.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(dto.getMeasurementPointId()
,dto.getStartTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN))
,null
,"/"
,pqReasonableRangeDto2.getIndexCode()
,pqReasonableRangeDto2.getIndexName()
,pqReasonableRangeDto2.getInfluxdbTableName()
,dto.getFeatureAmplitude()
,pqReasonableRangeDto2.getMinValue()
,pqReasonableRangeDto2.getMaxValue());
list.add(pqDataVerify);
}
}
return list;
}
public PqDataVerify getPqDataVerify(String lineId,String time,String valueType, String phasicType, String indexCode, String indexName, String indexTable, Double data, Double minValue, Double maxValue) {
PqDataVerify pqDataVerify = new PqDataVerify();
pqDataVerify.setLineId(lineId);
pqDataVerify.setTime(TimeUtils.StringToLocalDateTime(time));
pqDataVerify.setTime(time.contains(".")?TimeUtils.StringToLocalDateTimeSSS(time):TimeUtils.StringToLocalDateTime(time));
pqDataVerify.setValueType(Objects.isNull(valueType)?"avg":valueType);
pqDataVerify.setPhasicType(phasicType);
pqDataVerify.setIndexCode(indexCode);

5
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/dto/DataCleanJsonDTO.java
View File

@@ -25,6 +25,11 @@ public class DataCleanJsonDTO {
@ApiModelProperty("异常数据量")
private Integer errorCounts;
@ApiModelProperty("指标上限")
private Double minValue;
@ApiModelProperty("指标下限")
private Double maxValue;
}
@Data

4
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/enums/DataCleanEnum.java
View File

@@ -81,8 +81,8 @@ public enum DataCleanEnum {
V_Data("v_1","相（线）电压基波有效值"),
//r_mp_event_detail
VoltageDip("Voltage_Dip","相（线）电压基波有效值"),
VoltageRise("Voltage_Rise","相（线）电压基波有效值");
VoltageDip("Voltage_Dip","电压暂降特征幅值"),
VoltageRise("Voltage_Rise","电压暂升特征幅值");
private String code;

2
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/dto/AbnormalData.java
View File

@@ -65,6 +65,8 @@ public class AbnormalData {
@JSONField(ordinal = 4)
private String value;
@JSONField(ordinal = 5)
private float overLimitValue;
}
}

5
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/po/PqDataVerifyBak.java
View File

@@ -173,6 +173,11 @@ public class PqDataVerifyBak implements Serializable {
@TableField(value = "pst")
private Integer pst = 0;
@TableField(value = "event")
private Integer event = 0;
/**
* 文件路径
*/

5
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/util/TimeUtils.java
View File

@@ -16,6 +16,7 @@ import java.time.format.DateTimeFormatter;
public class TimeUtils {
private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private static final DateTimeFormatter DATE_TIME_FORMATTER_SSS = DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN);
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd");
/**
@@ -37,6 +38,10 @@ public class TimeUtils {
return LocalDateTime.parse(time, DATE_TIME_FORMATTER);
}
public static LocalDateTime StringToLocalDateTimeSSS(String time) {
return LocalDateTime.parse(time, DATE_TIME_FORMATTER_SSS);
}
/**
* LocalDataTime类型的yyyy-MM-dd HH:mm:ss转成String yyyy-MM-dd HH:mm:ss
* @param time