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

feat(communication): 添加装置数据查询功能并重构在线率统计逻辑

Browse Source
This commit is contained in:
xy
2026-05-08 10:29:24 +08:00
parent 3990ad2b9a
commit 57cb6a2900
4 changed files with 137 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsCommunicateService.java
View File

@@ -3,6 +3,7 @@ package com.njcn.csdevice.service;
import com.njcn.csdevice.param.LineCountEvaluateParam;
import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
import com.njcn.influx.pojo.po.PqsCommunicate;
import java.util.List;
@@ -20,6 +21,13 @@ public interface ICsCommunicateService {
*/
List<PqsCommunicateDto> getRawDataLatest(LineCountEvaluateParam lineParam);
/**
* 取出第一条装置数据
* @param lineParam
* @return
*/
List<PqsCommunicateDto> getRawDataOne(LineCountEvaluateParam lineParam);
/**
* 获取时间范围数据
* @param lineParam
@@ -27,6 +35,8 @@ public interface ICsCommunicateService {
*/
List<PqsCommunicateDto> getRawData(LineCountEvaluateParam lineParam);
List<PqsCommunicateDto> getRawData2(LineCountEvaluateParam lineParam);
/**
*是否有当天最后一条数据
* @param lineParam

13
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/IRStatOnlineRateDService.java
View File

@@ -16,6 +16,19 @@ import java.util.List;
*/
public interface IRStatOnlineRateDService extends IService<RStatOnlineRateD> {
/**
* 1.查询该设备当前计算周期是否存在上下线记录
* 2.
* a.存在记录，如果只有1条数据，则根据这一条记录的状态计算在线时长；如果是多条数据，则根据上下线时间计算在线时长；
* b.不存在记录，则需要借助历史记录判断：
* 1) 先查询设备第一条记录时间
* - 如果统计日期 < 第一条记录时间 → 设备未上线 → 离线
* 2) 再查询最新记录时间和状态
* - 如果统计日期 > 最新记录时间 → 延续最新状态
* * 最新是在线 → 全天在线
* * 最新是离线 → 全天离线
* @param param
*/
void addData(StatisticsDataParam param);
List<RStatOnlineRateD> getData(List<String> list, String startTime, String endTime);

33
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/InfluxdbCsCommunicateServiceImpl.java
View File

@@ -66,6 +66,27 @@ public class InfluxdbCsCommunicateServiceImpl implements ICsCommunicateService {
return result;
}
@Override
public List<PqsCommunicateDto> getRawDataOne(LineCountEvaluateParam lineParam) {
List<PqsCommunicateDto> result = new ArrayList<>();
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);
influxQueryWrapper.regular(PqsCommunicate::getDevId, lineParam.getLineId())
.select(PqsCommunicate::getTime)
.select(PqsCommunicate::getDevId)
.select(PqsCommunicate::getDescription)
.select(PqsCommunicate::getType)
.timeAsc()
.limit(1);
List<PqsCommunicate> list = pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper);
list.forEach(item -> {
PqsCommunicateDto dto = new PqsCommunicateDto();
BeanUtils.copyProperties(item, dto);
dto.setTime(DATE_TIME_FORMATTER.format(item.getTime()));
result.add(dto);
});
return result;
}
/**
* @Description: 获取时间段内的数据
* @Param:
@@ -87,6 +108,18 @@ public class InfluxdbCsCommunicateServiceImpl implements ICsCommunicateService {
return result;
}
@Override
public List<PqsCommunicateDto> getRawData2(LineCountEvaluateParam lineParam) {
List<PqsCommunicateDto> result = new ArrayList<>();
List<PqsCommunicate> list = getPqsCommunicateData(lineParam);
if (CollectionUtil.isNotEmpty(list)) {
list.forEach(item -> {
result.add(convertToDto(item));
});
}
return result;
}
/**
* 处理有数据的情况
*/

212
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/RStatOnlineRateDServiceImpl.java
View File

@@ -16,13 +16,10 @@ import com.njcn.csdevice.service.ICsCommunicateService;
import com.njcn.csdevice.service.IRStatOnlineRateDService;
import com.njcn.csdevice.util.TimeUtil;
import com.njcn.csharmonic.pojo.param.StatisticsDataParam;
import com.njcn.influx.deprecated.InfluxDBPublicParam;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collections;
@@ -62,48 +59,94 @@ public class RStatOnlineRateDServiceImpl extends MppServiceImpl<RStatOnlineRateD
return;
}
if (CollectionUtil.isNotEmpty(devList)) {
//获取需要计算的时间
List<String> dateRange = TimeUtil.getDateRangeAsString(param.getStartTime(), param.getEndTime());
for (String time : dateRange) {
List<PqsCommunicateDto> outCommunicateData = new ArrayList<>();
int onlineMinutes = 0;
//获取需要计算的时间
List<String> dateRange = TimeUtil.getDateRangeAsString(param.getStartTime(), param.getEndTime());
for (String time : dateRange) {
Date statDate = DateUtil.parse(time);
// 按设备分别统计
for (CsEquipmentDeliveryPO device : devList) {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(time + " 00:00:00");
lineParam.setEndTime(time + " 23:59:59");
for (CsEquipmentDeliveryPO s : devList) {
lineParam.setLineId(Collections.singletonList(s.getId()));
List<PqsCommunicateDto> data = pqsCommunicateService.getRawDataLatest(lineParam);
if (CollectionUtil.isEmpty(data)) {
PqsCommunicateDto dto = new PqsCommunicateDto();
dto.setTime(time + " 00:00:00");
dto.setDevId(s.getId());
if (s.getRunStatus() == 1) {
dto.setType(0);
dto.setDescription("通讯中断");
} else if (s.getRunStatus() == 2) {
dto.setType(1);
dto.setDescription("通讯正常");
lineParam.setLineId(Collections.singletonList(device.getId()));
lineParam.setStartTime(time);
lineParam.setEndTime(time);
List<PqsCommunicateDto> dayData = pqsCommunicateService.getRawData2(lineParam);
if (CollectionUtil.isNotEmpty(dayData)) {
if (dayData.size() == 1) {
PqsCommunicateDto singleRecord = dayData.get(0);
long minutesFromMidnight = DateUtil.between(statDate, DateUtil.parse(singleRecord.getTime()), DateUnit.MINUTE);
if (online.equals(singleRecord.getType())) {
// 如果是在线状态，则从该时间点之后都在线
onlineMinutes = 1440 - (int) minutesFromMidnight;
} else {
// 如果是离线状态，则从该时间点之前都在线（假设之前是在线的）
onlineMinutes = (int) minutesFromMidnight;
}
outCommunicateData.add(dto);
} else {
// 多条记录，逐段计算
long totalOnlineMinutes = 0L;
Date lastTime = statDate;
for (int i = 0; i < dayData.size(); i++) {
PqsCommunicateDto current = dayData.get(i);
Date currentTime = DateUtil.parse(current.getTime());
long intervalMinutes = DateUtil.between(lastTime, currentTime, DateUnit.MINUTE);
if (i == 0) {
// 处理第一段：从0点到第一条记录
// 如果第一条记录是离线(type=0)，则之前是在线；如果第一条是在线(type=1)，则之前是离线
if (!online.equals(current.getType())) {
totalOnlineMinutes += intervalMinutes;
}
} else {
// 处理后续段：根据上一条记录的状态判断
if (online.equals(dayData.get(i - 1).getType())) {
totalOnlineMinutes += intervalMinutes;
}
}
lastTime = currentTime;
}
// 处理最后一段：从最后一条记录到当天结束
Date endOfDay = DateUtil.beginOfDay(DateUtil.offsetDay(statDate, 1));
long lastInterval = DateUtil.between(lastTime, endOfDay, DateUnit.MINUTE);
if (online.equals(dayData.get(dayData.size() - 1).getType())) {
totalOnlineMinutes += lastInterval;
}
onlineMinutes = (int) Math.min(totalOnlineMinutes, 1440);
}
} else {
List<PqsCommunicateDto> firstData = pqsCommunicateService.getRawDataOne(lineParam);
if (CollectionUtil.isNotEmpty(firstData)) {
Date statDate2 = DateUtil.parse(firstData.get(0).getTime());
if (statDate.before(statDate2)) {
onlineMinutes = 0;
} else {
List<PqsCommunicateDto> latestData = pqsCommunicateService.getRawDataLatest(lineParam);
if (online.equals(latestData.get(0).getType())){
onlineMinutes = 1440;
} else {
onlineMinutes = 0;
}
}
} else {
onlineMinutes = 0;
}
outCommunicateData.addAll(data);
}
Date dateOut = DateUtil.parse(time);
for (PqsCommunicateDto pqsCommunicate : outCommunicateData) {
RStatOnlineRateD po = new RStatOnlineRateD();
Date newDate = DateUtil.parse(pqsCommunicate.getTime());
lineParam.setLineId(Collections.singletonList(pqsCommunicate.getDevId()));
RStatOnlineRateD onLineRate = onLineMinute(newDate, dateOut, pqsCommunicate.getType(), lineParam);
po.setTimeId(LocalDate.parse(time, DatePattern.NORM_DATE_FORMATTER));
po.setDevIndex(pqsCommunicate.getDevId());
po.setOnlineMin(onLineRate.getOnlineMin());
po.setOfflineMin(onLineRate.getOfflineMin());
list.add(po);
}
RStatOnlineRateD po = new RStatOnlineRateD();
po.setTimeId(LocalDate.parse(time, DatePattern.NORM_DATE_FORMATTER));
po.setDevIndex(device.getId());
po.setOnlineMin(onlineMinutes);
po.setOfflineMin(1440 - onlineMinutes);
list.add(po);
}
}
if (CollectionUtil.isNotEmpty(list)) {
this.saveOrUpdateBatchByMultiId(list,1000);
this.saveOrUpdateBatchByMultiId(list, 1000);
}
}
@@ -114,97 +157,4 @@ public class RStatOnlineRateDServiceImpl extends MppServiceImpl<RStatOnlineRateD
.between(RStatOnlineRateD::getTimeId,startTime,endTime)
.list();
}
/**
* new的时间和当前统计时间 不是/是 同一天
*/
private RStatOnlineRateD onLineMinute(Date newDate, Date date, Integer type, LineCountEvaluateParam lineParam) {
RStatOnlineRateD onLineRate = new RStatOnlineRateD();
Integer minute = 0;
/*new的时间和当前统计时间是同一天*/
if (DateUtil.isSameDay(newDate, date)) {
minute = processData(newDate, date, type, lineParam);
} else {
/*new的时间和当前统计时间不是同一天*/
Date nowDate = new Date();
/*数据补招的情况下*/
if (DateUtil.between(date, nowDate, DateUnit.DAY) > DateUtil.between(newDate, nowDate, DateUnit.DAY)) {
minute = processData(newDate, date, null, lineParam);
} else {
if (online.equals(type)) {
minute = InfluxDBPublicParam.DAY_MINUTE;
}
}
}
onLineRate.setOnlineMin(minute);
onLineRate.setOfflineMin(InfluxDBPublicParam.DAY_MINUTE - minute);
return onLineRate;
}
private Integer processData(Date newDate, Date date, Integer type,LineCountEvaluateParam lineParam) {
int minute = 0;
List<PqsCommunicateDto> communicateData = pqsCommunicateService.getRawData(lineParam);
/*当前统计时间内存在多条数据*/
if (communicateData.size() > 1) {
Date lastTime = null;
long onlineTime = 0;
long offlineTime = 0;
for (int i = 0; i < communicateData.size(); i++) {
long differ;
if (i == 0) {
/*首次比较取统计时间*/
differ = DateUtil.between(date, DateUtil.parse(communicateData.get(i).getTime()), DateUnit.MINUTE);
} else {
/*后续取上一次数据时间*/
differ = DateUtil.between(lastTime, DateUtil.parse(communicateData.get(i).getTime()), DateUnit.MINUTE);
}
if (online.equals(communicateData.get(i).getType())) {
offlineTime = offlineTime + differ;
} else {
onlineTime = onlineTime + differ;
}
lastTime = DateUtil.parse(communicateData.get(i).getTime());
}
if (online.equals(communicateData.get(communicateData.size() - 1).getType())) {
minute = InfluxDBPublicParam.DAY_MINUTE - (int) offlineTime;
} else {
minute = (int) onlineTime;
}
}
/*当前统计时间内仅有一条数据*/
else {
if (type != null) {
long differ = DateUtil.between(date, newDate, DateUnit.MINUTE);
if (online.equals(type)) {
minute = InfluxDBPublicParam.DAY_MINUTE - (int) differ;
} else {
minute = (int) differ;
}
} else {
List<PqsCommunicateDto> communicateDataOld = pqsCommunicateService.getRawDataEnd(lineParam);
// if (!communicateDataOld.isEmpty()){
// if (online.equals(communicateDataOld.get(0).getType())){
// minute = InfluxDBPublicParam.DAY_MINUTE;
// }
// }
if (!communicateDataOld.isEmpty()){
try {
if (online.equals(communicateDataOld.get(0).getType())){
minute = (int) DateUtil.between(new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN).parse(communicateDataOld.get(0).getTime()), new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN).parse(lineParam.getEndTime()), DateUnit.MINUTE);
} else {
minute = (int) DateUtil.between(new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN).parse(lineParam.getStartTime()), new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN).parse(communicateDataOld.get(0).getTime()), DateUnit.MINUTE);
}
} catch (ParseException e) {
throw new RuntimeException(e);
}
} else {
minute = InfluxDBPublicParam.DAY_MINUTE;
}
}
}
return minute;
}
}