Microservice/govern
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(harmonic): 新增移动端线路详情功能并优化波形数据处理

Browse Source 
- 添加 AppLineDetailVo 数据传输对象，支持移动端线路详情展示
- 增加 report 服务中的 buildHarmonic 相关方法重构，支持移动端线路详情查询
- 优化波形数据处理逻辑，新增波形数据抽点和裁剪功能，减少移动端数据传输量
- 修改 CommonStatisticalQueryParam 参数类，增加数据模型字段和电度事件类型支持
- 调整统计查询相关接口，支持全量和增量查询模式
- 移除 CredentialReqDTO 类，清理相关依赖
- 优化 CsAppReportServiceImpl 中的越限描述构建逻辑，使用时间转换工具
- 更新数据查询相关 Mapper XML 文件，调整数据过滤条件
- 修改设备用户服务实现，完善当前工程数据显示逻辑
- 优化 CsEquipmentDeliveryServiceImpl 中的数据集添加逻辑，支持电度数据类型
- 重构 CsEventController 和相关服务类，支持移动端波形数据分析
- 添加 Nacos 配置参数控制波形数据抽点和间隔区域处理行为
This commit is contained in:
xy
2026-06-03 10:22:18 +08:00
parent a6f424025a
commit cf691db0a6
39 changed files with 1393 additions and 900 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/controller/CsEventController.java
View File

@@ -15,7 +15,6 @@ import com.njcn.csharmonic.pojo.param.EventStatisticParam;
import com.njcn.csharmonic.pojo.po.CsEventPO;
import com.njcn.csharmonic.pojo.vo.CsEventVO;
import com.njcn.csharmonic.pojo.vo.EventDetailVO;
import com.njcn.csharmonic.pojo.vo.EventStatisticsVo;
import com.njcn.csharmonic.service.CsEventPOService;
import com.njcn.event.file.pojo.dto.WaveDataDTO;
import com.njcn.web.controller.BaseController;
@@ -78,10 +77,13 @@ public class CsEventController extends BaseController {
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@GetMapping("/analyseWave")
@ApiOperation("暂态事件波形分析")
@ApiImplicitParam(name = "eventId", value = "暂态事件索引", required = true)
public HttpResult<WaveDataDTO> analyseWave(String eventId) {
@ApiImplicitParams({
@ApiImplicitParam(name = "eventId", value = "暂态事件索引"),
@ApiImplicitParam(name = "isApp", value = "是否是移动端查询")
})
public HttpResult<WaveDataDTO> analyseWave(@RequestParam(value = "eventId") String eventId,@RequestParam(value = "isApp", required = false) Boolean isApp) {
String methodDescribe = getMethodDescribe("analyseWave");
WaveDataDTO wave = csEventPOService.analyseWave(eventId,2);
WaveDataDTO wave = csEventPOService.analyseWave(eventId,2,isApp);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, wave, methodDescribe);
}

11
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/controller/DataController.java
View File

@@ -9,6 +9,7 @@ import com.njcn.csdevice.pojo.vo.DataGroupEventVO;
import com.njcn.csdevice.pojo.vo.EachModuleVO;
import com.njcn.csdevice.pojo.vo.RecordVo;
import com.njcn.csharmonic.param.DataParam;
import com.njcn.csharmonic.pojo.vo.AppLineDetailVo;
import com.njcn.csharmonic.pojo.vo.RealTimeDataVo;
import com.njcn.csharmonic.service.IDataService;
import com.njcn.web.controller.BaseController;
@@ -83,4 +84,14 @@ public class DataController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, list, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/getLineDataByEngineer")
@ApiOperation("App-》工程下所有监测点数据")
@ApiImplicitParam(name = "id", value = "工程id", required = true)
public HttpResult<List<AppLineDetailVo>> getLineDataByEngineer(@RequestParam("id") String id) {
String methodDescribe = getMethodDescribe("getLineDataByEngineer");
List<AppLineDetailVo> list = dataService.getLineDataByEngineer(id);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, list, methodDescribe);
}
}

8
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/handler/MqttMessageHandler.java
View File

@@ -131,7 +131,7 @@ public class MqttMessageHandler {
List<ThdDataVO> tempList = new ArrayList<>();
//1.查询拓扑图配置的指标:拓扑图扑图配置：7677f94c749dedaff30f911949cbd724
List<EleEpdPqd> data = csStatisticalSetFeignClient.queryStatisticalSelect(typeId).getData();
List<EleEpdPqd> data = csStatisticalSetFeignClient.queryStatisticalSelect(Collections.singletonList(typeId)).getData();
data.forEach(temp -> {
if (Objects.nonNull(temp.getHarmStart()) && Objects.nonNull(temp.getHarmEnd())) {
FrequencyStatisticalQueryParam frequencyStatisticalQueryParam = new FrequencyStatisticalQueryParam();
@@ -180,7 +180,7 @@ public class MqttMessageHandler {
if (item.getTarget() != null && !item.getTarget().isEmpty()) {
List<ThdDataVO> l1 = new ArrayList<>();
//根据指标查询
List<EleEpdPqd> data1 = csStatisticalSetFeignClient.queryStatisticalSelect(item.getTarget()).getData();
List<EleEpdPqd> data1 = csStatisticalSetFeignClient.queryStatisticalSelect(Collections.singletonList(item.getTarget())).getData();
data1.forEach(temp -> {
CommonStatisticalQueryParam commonStatisticalQueryParam = new CommonStatisticalQueryParam();
commonStatisticalQueryParam.setDevId(devId);
@@ -219,7 +219,7 @@ public class MqttMessageHandler {
//根据指标查询
List<ThdDataVO> l1 = new ArrayList<>();
//根据指标查询
List<EleEpdPqd> data1 = csStatisticalSetFeignClient.queryStatisticalSelect(item.getTarget()).getData();
List<EleEpdPqd> data1 = csStatisticalSetFeignClient.queryStatisticalSelect(Collections.singletonList(item.getTarget())).getData();
data1.forEach(temp -> {
CommonStatisticalQueryParam commonStatisticalQueryParam = new CommonStatisticalQueryParam();
commonStatisticalQueryParam.setDevId(devId);
@@ -260,7 +260,7 @@ public class MqttMessageHandler {
//根据指标查询
List<ThdDataVO> l1 = new ArrayList<>();
//根据指标查询
List<EleEpdPqd> data1 = csStatisticalSetFeignClient.queryStatisticalSelect(item.getTarget()).getData();
List<EleEpdPqd> data1 = csStatisticalSetFeignClient.queryStatisticalSelect(Collections.singletonList(item.getTarget())).getData();
data1.forEach(temp -> {
CommonStatisticalQueryParam commonStatisticalQueryParam = new CommonStatisticalQueryParam();
commonStatisticalQueryParam.setDevId(devId);

5
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/CsEventPOService.java
View File

@@ -12,10 +12,7 @@ import com.njcn.csharmonic.pojo.po.CsEventPO;
import com.njcn.csharmonic.pojo.vo.CsEventVO;
import com.njcn.csharmonic.pojo.vo.CsWarnDescVO;
import com.njcn.csharmonic.pojo.vo.EventDetailVO;
import com.njcn.csharmonic.pojo.vo.EventStatisticsVo;
import com.njcn.event.file.pojo.dto.WaveDataDTO;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import javax.servlet.http.HttpServletResponse;
import java.time.LocalDateTime;
@@ -40,7 +37,7 @@ public interface CsEventPOService extends IService<CsEventPO>{
* @date 2023/9/20 14:23
* @return WaveDataDTO
*/
WaveDataDTO analyseWave(String eventId, int i);
WaveDataDTO analyseWave(String eventId, int i, Boolean isApp);
/***
* 获取事件波形图信息

3
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/IDataService.java
View File

@@ -4,6 +4,7 @@ import com.njcn.csdevice.pojo.vo.DataGroupEventVO;
import com.njcn.csdevice.pojo.vo.EachModuleVO;
import com.njcn.csdevice.pojo.vo.RecordVo;
import com.njcn.csharmonic.param.DataParam;
import com.njcn.csharmonic.pojo.vo.AppLineDetailVo;
import com.njcn.csharmonic.pojo.vo.RealTimeDataVo;
import java.util.List;
@@ -38,4 +39,6 @@ public interface IDataService {
List<EachModuleVO> getModuleState(String id);
List<AppLineDetailVo> getLineDataByEngineer(String id);
}

251
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/CsEventPOServiceImpl.java
View File

@@ -74,6 +74,7 @@ import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
@@ -126,6 +127,14 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
private final EventCauseFeignClient eventCauseFeignClient;
private final MsgSendFeignClient msgSendFeignClient;
/** 波形数据抽点窗口大小，每N个点为一个窗口进行MinMax抽点，可通过Nacos配置wave.sample.interval调整 */
@Value("${wave.sample.interval:0}")
private int waveSampleInterval;
/** 两段数据间隔区域补null点数量，可通过Nacos配置wave.gap.null.points调整 */
@Value("${wave.gap.null.points:5}")
private int waveGapNullPoints;
@Override
public List<EventDetailVO> queryEventList(CsEventUserQueryParam csEventUserQueryParam) {
@@ -638,7 +647,7 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
* iType == 3 高级算法原始波形（大于32）
*/
@Override
public WaveDataDTO analyseWave(String eventId, int iType) {
public WaveDataDTO analyseWave(String eventId, int iType, Boolean isApp) {
WaveDataDTO waveDataDTO;
//获取暂降事件
CsEventPO eventDetail = this.baseMapper.selectById(eventId);
@@ -674,6 +683,10 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
throw new BusinessException(WaveFileResponseEnum.WAVE_DATA_INVALID);
}
waveDataDTO = waveFileComponent.getValidData(waveDataDTO);
if (!Objects.isNull(isApp) && isApp) {
//根据持续时间裁剪波形数据，只保留开始段和结束段的关键数据
filterWaveDataByPersistTime(waveDataDTO, eventDetail.getPersistTime());
}
List<CsLinePO> csLinePOList = csLineFeignClient.queryLineById(Stream.of(eventDetail.getLineId()).collect(Collectors.toList())).getData();
if (CollectionUtil.isEmpty(csLinePOList)) {
throw new BusinessException(AlgorithmResponseEnum.LINE_DATA_MISS);
@@ -690,7 +703,7 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
CsEventPO eventDetail = this.baseMapper.selectById(eventId);
if (StrUtil.isBlank(eventDetail.getInstantPics())) {
//获取波形数据。然后绘图
WaveDataDTO waveDataDTO = this.analyseWave(eventId, iType);
WaveDataDTO waveDataDTO = this.analyseWave(eventId, iType ,true);
//数据筛选,如果是双路电压的话，会存在2个波形数据
List<WaveDataDetail> waveDataDetails = WaveUtil.filterWaveData(waveDataDTO);
//单通道处理
@@ -718,4 +731,238 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
return csEventVO;
}
private void filterWaveDataByPersistTime(WaveDataDTO waveDataDTO, Double persistTime) {
if (persistTime == null || persistTime <= 0) {
return;
}
double persistTimeMs = persistTime * 1000;
// 开始段：波形起始前后，-40ms到100ms
double startSegBegin = -100;
double startSegEnd = 100;
// 结束段：恢复点前后各40ms
double endSegBegin = persistTimeMs - 40;
double endSegEnd = persistTimeMs + 40;
// 关键时间点（抽点时必须保留，不可被抽掉）
Set<Double> keyTimes = new HashSet<>();
keyTimes.add(startSegBegin); // 开头-100
keyTimes.add(0.0); // 波形开始时间点0
keyTimes.add(persistTimeMs); // 持续时间点
keyTimes.add(endSegBegin); // 持续时间点往前-40
keyTimes.add(endSegEnd); // 持续时间点往后+40
List<List<Float>> originalWaveData = waveDataDTO.getListWaveData();
List<List<Float>> originalRmsData = waveDataDTO.getListRmsData();
if (originalWaveData.isEmpty()) {
return;
}
int waveColumnCount = originalWaveData.get(0).size();
int rmsColumnCount = originalRmsData.get(0).size();
boolean isMerged = endSegBegin <= startSegEnd;
List<List<Float>> filteredWaveData;
List<List<Float>> filteredRmsData;
if (isMerged) {
// 两段重叠或相邻，合并为一段连续数据
double mergedEnd = Math.max(startSegEnd, endSegEnd);
List<List<Float>> segWaveData = filterByTimeRange(originalWaveData, startSegBegin, mergedEnd);
List<List<Float>> segRmsData = filterByTimeRange(originalRmsData, startSegBegin, mergedEnd);
// 合并为1段：MinMax抽点 + 保留关键节点
filteredWaveData = downsampleMinMaxWithKeyPoints(segWaveData, keyTimes, waveSampleInterval);
filteredRmsData = downsampleMinMaxWithKeyPoints(segRmsData, keyTimes, waveSampleInterval);
} else {
// 两段分离，分别MinMax抽点 + 保留关键节点
List<List<Float>> seg1Wave = filterByTimeRange(originalWaveData, startSegBegin, startSegEnd);
List<List<Float>> seg2Wave = filterByTimeRange(originalWaveData, endSegBegin, endSegEnd);
List<List<Float>> seg1Rms = filterByTimeRange(originalRmsData, startSegBegin, startSegEnd);
List<List<Float>> seg2Rms = filterByTimeRange(originalRmsData, endSegBegin, endSegEnd);
List<List<Float>> ds1Wave = downsampleMinMaxWithKeyPoints(seg1Wave, keyTimes, waveSampleInterval);
List<List<Float>> ds2Wave = downsampleMinMaxWithKeyPoints(seg2Wave, keyTimes, waveSampleInterval);
List<List<Float>> ds1Rms = downsampleMinMaxWithKeyPoints(seg1Rms, keyTimes, waveSampleInterval);
List<List<Float>> ds2Rms = downsampleMinMaxWithKeyPoints(seg2Rms, keyTimes, waveSampleInterval);
// 两段之间补固定数量的null点，时间均匀分布
double gapStart = getTimeValue(ds1Wave.get(ds1Wave.size() - 1));
double gapEnd = getTimeValue(ds2Wave.get(0));
List<List<Float>> gapWaveData = createNullGapData(gapStart, gapEnd, waveGapNullPoints, waveColumnCount);
List<List<Float>> gapRmsData = createNullGapData(gapStart, gapEnd, waveGapNullPoints, rmsColumnCount);
// 合并：段1 + null间隔 + 段2
filteredWaveData = new ArrayList<>();
filteredWaveData.addAll(ds1Wave);
filteredWaveData.addAll(gapWaveData);
filteredWaveData.addAll(ds2Wave);
filteredRmsData = new ArrayList<>();
filteredRmsData.addAll(ds1Rms);
filteredRmsData.addAll(gapRmsData);
filteredRmsData.addAll(ds2Rms);
}
waveDataDTO.setListWaveData(filteredWaveData);
waveDataDTO.setListRmsData(filteredRmsData);
}
/**
* MinMax抽点：每N个点为一个窗口，保留窗口内峰值点、谷值点、首点及关键时间点
* 峰值=参考列最大值的数据行，谷值=参考列最小值的数据行
* 这样正弦波的峰谷特征得以保留，波形形状不会被破坏
*
* @param dataList 已按时间范围筛选的数据
* @param keyTimes 关键时间点集合(ms)
* @param sampleInterval 窗口大小(N)
*/
private List<List<Float>> downsampleMinMaxWithKeyPoints(List<List<Float>> dataList, Set<Double> keyTimes, int sampleInterval) {
if (dataList.isEmpty()) {
return dataList;
}
// 找到关键时间点对应的数据索引
Set<Integer> keyIndices = new HashSet<>();
for (Double keyTime : keyTimes) {
int closestIdx = findClosestIndex(dataList, keyTime);
if (closestIdx >= 0) {
keyIndices.add(closestIdx);
}
}
List<List<Float>> result = new ArrayList<>();
int size = dataList.size();
// 使用第1列（第一个值列，通常是A相电压）作为峰谷检测参考列
int refCol = 1;
for (int windowStart = 0; windowStart < size; windowStart += sampleInterval) {
int windowEnd = Math.min(windowStart + sampleInterval, size);
// 在窗口内找参考列的最大值点(峰值)和最小值点(谷值)
int maxIdx = windowStart;
int minIdx = windowStart;
float maxVal = getFloatValue(dataList.get(windowStart), refCol);
float minVal = getFloatValue(dataList.get(windowStart), refCol);
for (int i = windowStart + 1; i < windowEnd; i++) {
float val = getFloatValue(dataList.get(i), refCol);
if (val > maxVal) {
maxVal = val;
maxIdx = i;
}
if (val < minVal) {
minVal = val;
minIdx = i;
}
}
// 保留：窗口首点 + 峰值 + 谷值 + 窗口内的关键时间点
Set<Integer> keepSet = new HashSet<>();
keepSet.add(windowStart);
keepSet.add(maxIdx);
keepSet.add(minIdx);
for (int i = windowStart; i < windowEnd; i++) {
if (keyIndices.contains(i)) {
keepSet.add(i);
}
}
// 按原始顺序输出（索引顺序即时间顺序）
List<Integer> sortedKeep = new ArrayList<>(keepSet);
Collections.sort(sortedKeep);
for (int idx : sortedKeep) {
result.add(dataList.get(idx));
}
}
return result;
}
/**
* 安全获取Float值，null时返回0
*/
private float getFloatValue(List<Float> data, int colIdx) {
if (colIdx >= data.size()) {
return 0f;
}
Float val = data.get(colIdx);
return val != null ? val : 0f;
}
/**
* 找到与目标时间最接近的数据点索引
* 仅在容差1.0ms内匹配，超出容差视为该段数据中无此关键点
*/
private int findClosestIndex(List<List<Float>> dataList, double targetTime) {
int closestIdx = -1;
double minDiff = Double.MAX_VALUE;
for (int i = 0; i < dataList.size(); i++) {
double time = getTimeValue(dataList.get(i));
double diff = Math.abs(time - targetTime);
if (diff < minDiff) {
minDiff = diff;
closestIdx = i;
}
}
// 容差1.0ms，超出则认为该段不含此关键点
if (minDiff > 1.0) {
return -1;
}
return closestIdx;
}
private double getTimeValue(List<Float> data) {
return data.get(0).doubleValue();
}
/**
* 在两段数据的间隔区域补固定数量的null数据点
* null点的时间在gapStart和gapEnd之间均匀分布
* 每个null点：[时间, null, null, ...]，echarts遇到null值会渲染为断线
*
* @param gapStart 段1最后一个点时间(ms)
* @param gapEnd 段2第一个点时间(ms)
* @param nullPointCount 补null点的数量
* @param columnCount 每行数据列数（含时间列）
*/
private List<List<Float>> createNullGapData(double gapStart, double gapEnd, int nullPointCount, int columnCount) {
List<List<Float>> gapData = new ArrayList<>();
if (gapEnd <= gapStart || nullPointCount <= 0) {
return gapData;
}
// 将gapStart到gapEnd之间均匀分成 nullPointCount+1 段
// null点放在中间的 nullPointCount 个位置，两端留给段1末点和段2首点
double step = (gapEnd - gapStart) / (nullPointCount + 1);
for (int i = 1; i <= nullPointCount; i++) {
double time = gapStart + step * i;
List<Float> nullPoint = new ArrayList<>();
nullPoint.add((float) (Math.round(time * 100.0) / 100.0));
for (int j = 1; j < columnCount; j++) {
nullPoint.add(null);
}
gapData.add(nullPoint);
}
return gapData;
}
/**
* 按时间范围筛选数据列表
*
* @param dataList 原始数据列表，每行第一个元素为时间坐标(ms)
* @param startTime 起始时间(ms)
* @param endTime 结束时间(ms)
* @return 范围内的数据列表
*/
private List<List<Float>> filterByTimeRange(List<List<Float>> dataList, double startTime, double endTime) {
return dataList.stream()
.filter(data -> {
double time = ((Number) data.get(0)).doubleValue();
return time >= startTime && time <= endTime;
})
.collect(Collectors.toList());
}
}

738
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/DataServiceImpl.java
View File

@@ -8,15 +8,15 @@ import cn.hutool.core.util.ObjectUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.google.common.collect.Lists;
import com.njcn.access.api.CsTopicFeignClient;
import com.njcn.access.enums.AccessEnum;
import com.njcn.access.enums.TypeEnum;
import com.njcn.access.pojo.dto.AskDataDto;
import com.njcn.access.pojo.dto.ReqAndResDto;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient;
import com.njcn.csdevice.api.WlRecordFeignClient;
import com.njcn.csdevice.api.*;
import com.njcn.csdevice.pojo.dto.CsEquipmentDeliveryDTO;
import com.njcn.csdevice.pojo.dto.DevDetailDTO;
import com.njcn.csdevice.pojo.param.WlRecordParam;
import com.njcn.csdevice.pojo.po.CsDataSet;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
@@ -25,33 +25,37 @@ import com.njcn.csdevice.pojo.po.WlRecord;
import com.njcn.csdevice.pojo.vo.DataGroupEventVO;
import com.njcn.csdevice.pojo.vo.EachModuleVO;
import com.njcn.csdevice.pojo.vo.RecordVo;
import com.njcn.csdevice.utils.DataChangeUtil;
import com.njcn.csdevice.utils.ReflectUtils;
import com.njcn.csharmonic.constant.HarmonicConstant;
import com.njcn.csharmonic.mapper.CsDataSetMapper;
import com.njcn.csharmonic.param.DataParam;
import com.njcn.csharmonic.pojo.po.CsEventPO;
import com.njcn.csharmonic.pojo.vo.AppLineDetailVo;
import com.njcn.csharmonic.pojo.vo.RealTimeDataVo;
import com.njcn.csharmonic.service.CsEventPOService;
import com.njcn.csharmonic.service.IDataService;
import com.njcn.csharmonic.util.InfluxDbParamUtil;
import com.njcn.influx.constant.InfluxDbSqlConstant;
import com.njcn.influx.imapper.*;
import com.njcn.influx.pojo.bo.CommonQueryParam;
import com.njcn.influx.pojo.dto.EventDataSetDTO;
import com.njcn.influx.pojo.dto.StatisticalDataDTO;
import com.njcn.influx.pojo.po.*;
import com.njcn.influx.pojo.po.cs.ApfData;
import com.njcn.influx.query.InfluxQueryWrapper;
import com.njcn.influx.service.CommonService;
import com.njcn.influx.service.EvtDataService;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.api.CsStatisticalSetFeignClient;
import com.njcn.system.api.DictTreeFeignClient;
import com.njcn.system.api.EleEvtFeignClient;
import com.njcn.system.api.EpdFeignClient;
import com.njcn.system.pojo.po.EleEpdPqd;
import com.njcn.system.pojo.po.EleEvtParm;
import com.njcn.system.api.*;
import com.njcn.system.pojo.po.*;
import com.njcn.system.pojo.vo.DictTreeVO;
import lombok.AllArgsConstructor;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.text.DecimalFormat;
import java.time.Duration;
import java.time.Instant;
@@ -91,7 +95,37 @@ public class DataServiceImpl implements IDataService {
private static Integer mid = 1;
private final CsTopicFeignClient csTopicFeignClient;
private final MqttPublisher publisher;
private static final List<String> ONLINE_STATES = Arrays.asList("运行", "停止", "故障");
private final CsLedgerFeignClient csLedgerFeignClient;
private final DicDataFeignClient dicDataFeignClient;
@Resource
private DataVMapper dataVMapper;
@Resource
private DataIMapper dataIMapper;
@Resource
private DataFlickerMapper dataFlickerMapper;
@Resource
private DataFlucMapper dataFlucMapper;
@Resource
private DataHarmPhasicVMapper dataHarmPhasicVMapper;
@Resource
private DataHarmPhasicIMapper dataHarmPhasicIMapper;
@Resource
private DataHarmPowerPMapper dataHarmPowerPMapper;
@Resource
private DataHarmPowerQMapper dataHarmPowerQMapper;
@Resource
private DataHarmPowerSMapper dataHarmPowerSMapper;
@Resource
private DataHarmRateVMapper dataHarmRateVMapper;
@Resource
private DataHarmRateIMapper dataHarmRateIMapper;
@Resource
private DataInHarmVMapper dataInHarmVMapper;
@Resource
private DataPltMapper dataPltMapper;
@Resource
private ApfDataMapper apfDataMapper;
private final DataSetFeignClient dataSetFeignClient;
@Override
public List<RealTimeDataVo> getRealTimeData(DataParam param) {
@@ -108,7 +142,7 @@ public class DataServiceImpl implements IDataService {
//根据分组获取对应指标
List<RealTimeDataVo> finalResult = result;
dictTreeVOList.forEach(item->{
List<EleEpdPqd> epdPqdList = csStatisticalSetFeignClient.queryStatisticalSelect(item.getId()).getData();
List<EleEpdPqd> epdPqdList = csStatisticalSetFeignClient.queryStatisticalSelect(Collections.singletonList(item.getId())).getData();
epdPqdList.forEach(item2->{
if (Objects.isNull(item2.getHarmStart())) {
RealTimeDataVo vo = getBaseData(item2,param.getLineId(),param.getDataLevel(),csDataSet.getDataLevel(),pt,ct);
@@ -435,6 +469,688 @@ public class DataServiceImpl implements IDataService {
return result;
}
//1、根据监测点集合 100为1个单位进行查询
//2、根据表名来进行查询 指标可以用循环的方式拼接起来 influxQueryWrapper.select("rms","rms")
//3、根据相别来分开查询，数据类型统一查询avg
@Override
public List<AppLineDetailVo> getLineDataByEngineer(String id) {
List<AppLineDetailVo> result = new ArrayList<>();
//根据工程获取监测点信息
List<DevDetailDTO> data = csLedgerFeignClient.getDevInfoByEngineerIds(Collections.singletonList(id)).getData();
if (CollectionUtil.isNotEmpty(data)) {
List<DataV> data1 = new ArrayList<>();
List<DataI> data2 = new ArrayList<>();
List<DataFlicker> data3 = new ArrayList<>();
List<DataFluc> data4 = new ArrayList<>();
List<DataHarmPhasicV> data5 = new ArrayList<>();
List<DataHarmPhasicI> data6 = new ArrayList<>();
List<DataHarmPowerP> data7 = new ArrayList<>();
List<DataHarmPowerQ> data8 = new ArrayList<>();
List<DataHarmPowerS> data9 = new ArrayList<>();
List<DataHarmRateV> data10 = new ArrayList<>();
List<DataHarmRateI> data11 = new ArrayList<>();
List<DataInHarmV> data12 = new ArrayList<>();
List<DataPlt> data13 = new ArrayList<>();
List<ApfData> data14 = new ArrayList<>();
//获取监测点集合 针对监测点的类型区分查询数据-治理监测点、监测监测点
List<String> lineList = data.stream().flatMap(v -> v.getLineList().stream()).collect(Collectors.toList());
if (CollectionUtil.isEmpty(lineList)) {
return result;
}
List<CsLinePO> linePo = csLineFeignClient.queryLineById(lineList).getData();
Map<String,CsLinePO> lineMap = linePo.stream().collect(Collectors.toMap(CsLinePO::getLineId, item->item));
//拆分治理监测点 、 监测监测点
List<CsLinePO> line1 = linePo.stream().filter(item->item.getClDid()==0).collect(Collectors.toList());
List<CsLinePO> line2 = linePo.stream().filter(item->item.getClDid()!=0).collect(Collectors.toList());
List<String> lineList2 = line2.stream().map(CsLinePO::getLineId).collect(Collectors.toList());
List<List<String>> partitions = Lists.partition(lineList2, 95);
//表集合
List<DictData> dataList = dicDataFeignClient.getDicDataByTypeCode("Data").getData();
Map<String, DictData> map = dataList.stream().collect(Collectors.toMap(DictData::getId, v -> v));
//获取需要查询的指标
List<SysDicTreePO> list = dictTreeFeignClient.queryByCodeList("Key_Power_Quality").getData();
List<SysDicTreePO> children = list.get(0).getChildren();
children.sort(Comparator.comparing(SysDicTreePO::getSort));
List<String> idList = children.stream().map(SysDicTreePO::getId).collect(Collectors.toList());
List<EleEpdPqd> epdPqdList = csStatisticalSetFeignClient.queryStatisticalSelect(idList).getData();
Map<String,List<EleEpdPqd>> epdPqdMap = epdPqdList.stream().collect(Collectors.groupingBy(EleEpdPqd::getClassId));
Map<String,EleEpdPqd> epdPqdMap2 = epdPqdList.stream().collect(Collectors.toMap(EleEpdPqd::getId, v -> v));
//获取监测点的数据集，转换一二次值
List<String> dataSetIdList = linePo.stream().map(CsLinePO::getDataSetId).distinct().collect(Collectors.toList());
List<CsDataSet> csDataSetList = dataSetFeignClient.getDataSetBySetIds(dataSetIdList).getData();
Map<String,CsDataSet> csDataSetMap = csDataSetList.stream().collect(Collectors.toMap(CsDataSet::getId, v -> v));
//获取指标名称和对应的指标集合
List<CsStatisticalSetPO> csStatisticalSetPOList = csStatisticalSetFeignClient.queryStatisticalById(idList).getData();
Map<String,List<CsStatisticalSetPO>> csStatisticalSetPOMap = csStatisticalSetPOList.stream().collect(Collectors.groupingBy(CsStatisticalSetPO::getStatisicalId));
epdPqdMap.forEach((k,v) ->{
if (Objects.equals(map.get(k).getName(), "data_v")) {
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataV::getLineId, l)
.eq(DataV::getValueType,"AVG")
.groupBy(DataV::getPhaseType)
.groupBy(DataV::getLineId)
.timeDesc().limit(1);
List<DataV> dataV = dataVMapper.selectByQueryWrapper(influxQueryWrapper);
data1.addAll(dataV);
});
} else if (Objects.equals(map.get(k).getName(), "data_i")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataI.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataI::getLineId, l)
.eq(DataI::getValueType,"AVG")
.groupBy(DataI::getPhaseType)
.groupBy(DataI::getLineId)
.timeDesc().limit(1);
List<DataI> dataI = dataIMapper.selectByQueryWrapper(influxQueryWrapper);
data2.addAll(dataI);
});
} else if (Objects.equals(map.get(k).getName(), "data_flicker")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataFlicker.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataFlicker::getLineId, l)
.eq(DataFlicker::getValueType,"AVG")
.groupBy(DataFlicker::getPhaseType)
.groupBy(DataFlicker::getLineId)
.timeDesc().limit(1);
List<DataFlicker> dataFlicker = dataFlickerMapper.selectByQueryWrapper(influxQueryWrapper);
data3.addAll(dataFlicker);
});
} else if (Objects.equals(map.get(k).getName(), "data_fluc")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataFluc.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataFluc::getLineId, l)
.eq(DataFluc::getValueType,"AVG")
.groupBy(DataFluc::getPhaseType)
.groupBy(DataFluc::getLineId)
.timeDesc().limit(1);
List<DataFluc> dataFluc = dataFlucMapper.selectByQueryWrapper(influxQueryWrapper);
data4.addAll(dataFluc);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmphasic_v")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmPhasicV.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmPhasicV::getLineId, l)
.eq(DataHarmPhasicV::getValueType,"AVG")
.groupBy(DataHarmPhasicV::getPhaseType)
.groupBy(DataHarmPhasicV::getLineId)
.timeDesc().limit(1);
List<DataHarmPhasicV> dataHarmPhasicV = dataHarmPhasicVMapper.selectByQueryWrapper(influxQueryWrapper);
data5.addAll(dataHarmPhasicV);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmphasic_i")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmPhasicI.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmPhasicI::getLineId, l)
.eq(DataHarmPhasicI::getValueType,"AVG")
.groupBy(DataHarmPhasicI::getPhaseType)
.groupBy(DataHarmPhasicI::getLineId)
.timeDesc().limit(1);
List<DataHarmPhasicI> dataHarmPhasicI = dataHarmPhasicIMapper.selectByQueryWrapper(influxQueryWrapper);
data6.addAll(dataHarmPhasicI);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmpower_p")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmPowerP.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmPowerP::getLineId, l)
.eq(DataHarmPowerP::getValueType,"AVG")
.groupBy(DataHarmPowerP::getPhaseType)
.groupBy(DataHarmPowerP::getLineId)
.timeDesc().limit(1);
List<DataHarmPowerP> dataHarmPowerP = dataHarmPowerPMapper.selectByQueryWrapper(influxQueryWrapper);
data7.addAll(dataHarmPowerP);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmpower_q")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmPowerQ.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmPowerQ::getLineId, l)
.eq(DataHarmPowerQ::getValueType,"AVG")
.groupBy(DataHarmPowerQ::getPhaseType)
.groupBy(DataHarmPowerQ::getLineId)
.timeDesc().limit(1);
List<DataHarmPowerQ> dataHarmPowerQ = dataHarmPowerQMapper.selectByQueryWrapper(influxQueryWrapper);
data8.addAll(dataHarmPowerQ);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmpower_s")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmPowerS.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmPowerS::getLineId, l)
.eq(DataHarmPowerS::getValueType,"AVG")
.groupBy(DataHarmPowerS::getPhaseType)
.groupBy(DataHarmPowerS::getLineId)
.timeDesc().limit(1);
List<DataHarmPowerS> dataHarmPowerS = dataHarmPowerSMapper.selectByQueryWrapper(influxQueryWrapper);
data9.addAll(dataHarmPowerS);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmrate_v")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmRateV.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmRateV::getLineId, l)
.eq(DataHarmRateV::getValueType,"AVG")
.groupBy(DataHarmRateV::getPhaseType)
.groupBy(DataHarmRateV::getLineId)
.timeDesc().limit(1);
List<DataHarmRateV> dataHarmRateV = dataHarmRateVMapper.selectByQueryWrapper(influxQueryWrapper);
data10.addAll(dataHarmRateV);
});
} else if (Objects.equals(map.get(k).getName(), "data_harmrate_i")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmRateI.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataHarmRateI::getLineId, l)
.eq(DataHarmRateI::getValueType,"AVG")
.groupBy(DataHarmRateI::getPhaseType)
.groupBy(DataHarmRateI::getLineId)
.timeDesc().limit(1);
List<DataHarmRateI> DataHarmRateI = dataHarmRateIMapper.selectByQueryWrapper(influxQueryWrapper);
data11.addAll(DataHarmRateI);
});
} else if (Objects.equals(map.get(k).getName(), "data_inharm_v")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataInHarmV.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataInHarmV::getLineId, l)
.eq(DataInHarmV::getValueType,"AVG")
.groupBy(DataInHarmV::getPhaseType)
.groupBy(DataInHarmV::getLineId)
.timeDesc().limit(1);
List<DataInHarmV> dataInHarmV = dataInHarmVMapper.selectByQueryWrapper(influxQueryWrapper);
data12.addAll(dataInHarmV);
});
} else if (Objects.equals(map.get(k).getName(), "data_plt")){
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataPlt.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(DataPlt::getLineId, l)
.eq(DataPlt::getValueType,"AVG")
.groupBy(DataPlt::getPhaseType)
.groupBy(DataPlt::getLineId)
.timeDesc().limit(1);
List<DataPlt> dataPlt = dataPltMapper.selectByQueryWrapper(influxQueryWrapper);
data13.addAll(dataPlt);
});
}
});
//监测点第一层
line2.forEach(item->{
AppLineDetailVo vo = new AppLineDetailVo();
for (DevDetailDTO item2 : data) {
if (item2.getLineList() != null && item2.getLineList().contains(item.getLineId())) {
vo.setProjectName(item2.getProjectName());
vo.setDeviceName(item2.getEquipmentName());
break;
}
}
vo.setLineType(1);
vo.setPointId(item.getLineId());
vo.setPointName(lineMap.get(item.getLineId()).getName());
List<AppLineDetailVo.targetDetail> children2 = new ArrayList<>();
//指标第二层
List<AppLineDetailVo.targetDetail> finalChildren = children2;
children.forEach(item3->{
List<CsStatisticalSetPO> l1 = csStatisticalSetPOMap.get(item3.getId());
l1.forEach(item1 -> {
EleEpdPqd epdPqd = epdPqdMap2.get(item1.getTargetId());
AppLineDetailVo.targetDetail targetDetail = new AppLineDetailVo.targetDetail();
targetDetail.setTargetId(item3.getId());
targetDetail.setName(item3.getName());
targetDetail.setPhase(epdPqd.getPhase());
targetDetail.setUnit(epdPqd.getUnit());
targetDetail.setSort(item3.getSort());
targetDetail.setPrimaryFormula(epdPqd.getPrimaryFormula());
if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_v")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataV matchedDataV = data1.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_i")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataI matchedDataV = data2.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_flicker")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataFlicker matchedDataV = data3.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_fluc")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataFluc matchedDataV = data4.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmphasic_v")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmPhasicV matchedDataV = data5.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmphasic_i")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmPhasicI matchedDataV = data6.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmpower_p")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmPowerP matchedDataV = data7.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmpower_q")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmPowerQ matchedDataV = data8.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmpower_s")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmPowerS matchedDataV = data9.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmrate_v")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmRateV matchedDataV = data10.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_harmrate_i")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataHarmRateI matchedDataV = data11.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_inharm_v")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataInHarmV matchedDataV = data12.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
} else if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "data_plt")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
DataPlt matchedDataV = data13.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
}
//一二次值处理
if (!Objects.isNull(targetDetail.getData())) {
String dataLevel = csDataSetMap.get(lineMap.get(item.getLineId()).getDataSetId()).getDataLevel();
if (Objects.equals(dataLevel, "Primary")) {
if (targetDetail.getPrimaryFormula() != null && ("*PT".equals(targetDetail.getPrimaryFormula()) || "*PT*CT".equals(targetDetail.getPrimaryFormula()))) {
targetDetail.setUnit("k" + targetDetail.getUnit());
targetDetail.setData(Math.round(targetDetail.getData() * 100.0) / 100.0);
}
} else if (Objects.equals(dataLevel, "Secondary")) {
Double ct = item.getCtRatio() / (Objects.isNull(item.getCt2Ratio())?1.0:item.getCt2Ratio());
Double pt = item.getPtRatio() / (Objects.isNull(item.getPt2Ratio())?1.0:item.getPt2Ratio());
String formula = targetDetail.getPrimaryFormula();
if (formula == null) {
targetDetail.setData(Math.round(targetDetail.getData() * 100.0) / 100.0);
}
if ("*PT".equals(formula) || "*PT*CT".equals(formula)) {
targetDetail.setUnit("k" + targetDetail.getUnit());
double finalData = DataChangeUtil.secondaryToPrimary(formula, targetDetail.getData(), pt, ct);
targetDetail.setData(Math.round((finalData/1000) * 100.0) / 100.0);
}
if ("*CT".equals(formula)) {
double finalData = DataChangeUtil.secondaryToPrimary(formula, targetDetail.getData(), pt, ct);
targetDetail.setData(Math.round(finalData * 100.0) / 100.0);
}
}
}
finalChildren.add(targetDetail);
});
});
// 根据接线方式处理相别
chanelPhase(lineMap, item, finalChildren, children2);
if (CollUtil.isNotEmpty(children2)) {
children2.sort(Comparator.comparingInt(AppLineDetailVo.targetDetail::getSort));
}
vo.setChildren(children2);
result.add(vo);
});
if (CollectionUtil.isNotEmpty(line1)) {
List<AppLineDetailVo> zhiLiLineData = getZhiLiLineData(line1,data14, map, data, lineMap, csDataSetMap);
result.addAll(zhiLiLineData);
}
}
if (CollUtil.isNotEmpty(result)) {
result.sort(Comparator.comparing(AppLineDetailVo::getProjectName)
.thenComparing(AppLineDetailVo::getDeviceName)
.thenComparing(AppLineDetailVo::getPointId));
}
return result;
}
public void chanelPhase(Map<String,CsLinePO> lineMap,CsLinePO item,List<AppLineDetailVo.targetDetail> finalChildren, List<AppLineDetailVo.targetDetail> children2) {
// 根据接线方式处理相别
Integer conType = lineMap.get(item.getLineId()).getConType();
if (conType != null && conType == 0) {
// 星型接线: 保留A/B/C/T，去除AB/BC/CA
finalChildren.removeIf(d -> "AB".equals(d.getPhase()) || "BC".equals(d.getPhase()) || "CA".equals(d.getPhase()));
} else if (conType != null) {
// 角型/V型接线: A/B/C数据用AB/BC/CA替换，然后去除AB/BC/CA
Map<String, List<AppLineDetailVo.targetDetail>> groupByTargetId = finalChildren.stream()
.collect(Collectors.groupingBy(AppLineDetailVo.targetDetail::getTargetId));
List<AppLineDetailVo.targetDetail> newChildren2 = new ArrayList<>();
for (List<AppLineDetailVo.targetDetail> group : groupByTargetId.values()) {
// 提取AB/BC/CA的数据
Map<String, Double> abDataMap = new HashMap<>();
for (AppLineDetailVo.targetDetail d : group) {
if ("AB".equals(d.getPhase()) || "BC".equals(d.getPhase()) || "CA".equals(d.getPhase())) {
abDataMap.put(d.getPhase(), d.getData());
}
}
// 用AB/BC/CA替换A/B/C数据，保留T相
for (AppLineDetailVo.targetDetail d : group) {
if ("A".equals(d.getPhase())) {
if (abDataMap.containsKey("AB")) {
d.setData(abDataMap.get("AB"));
}
newChildren2.add(d);
} else if ("B".equals(d.getPhase())) {
if (abDataMap.containsKey("BC")) {
d.setData(abDataMap.get("BC"));
}
newChildren2.add(d);
} else if ("C".equals(d.getPhase())) {
if (abDataMap.containsKey("CA")) {
d.setData(abDataMap.get("CA"));
}
newChildren2.add(d);
} else if ("T".equals(d.getPhase())) {
newChildren2.add(d);
}
}
}
children2 = newChildren2;
}
}
//处理治理监测点
public List<AppLineDetailVo> getZhiLiLineData(List<CsLinePO> poList
,List<ApfData> data14
, Map<String, DictData> map
, List<DevDetailDTO> data
, Map<String,CsLinePO> lineMap
, Map<String,CsDataSet> csDataSetMap) {
List<AppLineDetailVo> result = new ArrayList<>();
if (CollUtil.isEmpty(poList)) {
return result;
}
List<String> lineList = poList.stream().map(CsLinePO::getLineId).collect(Collectors.toList());
List<List<String>> partitions = Lists.partition(lineList, 95);
//获取需要查询的指标
List<SysDicTreePO> list = dictTreeFeignClient.queryByCodeList("Key_Power_Quality_Pq_Com").getData();
List<SysDicTreePO> children = list.get(0).getChildren();
children.sort(Comparator.comparing(SysDicTreePO::getSort));
List<String> idList = children.stream().map(SysDicTreePO::getId).collect(Collectors.toList());
List<EleEpdPqd> epdPqdList = csStatisticalSetFeignClient.queryStatisticalSelect(idList).getData();
Map<String,List<EleEpdPqd>> epdPqdMap = epdPqdList.stream().collect(Collectors.groupingBy(EleEpdPqd::getClassId));
Map<String,EleEpdPqd> epdPqdMap2 = epdPqdList.stream().collect(Collectors.toMap(EleEpdPqd::getId, v -> v));
//获取指标名称和对应的指标集合
List<CsStatisticalSetPO> csStatisticalSetPOList = csStatisticalSetFeignClient.queryStatisticalById(idList).getData();
Map<String,List<CsStatisticalSetPO>> csStatisticalSetPOMap = csStatisticalSetPOList.stream().collect(Collectors.groupingBy(CsStatisticalSetPO::getStatisicalId));
epdPqdMap.forEach((k,v) ->{
if (Objects.equals(map.get(k).getName(), "apf_data")) {
partitions.forEach(l -> {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(ApfData.class);
v.stream().map(EleEpdPqd::getOtherName).distinct().forEach(otherName -> influxQueryWrapper.select(otherName, otherName));
influxQueryWrapper.regular(ApfData::getLineId, l)
.eq(ApfData::getValueType,"AVG")
.groupBy(ApfData::getPhaseType)
.groupBy(ApfData::getLineId)
.timeDesc().limit(1);
List<ApfData> apfData = apfDataMapper.selectByQueryWrapper(influxQueryWrapper);
data14.addAll(apfData);
});
}
});
//监测点第一层
poList.forEach(item->{
AppLineDetailVo vo = new AppLineDetailVo();
for (DevDetailDTO item2 : data) {
if (item2.getLineList() != null && item2.getLineList().contains(item.getLineId())) {
vo.setProjectName(item2.getProjectName());
vo.setDeviceName(item2.getEquipmentName());
break;
}
}
vo.setLineType(0);
vo.setPointId(item.getLineId());
vo.setPointName(lineMap.get(item.getLineId()).getName());
List<AppLineDetailVo.targetDetail> children2 = new ArrayList<>();
//指标第二层
List<AppLineDetailVo.targetDetail> finalChildren = children2;
children.forEach(item3->{
List<CsStatisticalSetPO> l1 = csStatisticalSetPOMap.get(item3.getId());
l1.forEach(item1 -> {
EleEpdPqd epdPqd = epdPqdMap2.get(item1.getTargetId());
AppLineDetailVo.targetDetail targetDetail = new AppLineDetailVo.targetDetail();
targetDetail.setTargetId(item3.getId());
targetDetail.setName(item3.getName());
targetDetail.setPhase(epdPqd.getPhase());
targetDetail.setUnit(epdPqd.getUnit());
targetDetail.setSort(item3.getSort());
targetDetail.setPrimaryFormula(epdPqd.getPrimaryFormula());
if (Objects.equals(map.get(epdPqd.getClassId()).getName(), "apf_data")) {
//根据otherName反射获取DataV中的字段值，按lineId和相别匹配
String otherName = epdPqd.getOtherName();
ApfData matchedDataV = data14.stream()
.filter(d -> Objects.equals(d.getLineId(), item.getLineId())
&& Objects.equals(d.getPhaseType(), epdPqd.getPhase()))
.findFirst()
.orElse(null);
if (matchedDataV != null) {
Object value = ReflectUtils.getValue(matchedDataV, toCamelCase(otherName));
if (value instanceof Number) {
targetDetail.setData(((Number) value).doubleValue());
}
vo.setDataTime(matchedDataV.getTime());
}
}
//一二次值处理
if (!Objects.isNull(targetDetail.getData())) {
String dataLevel = csDataSetMap.get(lineMap.get(item.getLineId()).getDataSetId()).getDataLevel();
if (Objects.equals(dataLevel, "Primary")) {
targetDetail.setUnit(targetDetail.getUnit());
targetDetail.setData(Math.round(targetDetail.getData() * 100.0) / 100.0);
} else if (Objects.equals(dataLevel, "Secondary")) {
Double ct = item.getCtRatio() / (Objects.isNull(item.getCt2Ratio())?1.0:item.getCt2Ratio());
Double pt = item.getPtRatio() / (Objects.isNull(item.getPt2Ratio())?1.0:item.getPt2Ratio());
String formula = targetDetail.getPrimaryFormula();
if (formula == null) {
targetDetail.setData(Math.round(targetDetail.getData() * 100.0) / 100.0);
}
if ("*PT".equals(formula) || "*PT*CT".equals(formula)) {
targetDetail.setUnit("k" + targetDetail.getUnit());
double finalData = DataChangeUtil.secondaryToPrimary(formula, targetDetail.getData(), pt, ct);
targetDetail.setData(Math.round((finalData/1000) * 100.0) / 100.0);
}
if ("*CT".equals(formula)) {
double finalData = DataChangeUtil.secondaryToPrimary(formula, targetDetail.getData(), pt, ct);
targetDetail.setData(Math.round(finalData * 100.0) / 100.0);
}
}
}
finalChildren.add(targetDetail);
});
});
// 根据接线方式处理相别
chanelPhase(lineMap, item, finalChildren, children2);
if (CollUtil.isNotEmpty(children2)) {
children2.sort(Comparator.comparingInt(AppLineDetailVo.targetDetail::getSort));
}
vo.setChildren(children2);
result.add(vo);
});
return result;
}
public String toCamelCase(String input) {
if (input == null || input.isEmpty()) {
return input;
}
StringBuilder result = new StringBuilder();
boolean toUpper = false;
for (char c : input.toCharArray()) {
if (c == '_') {
toUpper = true;
} else {
if (toUpper) {
result.append(Character.toUpperCase(c));
toUpper = false;
} else {
result.append(c);
}
}
}
// 确保首字母小写，以匹配Java实体字段命名规范
if (result.length() > 0 && Character.isUpperCase(result.charAt(0))) {
result.setCharAt(0, Character.toLowerCase(result.charAt(0)));
}
return result.toString();
}
public int compareWithCurrentTime(String timeString) {
// 定义日期时间格式
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

81
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/StableDataServiceImpl.java
View File

@@ -36,10 +36,7 @@ import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -264,7 +261,7 @@ public class StableDataServiceImpl implements StableDataService {
Optional.ofNullable(csLinePOList).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR));
List<CsLinePO> csLinePOList1 = csLinePOList.stream().filter(temp -> Objects.equals(areaId, temp.getPosition())).collect(Collectors.toList());
List<EleEpdPqd> data = csStatisticalSetFeignClient.queryStatisticalSelect(commonStatisticalQueryParam.getStatisticalId()).getData();
List<EleEpdPqd> data = csStatisticalSetFeignClient.queryStatisticalSelect(Collections.singletonList(commonStatisticalQueryParam.getStatisticalId())).getData();
// EleEpdPqd data = epdFeignClient.selectById(commonStatisticalQueryParam.getStatisticalId()).getData();
// Optional.ofNullable(data).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.ELEEPDPQD_DATA_ERROR));
@@ -322,80 +319,6 @@ public class StableDataServiceImpl implements StableDataService {
return result;
}
// @Override
// public List<ThdDataVO> queryLineCommonStatistical(CommonStatisticalQueryParam commonStatisticalQueryParam) {
// List<ThdDataVO> result = new ArrayList();
// if(CollectionUtil.isEmpty(commonStatisticalQueryParam.getLineList())){
// throw new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR);
// }
// List<CsLinePO> csLinePOList = csLineFeignClient.queryLineById(commonStatisticalQueryParam.getLineList()).getData();
// if(CollectionUtil.isEmpty(csLinePOList)){
// throw new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR);
// }
//
// LineParamDTO lineParamDTO = new LineParamDTO();
// lineParamDTO.setLineId(commonStatisticalQueryParam.getLineList().get(0));
// List<CsLedger> csLedgers = csLedgerFeignClient.queryLine(lineParamDTO).getData();
// List<CsEquipmentDeliveryDTO> data1 = equipmentFeignClient.queryDeviceById(Stream.of(csLedgers.get(0).getPid()).collect(Collectors.toList())).getData();
//
//
// List<String> collect = csLinePOList.stream().map(CsLinePO::getLineId).collect(Collectors.toList());
// String areaId = dicDataFeignClient.getDicDataByCode(DicDataEnum.OUTPUT_SIDE.getCode()).getData().getId();
// Optional.ofNullable(csLinePOList).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR));
//// collect = csLinePOList.stream().filter(temp->Objects.equals(areaId,temp.getPosition())).map(CsLinePO::getLineId).collect(Collectors.toList());
//
// List<EleEpdPqd> data = csStatisticalSetFeignClient.queryStatisticalSelect(commonStatisticalQueryParam.getStatisticalId()).getData();
//
//// EleEpdPqd data = epdFeignClient.selectById(commonStatisticalQueryParam.getStatisticalId()).getData();
// Optional.ofNullable(data).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.ELEEPDPQD_DATA_ERROR));
//
// String frequency = Optional.ofNullable(commonStatisticalQueryParam.getFrequency()).orElse("");
// if(CollectionUtil.isNotEmpty(data)){
// List<String> finalCollect = collect;
// data.forEach(epdPqd->{
// List<CommonQueryParam> commonQueryParams = finalCsLinePOList.stream().map(temp -> {
// CommonQueryParam commonQueryParam = new CommonQueryParam();
// commonQueryParam.setLineId(temp.getLineId());
// commonQueryParam.setTableName(getTableNameByClassId(epdPqd.getClassId()));
// commonQueryParam.setColumnName(epdPqd.getName()+ finalFrequency1);
// commonQueryParam.setPhasic(epdPqd.getPhase());
// commonQueryParam.setStartTime(commonStatisticalQueryParam.getStartTime());
// commonQueryParam.setEndTime(commonStatisticalQueryParam.getEndTime());
// commonQueryParam.setDataType(commonStatisticalQueryParam.getValueType());
// commonQueryParam.setProcess(data1.get(0).getProcess()+"");
// commonQueryParam.setClDid(getClDidByLineId(temp.getLineId()));
//
// return commonQueryParam;
// }).collect(Collectors.toList());
// List<StatisticalDataDTO> deviceRtData = commonService.getDeviceRtDataByTime(finalCollect, epdPqd.getClassId(), epdPqd.getName()+frequency, epdPqd.getPhase(), commonStatisticalQueryParam.getValueType(),commonStatisticalQueryParam.getStartTime(),commonStatisticalQueryParam.getEndTime(),data1.get(0).getProcess()+"");
//
//
// List<ThdDataVO> collect1 = deviceRtData.stream().map(temp -> {
// ThdDataVO vo = new ThdDataVO();
// vo.setLineId(temp.getLineId());
// vo.setPhase(temp.getPhaseType());
// String position = csLinePOList.stream().filter(csLinePO -> Objects.equals(csLinePO.getLineId(), vo.getLineId())).collect(Collectors.toList()).get(0).getPosition();
// String lineName = csLinePOList.stream().filter(csLinePO -> Objects.equals(csLinePO.getLineId(), vo.getLineId())).collect(Collectors.toList()).get(0).getName();
//
// vo.setLineName(lineName);
// vo.setPosition(position);
// vo.setTime(temp.getTime());
// vo.setStatMethod(temp.getValueType());
// vo.setStatisticalData(Double.valueOf(df.format(temp.getValue())));
// vo.setStatisticalIndex(epdPqd.getId());
// vo.setUnit(epdPqd.getUnit());
// vo.setStatisticalName(epdPqd.getName());
// vo.setAnotherName(epdPqd.getShowName());
// return vo;
// }).collect(Collectors.toList());
// collect1 = collect1.stream().distinct().collect(Collectors.toList());
// result.addAll(collect1);
// });
// }
//
// return result;
// }
private String phaseReflection(String phase){
switch (phase) {
case "AB":