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

算法清洗，调整空指针问题

Browse Source
This commit is contained in:
wr
2025-07-17 15:34:23 +08:00
parent fdb54dc678
commit a3e180c070
3 changed files with 94 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,7 +26,6 @@ import com.njcn.device.pq.pojo.vo.LineDetailVO;
import com.njcn.oss.constant.OssPath;
import com.njcn.oss.utils.FileStorageUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,7 +57,7 @@ public class DataCleanServiceImpl implements IDataCleanService {
private static final Logger logger = LoggerFactory.getLogger(DataCleanServiceImpl.class);
private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault());
@Value("${line.num}")
private Integer NUM = 100;
@@ -1538,21 +1537,23 @@ public class DataCleanServiceImpl implements IDataCleanService {
// 50次谐波
for (int i = 1; i <= 50; i++) {
// 通过反射获取dto.getV(i)的值
double vValue = getVValue(dto, i);
if (vValue < pqReasonableRangeDto.getMinValue() || vValue > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(
dto.getLineId(),
dto.getMinTime(),
dto.getValueType(),
dto.getPhasicType(),
pqReasonableRangeDto.getIndexCode(),
(i - 0.5) + pqReasonableRangeDto.getIndexName(),
pqReasonableRangeDto.getInfluxdbTableName(),
vValue,
pqReasonableRangeDto.getMinValue(),
pqReasonableRangeDto.getMaxValue()
);
list.add(pqDataVerify);
Double vValue = getVValue(dto, i);
if(ObjectUtil.isNotNull(vValue)){
if (vValue < pqReasonableRangeDto.getMinValue() || vValue > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(
dto.getLineId(),
dto.getMinTime(),
dto.getValueType(),
dto.getPhasicType(),
pqReasonableRangeDto.getIndexCode(),
(i - 0.5) + pqReasonableRangeDto.getIndexName(),
pqReasonableRangeDto.getInfluxdbTableName(),
vValue,
pqReasonableRangeDto.getMinValue(),
pqReasonableRangeDto.getMaxValue()
);
list.add(pqDataVerify);
}
}
}
}
@@ -1570,21 +1571,23 @@ public class DataCleanServiceImpl implements IDataCleanService {
// 50次谐波
for (int i = 2; i <= 50; i++) {
// 通过反射获取dto.getV(i)的值
double vValue = getVValue(dto, i);
if (vValue < pqReasonableRangeDto.getMinValue() || vValue > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(
dto.getLineId(),
dto.getMinTime(),
dto.getValueType(),
dto.getPhasicType(),
pqReasonableRangeDto.getIndexCode(),
i + pqReasonableRangeDto.getIndexName(),
pqReasonableRangeDto.getInfluxdbTableName(),
vValue,
pqReasonableRangeDto.getMinValue(),
pqReasonableRangeDto.getMaxValue()
);
list.add(pqDataVerify);
Double vValue = getVValue(dto, i);
if(ObjectUtil.isNotNull(vValue)){
if (vValue < pqReasonableRangeDto.getMinValue() || vValue > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(
dto.getLineId(),
dto.getMinTime(),
dto.getValueType(),
dto.getPhasicType(),
pqReasonableRangeDto.getIndexCode(),
i + pqReasonableRangeDto.getIndexName(),
pqReasonableRangeDto.getInfluxdbTableName(),
vValue,
pqReasonableRangeDto.getMinValue(),
pqReasonableRangeDto.getMaxValue()
);
list.add(pqDataVerify);
}
}
}
}
@@ -1627,27 +1630,29 @@ public class DataCleanServiceImpl implements IDataCleanService {
// 50次谐波
for (int i = 1; i <= 50; i++) {
// 通过反射获取dto.getV(i)的值
double vValue = getVValue(dto, i);
if (vValue < -180) {
vValue = vValue + 360;
}
if (vValue > 180) {
vValue = vValue - 360;
}
if (vValue < pqReasonableRangeDto.getMinValue() || vValue > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(
dto.getLineId(),
dto.getMinTime(),
dto.getValueType(),
dto.getPhasicType(),
pqReasonableRangeDto.getIndexCode(),
i + pqReasonableRangeDto.getIndexName(),
pqReasonableRangeDto.getInfluxdbTableName(),
vValue,
pqReasonableRangeDto.getMinValue(),
pqReasonableRangeDto.getMaxValue()
);
list.add(pqDataVerify);
Double vValue = getVValue(dto, i);
if(ObjectUtil.isNotNull(vValue)){
if (vValue < -180) {
vValue = vValue + 360;
}
if (vValue > 180) {
vValue = vValue - 360;
}
if (vValue < pqReasonableRangeDto.getMinValue() || vValue > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(
dto.getLineId(),
dto.getMinTime(),
dto.getValueType(),
dto.getPhasicType(),
pqReasonableRangeDto.getIndexCode(),
i + pqReasonableRangeDto.getIndexName(),
pqReasonableRangeDto.getInfluxdbTableName(),
vValue,
pqReasonableRangeDto.getMinValue(),
pqReasonableRangeDto.getMaxValue()
);
list.add(pqDataVerify);
}
}
}
}
@@ -1727,12 +1732,16 @@ public class DataCleanServiceImpl implements IDataCleanService {
return pqDataVerify;
}
private double getVValue(DataHarmDto dto, int index) {
private Double getVValue(DataHarmDto dto, int index) {
try {
Method method = DataHarmDto.class.getMethod("getV" + index);
return (double) method.invoke(dto);
Object invoke = method.invoke(dto);
if(invoke==null){
return null;
}
return (Double) invoke;
} catch (Exception e) {
throw new RuntimeException("Failed to get V value for index: " + index, e);
throw new RuntimeException(dto+"Failed to get V value for index: " + index, e);
}
}

64
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/controller/MigrationInfluxDBController.java
View File

@@ -2,19 +2,14 @@ package com.njcn.migration.read.controller;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.util.CharsetUtil;
import com.njcn.param.LineCountEvaluateParam;
import com.njcn.migration.read.service.MigrationService;
import io.swagger.annotations.*;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.net.URLEncoder;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
@@ -71,42 +66,31 @@ public class MigrationInfluxDBController {
}
return true;
}
@GetMapping(value = "/importExcel",produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
@ApiOperation(value ="获取最新数据信息文件", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
public void importDistributionAreaExcel(HttpServletResponse response) throws IOException {
migrationService.initializeExcel();
FileInputStream fileInputStream = new FileInputStream("/usr/local/jar/sj.xlsx");
response.setHeader("Content-Disposition", "attachment;filename="+ URLEncoder.encode("最新数据信息.xlsx", CharsetUtil.UTF_8));
// 输出文件内容
OutputStream responseOutputStream = response.getOutputStream();
byte[] buffer = new byte[4096];
int bytesRead;
while ((bytesRead = fileInputStream.read(buffer)) != -1) {
responseOutputStream.write(buffer, 0, bytesRead);
@GetMapping(value = "/importExcel")
@ApiOperation(value ="是否更新数据（1是 0否表示用于定时任务）")
public void importDistributionAreaExcel(@RequestParam("type") Integer type) {
if(type==0){
migrationService.initializeExcel();
}else{
System.out.println("--------------------------------手动1influxdb同步------------------------------------");
// 获取当前时间
LocalDateTime now = LocalDateTime.now();
// 减去2个小时
LocalDateTime oneHourAgo = now.minusHours(2);
// 将分钟和秒设置为0
LocalDateTime result = oneHourAgo.truncatedTo(ChronoUnit.HOURS);
// 加上59分钟59秒
LocalDateTime modifiedResult = result.plusMinutes(59).plusSeconds(59);
LineCountEvaluateParam param = new LineCountEvaluateParam();
param.setIsManual(false);
param.setStartTime(result.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
param.setEndTime(modifiedResult.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
migrationService.hourseLineDataBacthSysc(param);
migrationService.hourseDevDataBacthSysc(param);
}
fileInputStream.close();
responseOutputStream.close();
}
@GetMapping(value = "/influxdbCs",produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
@ApiOperation(value ="influxdb数据同步->天数按小时进行分组同步测试", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
public void influxdbCs() {
System.out.println("--------------------------------influxdb同步------------------------------------");
// 获取当前时间
LocalDateTime now = LocalDateTime.now();
// 减去2个小时
LocalDateTime oneHourAgo = now.minusHours(2);
// 将分钟和秒设置为0
LocalDateTime result = oneHourAgo.truncatedTo(ChronoUnit.HOURS);
// 加上59分钟59秒
LocalDateTime modifiedResult = result.plusMinutes(59).plusSeconds(59);
LineCountEvaluateParam param = new LineCountEvaluateParam();
param.setIsManual(false);
param.setStartTime(result.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
param.setEndTime(modifiedResult.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
migrationService.hourseLineDataBacthSysc(param);
migrationService.hourseDevDataBacthSysc(param);
}
}

7
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/service/impl/MigrationServiceImpl.java
View File

@@ -94,12 +94,13 @@ public class MigrationServiceImpl implements MigrationService {
migrationInsertFeignClient.insertDataInharmI(dataInharmI.listDataInharmI(evaluateParam));
migrationInsertFeignClient.insertDataInharmV(dataInharmV.listDataInharmV(evaluateParam));
migrationInsertFeignClient.insertDataPlt(dataPlt.listDataPlt(evaluateParam));
migrationInsertFeignClient.batchInsertion(eventDetail.getRawData(evaluateParam));
// migrationInsertFeignClient.batchInsertion(eventDetail.getRawData(evaluateParam));
if (!param.getIsManual() && StrUtil.isNotBlank(format)) {
TimeUtil.putLineTime(lineId, format);
}
});
System.gc();
}
@Override
@@ -120,7 +121,7 @@ public class MigrationServiceImpl implements MigrationService {
String format = null;
if (!param.getIsManual()) {
if (StrUtil.isNotBlank(time)) {
param.setStartTime(time);
evaluateParam.setStartTime(time);
}
}
List<PqsCommunicate> pqsCommunicates = pqsCommunicate.listPqsCommunicate(evaluateParam);
@@ -134,9 +135,11 @@ public class MigrationServiceImpl implements MigrationService {
TimeUtil.putDevTime(lineId, format);
}
});
System.gc();
}
@Override
@Async("asyncInfluxDBExecutor")
public void initializeExcel() {
File file = new File("/usr/local/jar/sj.xlsx");
Map<String, String> map = TimeUtil.getLineMap();