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

事件解析调整

Browse Source
This commit is contained in:
xuyang
2023-09-27 18:48:21 +08:00
parent b76aed1a56
commit 29407e4389
13 changed files with 204 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
View File

@@ -409,7 +409,6 @@ public class MqttMessageHandler {
AutoDataDto dataDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), AutoDataDto.class);
switch (dataDto.getMsg().getDataAttr()) {
//暂态事件、录波处理
//todo 后期告警可能也是在这处理,通过是告警还是事件来区分暂态和稳态
case 0:
log.info("处理事件");
log.info("事件报文为:" + new String(message.getPayload(), StandardCharsets.UTF_8));

13
iot-access/access-boot/src/main/java/com/njcn/access/listener/RedisKeyExpirationListener.java
View File

@@ -84,12 +84,13 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
record.setOfflineTime(LocalDateTime.now());
onlineLogsService.updateById(record);
//立马发起接入请求
String version = csTopicService.getVersion(nDid);
log.info("装置掉线立马发送接入请求，接入失败则进入定时接入任务");
csDeviceService.devAccess(nDid,version);
//接入再次失败，则定时发起接入请求
try {
//装置掉线等待10分钟,发起接入请求
String version = csTopicService.getVersion(nDid);
log.info("装置掉线10分钟发送接入请求，接入失败则进入定时接入任务");
Thread.sleep(600000);
csDeviceService.devAccess(nDid,version);
//接入再次失败，则定时发起接入请求
Thread.sleep(1000);
Integer status = csEquipmentDeliveryService.queryEquipmentByndid(nDid).getRunStatus();
if (!Objects.isNull(status) && Objects.equals(status,AccessEnum.OFFLINE.getCode())){
@@ -104,7 +105,7 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
//记录日志
logDto.setOperate("装置掉线,定时发送接入请求，装置为:" + nDid + ",请求的时间戳为:" + System.currentTimeMillis());
csLogsFeignClient.addUserLog(logDto);
}, 1, 3600, TimeUnit.SECONDS);
}, 1, 600, TimeUnit.SECONDS);
}
} catch (InterruptedException e) {
e.printStackTrace();

6
iot-analysis/analysis-stat/stat-boot/pom.xml
View File

@@ -65,14 +65,10 @@
<artifactId>system-api</artifactId>
<version>${project.version}</version>
</dependency>
<!--pqs-influx-->
<dependency>
<groupId>com.njcn</groupId>
<artifactId>pqs-influx</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-influxDB</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>

2
iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/StatBootApplication.java
View File

@@ -5,6 +5,7 @@ import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.DependsOn;
/**
@@ -13,6 +14,7 @@ import org.springframework.cloud.openfeign.EnableFeignClients;
* @date 2021年12月09日 20:59
*/
@Slf4j
@DependsOn("proxyMapperRegister")
@MapperScan("com.njcn.**.mapper")
@EnableFeignClients(basePackages = "com.njcn")
@SpringBootApplication(scanBasePackages = "com.njcn")

2
iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
View File

@@ -10,7 +10,7 @@ import com.njcn.csdevice.pojo.param.DataArrayParam;
import com.njcn.csdevice.pojo.po.CsDataArray;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
import com.njcn.influxdb.utils.InfluxDbUtils;
import com.njcn.influx.utils.InfluxDbUtils;
import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;

74
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/po/CsWave.java Normal file
View File

@@ -0,0 +1,74 @@
package com.njcn.zlevent.pojo.po;
import com.baomidou.mybatisplus.annotation.TableName;
import com.njcn.db.bo.BaseEntity;
import java.io.Serializable;
import java.time.LocalDateTime;
import lombok.Getter;
import lombok.Setter;
/**
* <p>
* 治理波形文件记录表
* </p>
*
* @author xuyang
* @since 2023-09-27
*/
@TableName("cs_wave")
public class CsWave {
private static final long serialVersionUID = 1L;
/**
* id
*/
private String id;
/**
* 装置ndid
*/
private String ndid;
/**
* 波形起始时间
*/
private LocalDateTime startTime;
/**
* 波形结束时间
*/
private LocalDateTime endTime;
/**
* 波形文件名称
*/
private String rcdName;
/**
* 位置信息(grid:电网侧，load:负载侧)
*/
private String location;
/**
* 文件大小
*/
private Integer fileSize;
/**
* 校验方式(crc、md5)
*/
private String checkType;
/**
* 校验码
*/
private String checkNumber;
/**
* 创建时间
*/
private LocalDateTime createTime;
}

30
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/CsWaveController.java Normal file
View File

@@ -0,0 +1,30 @@
package com.njcn.zlevent.controller;
import io.swagger.annotations.Api;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.njcn.web.controller.BaseController;
import springfox.documentation.annotations.ApiIgnore;
/**
* <p>
* 治理波形文件记录表 前端控制器
* </p>
*
* @author xuyang
* @since 2023-09-27
*/
@Slf4j
@RestController
@RequestMapping("/csWave")
@Api(tags = "波形文件")
@AllArgsConstructor
@ApiIgnore
public class CsWaveController extends BaseController {
}

4
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/WaveController.java
View File

@@ -7,7 +7,7 @@ import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.mq.message.AppEventMessage;
import com.njcn.web.controller.BaseController;
import com.njcn.zlevent.service.ICsWaveService;
import com.njcn.zlevent.service.ICsWaveAnalysisService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiOperation;
@@ -32,7 +32,7 @@ import org.springframework.web.bind.annotation.RestController;
@AllArgsConstructor
public class WaveController extends BaseController {
private final ICsWaveService csWaveService;
private final ICsWaveAnalysisService csWaveService;
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/analysis")

2
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/ICsWaveService.java → iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/ICsWaveAnalysisService.java
View File

@@ -10,7 +10,7 @@ import com.njcn.mq.message.AppEventMessage;
* @createTime 2023/9/5 14:52
*/
public interface ICsWaveService {
public interface ICsWaveAnalysisService {
/**
* 1.获取波形报文的文件名称

38
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsWaveServiceImpl.java → iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsWaveAnalysisServiceImpl.java
View File

@@ -20,7 +20,7 @@ import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.pojo.po.DictData;
import com.njcn.zlevent.pojo.constant.ZlConstant;
import com.njcn.zlevent.pojo.dto.WaveTimeDto;
import com.njcn.zlevent.service.ICsWaveService;
import com.njcn.zlevent.service.ICsWaveAnalysisService;
import lombok.AllArgsConstructor;
import net.sf.json.JSONObject;
import org.springframework.stereotype.Service;
@@ -38,7 +38,7 @@ import java.util.stream.Collectors;
*/
@Service
@AllArgsConstructor
public class CsWaveServiceImpl implements ICsWaveService {
public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
private final EquipmentFeignClient equipmentFeignClient;
@@ -115,26 +115,26 @@ public class CsWaveServiceImpl implements ICsWaveService {
* 时间处理
*/
public void channelTimeRange(String fileName, long time, long subtleTime, Double millisecond, String deviceId, String lineId, String location) {
long endTime;
// 将startTime减去8小时（8 * 3600秒）
time -= 8 * 3600;
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
time = time - 8*3600;
// 将millisecond转换为长整型，并乘以1000以获取微秒
long millisecondValue = millisecond.longValue() * 1000;
// 计算最终时间
long finalTime = subtleTime + millisecondValue;
// 如果finalTime大于等于1000000，将startTime增加1秒，finalTime减去1000000
if (finalTime >= 1000000) {
endTime = time + 1;
finalTime -= 1000000;
} else {
endTime = time;
}
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String formatTime1 = format.format(time * 1000);
String formatTime2 = format.format(endTime * 1000);
long time1 = time * 1000000 + subtleTime;
long time2 = time * 1000000 + subtleTime + millisecondValue;
String time1String = String.valueOf(time1);
String time2String = String.valueOf(time2);
String time11 = time1String.substring(time1String.length() - 6);
String time111 = time1String.substring(0,time1String.length() - 6);
String formatTime1 = format.format(Long.parseLong(time111) * 1000);
String time22 = time2String.substring(time2String.length() - 6);
String time222 = time2String.substring(0,time2String.length() - 6);
String formatTime2 = format.format(Long.parseLong(time222) * 1000);
WaveTimeDto waveTimeDto = new WaveTimeDto();
waveTimeDto.setStartTime(formatTime1 + "." + subtleTime);
waveTimeDto.setEndTime(formatTime2 + "." + finalTime);
waveTimeDto.setStartTime(formatTime1 + "." + time11);
waveTimeDto.setEndTime(formatTime2 + "." + time22);
waveTimeDto.setDeviceId(deviceId);
waveTimeDto.setLineId(lineId);
waveTimeDto.setLocation(location);

46
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java
View File

@@ -4,12 +4,10 @@ import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.CsDeviceUserFeignClient;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.csharmonic.pojo.po.CsEventPO;
import com.njcn.csharmonic.pojo.po.CsEventUserPO;
import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
import com.njcn.influx.utils.InfluxDbUtils;
import com.njcn.mq.message.AppEventMessage;
@@ -25,7 +23,6 @@ import com.njcn.zlevent.pojo.constant.ZlConstant;
import com.njcn.zlevent.pojo.po.CsEventLogs;
import com.njcn.zlevent.service.ICsEventLogsService;
import com.njcn.zlevent.service.ICsEventService;
import com.njcn.zlevent.service.ICsEventUserService;
import com.njcn.zlevent.service.IEventService;
import com.njcn.zlevent.utils.SendEventUtils;
import lombok.AllArgsConstructor;
@@ -35,7 +32,6 @@ import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.yaml.snakeyaml.scanner.Constant;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
@@ -69,10 +65,6 @@ public class EventServiceImpl implements IEventService {
private final InfluxDbUtils influxDbUtils;
private final ICsEventUserService csEventUserService;
private final CsDeviceUserFeignClient csDeviceUserFeignClient;
private final ICsEventLogsService csEventLogsService;
private final SendEventUtils sendEventUtils;
@@ -82,7 +74,6 @@ public class EventServiceImpl implements IEventService {
public void analysis(AppEventMessage appEventMessage) {
List<CsEventPO> list1 = new ArrayList<>();
List<String> records = new ArrayList<String>();
List<CsEventUserPO> list2 = new ArrayList<>();
String lineId = null,id = null,tag = null;
LocalDateTime eventTime = null;
Object object1 = redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId());
@@ -151,8 +142,6 @@ public class EventServiceImpl implements IEventService {
}
csEvent.setLineId(lineId);
list1.add(csEvent);
//事件用户关系入库
list2 = deviceUserList(deviceId,id);
//事件处理日志库
CsEventLogs csEventLogs = new CsEventLogs();
csEventLogs.setLineId(lineId);
@@ -167,15 +156,11 @@ public class EventServiceImpl implements IEventService {
if (CollectionUtil.isNotEmpty(list1)){
csEventService.saveBatch(list1);
}
//cs_device_user入库
if (CollectionUtil.isNotEmpty(list2)){
csEventUserService.saveBatch(list2);
}
//evt_data入库
if (CollectionUtil.isNotEmpty(records)) {
influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, records);
}
//推送事件逻辑处理
//推送事件逻辑处理 && cs_event_user入库
for (AppEventMessage.DataArray item : dataArray) {
sendEventUtils.sendUser(1,item.getType(),deviceId,item.getName(),eventTime,appEventMessage.getId(),id);
}
@@ -237,32 +222,19 @@ public class EventServiceImpl implements IEventService {
*/
public LocalDateTime timeFormat(Long time1, Long time2) {
String time;
//设置格式
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//todo 这边暂时先这样处理,减去8小时。
String timeText = format.format((time1-8*3600) * 1000);
time1 = time1 - 8*3600;
long t1 = time1 * 1000000 + time2;
String time1String = String.valueOf(t1);
String time11 = time1String.substring(time1String.length() - 6);
String time111 = time1String.substring(0,time1String.length() - 6);
String formatTime1 = format.format(Long.parseLong(time111) * 1000);
if (time2 == 0){
time = timeText + ".000000";
time = formatTime1 + ".000000";
} else {
time = timeText + "." + time2;
time = formatTime1 + "." + time11;
}
DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
return LocalDateTime.parse(time, fmt);
}
/**
* 获取用户设备关系
*/
public List<CsEventUserPO> deviceUserList(String devId, String id) {
List<CsEventUserPO> result = new ArrayList<>();
List<String> list = csDeviceUserFeignClient.findUserById(devId).getData();
list.forEach(item->{
CsEventUserPO csEventUser = new CsEventUserPO();
csEventUser.setUserId(item);
csEventUser.setStatus(0);
csEventUser.setEventId(id);
result.add(csEventUser);
});
return result;
}
}

3
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/FileServiceImpl.java
View File

@@ -148,6 +148,7 @@ public class FileServiceImpl implements IFileService {
}
csEventLogs.setStatus(1);
csEventLogs.setRemark("当前文件1帧,全部收到,解析成功!");
//todo 记录文件信息
} else {
//缓存文件
map.put(appFileMessage.getMid(),appFileMessage.getMsg().getData());
@@ -174,6 +175,8 @@ public class FileServiceImpl implements IFileService {
}
csEventLogs.setStatus(1);
csEventLogs.setRemark("当前文件"+l1.size()+"帧,这是第"+l1.size()+"帧,全部收到,解析成功!");
//todo 记录文件信息
} else {
//缓存
fileStreamDto = new FileStreamDto();

58
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/utils/SendEventUtils.java
View File

@@ -5,13 +5,15 @@ import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.njcn.csdevice.api.CsDeviceUserFeignClient;
import com.njcn.csdevice.api.EventLogsFeignClient;
import com.njcn.csdevice.pojo.po.CsEventSendMsg;
import com.njcn.csharmonic.pojo.po.CsEventUserPO;
import com.njcn.system.api.EpdFeignClient;
import com.njcn.user.api.AppInfoSetFeignClient;
import com.njcn.user.api.AppUserFeignClient;
import com.njcn.user.api.UserFeignClient;
import com.njcn.user.pojo.po.User;
import com.njcn.user.pojo.po.app.AppInfoSet;
import com.njcn.zlevent.pojo.dto.NoticeUserDto;
import lombok.AllArgsConstructor;
import com.njcn.zlevent.service.ICsEventUserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
@@ -45,6 +47,9 @@ public class SendEventUtils {
@Resource
private UserFeignClient userFeignClient;
@Resource
private AppUserFeignClient appUserFeignClient;
@Resource
private CsDeviceUserFeignClient csDeviceUserFeignClient;
@@ -57,6 +62,9 @@ public class SendEventUtils {
@Resource
private EpdFeignClient epdFeignClient;
@Resource
private ICsEventUserService csEventUserService;
/**
* 事件推送给相关用户
* @param eventType 事件类型 1:事件 2:告警
@@ -76,14 +84,21 @@ public class SendEventUtils {
NoticeUserDto.Payload payload = new NoticeUserDto.Payload();
String content;
eventName = epdFeignClient.findByName(eventName).getData().getShowName();
List<CsEventUserPO> result = new ArrayList<>();
//事件处理
if (eventType == 1){
switch (type) {
case "1":
//设备自身事件 不推送给用户，推送给业务管理
users = userFeignClient.getAdminInfo().getData();
users = appUserFeignClient.getAdminInfo().getData();
noticeUserDto.setPushClientId(Collections.singletonList(users.get(0).getDevCode()));
noticeUserDto.setTitle("设备事件");
//记录需要通知的用户和事件关系
CsEventUserPO csEventUser = new CsEventUserPO();
csEventUser.setUserId(users.get(0).getId());
csEventUser.setStatus(0);
csEventUser.setEventId(id);
result.add(csEventUser);
break;
case "2":
//暂态事件
@@ -91,6 +106,15 @@ public class SendEventUtils {
devCodeList = users.stream().map(User::getDevCode).collect(Collectors.toList());
noticeUserDto.setPushClientId(devCodeList);
noticeUserDto.setTitle("暂态事件");
//记录需要通知的用户和事件关系
List<String> list2 = users.stream().map(User::getId).collect(Collectors.toList());
list2.forEach(item->{
CsEventUserPO csEventUser2 = new CsEventUserPO();
csEventUser2.setUserId(item);
csEventUser2.setStatus(0);
csEventUser2.setEventId(id);
result.add(csEventUser2);
});
break;
case "3":
//稳态事件
@@ -98,6 +122,15 @@ public class SendEventUtils {
devCodeList = users.stream().map(User::getDevCode).collect(Collectors.toList());
noticeUserDto.setPushClientId(devCodeList);
noticeUserDto.setTitle("稳态事件");
//记录需要通知的用户和事件关系
List<String> list3 = users.stream().map(User::getId).collect(Collectors.toList());
list3.forEach(item->{
CsEventUserPO csEventUser3 = new CsEventUserPO();
csEventUser3.setUserId(item);
csEventUser3.setStatus(0);
csEventUser3.setEventId(id);
result.add(csEventUser3);
});
break;
default:
break;
@@ -112,8 +145,14 @@ public class SendEventUtils {
switch (type) {
case "1":
//Ⅰ级告警 不推送给用户，推送给业务管理
users = userFeignClient.getAdminInfo().getData();
users = appUserFeignClient.getAdminInfo().getData();
noticeUserDto.setPushClientId(Collections.singletonList(users.get(0).getDevCode()));
//记录需要通知的用户和事件关系
CsEventUserPO csEventUser = new CsEventUserPO();
csEventUser.setUserId(users.get(0).getId());
csEventUser.setStatus(0);
csEventUser.setEventId(id);
result.add(csEventUser);
break;
case "2":
case "3":
@@ -121,6 +160,15 @@ public class SendEventUtils {
users = getEventUser(devId);
devCodeList = users.stream().map(User::getDevCode).collect(Collectors.toList());
noticeUserDto.setPushClientId(devCodeList);
//记录需要通知的用户和事件关系
List<String> list4 = users.stream().map(User::getId).collect(Collectors.toList());
list4.forEach(item->{
CsEventUserPO csEventUser4 = new CsEventUserPO();
csEventUser4.setUserId(item);
csEventUser4.setStatus(0);
csEventUser4.setEventId(id);
result.add(csEventUser4);
});
break;
default:
break;
@@ -148,6 +196,10 @@ public class SendEventUtils {
csEventSendMsgList.add(csEventSendMsg);
}
eventLogsFeignClient.addLogs(csEventSendMsgList);
//事件用户关系入库
if (CollectionUtil.isNotEmpty(result)){
csEventUserService.saveBatch(result);
}
}
/**