From 722f3d8cfd48e3db2836578b36cff648dee1f72a Mon Sep 17 00:00:00 2001
From: liu <2021943741@qq.com>
Date: 星期二, 03 十一月 2026 13:22:09 +0800
Subject: [PATCH] feat: 设备遥测实时采集与报警推送
---
src/main/java/com/ruoyi/unattended/telemetry/service/impl/DeviceTelemetryServiceImpl.java | 160 ++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 158 insertions(+), 2 deletions(-)
diff --git a/src/main/java/com/ruoyi/unattended/telemetry/service/impl/DeviceTelemetryServiceImpl.java b/src/main/java/com/ruoyi/unattended/telemetry/service/impl/DeviceTelemetryServiceImpl.java
index d3e36cf..4b5835b 100644
--- a/src/main/java/com/ruoyi/unattended/telemetry/service/impl/DeviceTelemetryServiceImpl.java
+++ b/src/main/java/com/ruoyi/unattended/telemetry/service/impl/DeviceTelemetryServiceImpl.java
@@ -7,8 +7,16 @@
import com.ruoyi.unattended.telemetry.dto.DeviceMonitorVo;
import com.ruoyi.unattended.telemetry.mapper.DeviceTelemetryMapper;
import com.ruoyi.unattended.telemetry.pojo.DeviceTelemetry;
+import com.ruoyi.unattended.telemetry.service.DeviceAlarmPushService;
import com.ruoyi.unattended.telemetry.service.DeviceTelemetryService;
import lombok.extern.slf4j.Slf4j;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
@@ -16,7 +24,9 @@
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.*;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import com.alibaba.fastjson2.JSON;
/**
* 璁惧閬ユ祴鏈嶅姟銆�
@@ -36,6 +46,24 @@
@Resource
private DeviceTelemetryMapper telemetryMapper;
+ @Resource
+ private DeviceAlarmPushService alarmPushService;
+
+ @Value("${unattended.device-control.gateway-url:}")
+ private String gatewayUrl;
+
+ @Value("${unattended.device-control.auth-token:}")
+ private String gatewayToken;
+
+ @Value("${unattended.telemetry.auto-collect-enabled:true}")
+ private boolean autoCollectEnabled;
+
+ private final OkHttpClient controlClient = new OkHttpClient.Builder()
+ .connectTimeout(5, TimeUnit.SECONDS)
+ .readTimeout(8, TimeUnit.SECONDS)
+ .writeTimeout(8, TimeUnit.SECONDS)
+ .build();
+
private static final BigDecimal DEFAULT_TEMP_MIN = new BigDecimal("20");
private static final BigDecimal DEFAULT_TEMP_MAX = new BigDecimal("80");
private static final BigDecimal DEFAULT_PRESSURE_MIN = new BigDecimal("0.20");
@@ -53,6 +81,7 @@
}
LocalDateTime now = LocalDateTime.now();
List<DeviceTelemetry> batch = new ArrayList<>();
+ Set<Long> alarmDeviceIds = new HashSet<>();
for (DeviceLedger device : devices) {
DeviceTelemetry last = telemetryMapper.selectOne(
Wrappers.<DeviceTelemetry>lambdaQuery()
@@ -65,11 +94,87 @@
telemetry.setDeviceName(device.getDeviceName());
judge(device, telemetry);
batch.add(telemetry);
+ if (shouldPush(last, telemetry)) {
+ alarmDeviceIds.add(device.getId());
+ }
}
if (!batch.isEmpty()) {
saveBatch(batch);
}
- return monitorList();
+ List<DeviceMonitorVo> result = monitorList();
+ publishAlarms(result, alarmDeviceIds);
+ return result;
+ }
+
+ /**
+ * 榛樿姣� 30 绉掗噰闆嗕竴娆°�傜幇鍦烘帴鍏ョ湡瀹炵綉鍏冲悗鍙皢 auto-collect-enabled 璁句负 false锛�
+ * 鏀圭敱 /report 鎺ュ彛鎺ユ敹 PLC/杈圭紭缃戝叧鏁版嵁銆�
+ */
+ @Scheduled(fixedDelayString = "${unattended.telemetry.collect-interval-ms:30000}",
+ initialDelayString = "${unattended.telemetry.initial-delay-ms:10000}")
+ public void scheduledCollect() {
+ if (!autoCollectEnabled) {
+ return;
+ }
+ try {
+ collectOnce();
+ } catch (Exception e) {
+ log.error("璁惧閬ユ祴瀹氭椂閲囬泦澶辫触", e);
+ }
+ }
+
+ @Override
+ public DeviceMonitorVo report(DeviceTelemetry telemetry) {
+ if (telemetry == null || telemetry.getDeviceLedgerId() == null) {
+ throw new IllegalArgumentException("璁惧id涓嶈兘涓虹┖");
+ }
+ DeviceLedger device = deviceLedgerMapper.selectById(telemetry.getDeviceLedgerId());
+ if (device == null) {
+ throw new IllegalArgumentException("璁惧涓嶅瓨鍦�");
+ }
+ DeviceTelemetry last = latestTelemetry(device.getId());
+ telemetry.setId(null);
+ telemetry.setDeviceName(device.getDeviceName());
+ if (telemetry.getCollectTime() == null) {
+ telemetry.setCollectTime(LocalDateTime.now());
+ }
+ judge(device, telemetry);
+ save(telemetry);
+ DeviceMonitorVo result = monitorList().stream()
+ .filter(item -> device.getId().equals(item.getDeviceId()))
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("璁惧鐩戞帶鏁版嵁鐢熸垚澶辫触"));
+ if (shouldPush(last, telemetry)) {
+ alarmPushService.publish(result);
+ }
+ return result;
+ }
+
+ private DeviceTelemetry latestTelemetry(Long deviceId) {
+ return telemetryMapper.selectOne(
+ Wrappers.<DeviceTelemetry>lambdaQuery()
+ .eq(DeviceTelemetry::getDeviceLedgerId, deviceId)
+ .orderByDesc(DeviceTelemetry::getCollectTime)
+ .last("limit 1"));
+ }
+
+ private boolean shouldPush(DeviceTelemetry previous, DeviceTelemetry current) {
+ int currentLevel = alarmLevel(current == null ? null : current.getStatus());
+ int previousLevel = alarmLevel(previous == null ? null : previous.getStatus());
+ return currentLevel > 0 && (previousLevel == 0 || currentLevel > previousLevel);
+ }
+
+ private int alarmLevel(String status) {
+ if ("alarm".equalsIgnoreCase(status)) return 2;
+ if ("warning".equalsIgnoreCase(status)) return 1;
+ return 0;
+ }
+
+ private void publishAlarms(List<DeviceMonitorVo> devices, Set<Long> alarmDeviceIds) {
+ if (alarmDeviceIds.isEmpty()) return;
+ devices.stream()
+ .filter(item -> alarmDeviceIds.contains(item.getDeviceId()))
+ .forEach(alarmPushService::publish);
}
/**
@@ -219,6 +324,7 @@
} else {
throw new IllegalArgumentException("涓嶆敮鎸佺殑鎿嶄綔:" + action);
}
+ Map<String, Object> dispatch = dispatchControl(device, action);
DeviceLedger update = new DeviceLedger();
update.setId(deviceId);
update.setControlStatus(status);
@@ -229,11 +335,61 @@
result.put("deviceName", device.getDeviceName());
result.put("controlStatus", status);
result.put("commandTime", LocalDateTime.now().toString());
- result.put("message", "running".equals(status) ? "鍚姩鎸囦护宸蹭笅鍙�" : "鍋滄鎸囦护宸蹭笅鍙�");
+ result.putAll(dispatch);
+ boolean localMode = "local".equals(dispatch.get("dispatchMode"));
+ if (localMode) {
+ result.put("message", "鏈湴鑱旇皟鐘舵�佸凡鏇存柊");
+ } else {
+ result.put("message", "running".equals(status) ? "璁惧鍚姩鎸囦护宸茬‘璁�" : "璁惧鍋滄鎸囦护宸茬‘璁�");
+ }
log.info("[杩滅▼鎺у埗] deviceId={}, action={}, status={}", deviceId, action, status);
return result;
}
+ /**
+ * 璋冪敤缁熶竴璁惧鎺у埗缃戝叧銆�
+ * 鏈厤缃綉鍏虫椂浣跨敤鏈湴鑱旇皟妯″紡锛屼娇绯荤粺鍙湪纭欢鍒板満鍓嶅畬鎴愪笟鍔￠獙鏀躲��
+ */
+ private Map<String, Object> dispatchControl(DeviceLedger device, String action) {
+ String endpoint = trimToNull(gatewayUrl);
+ Map<String, Object> result = new LinkedHashMap<>();
+ if (endpoint == null) {
+ result.put("dispatchMode", "local");
+ result.put("gatewayAcknowledged", true);
+ return result;
+ }
+
+ Map<String, Object> command = new LinkedHashMap<>();
+ command.put("deviceId", device.getId());
+ command.put("deviceName", device.getDeviceName());
+ command.put("action", action.toLowerCase(Locale.ROOT));
+ command.put("commandTime", LocalDateTime.now().toString());
+ RequestBody body = RequestBody.create(JSON.toJSONString(command),
+ MediaType.parse("application/json; charset=utf-8"));
+ Request.Builder builder = new Request.Builder().url(endpoint).post(body);
+ if (trimToNull(gatewayToken) != null) {
+ builder.header("Authorization", "Bearer " + gatewayToken.trim());
+ }
+ try (Response response = controlClient.newCall(builder.build()).execute()) {
+ if (!response.isSuccessful()) {
+ throw new IllegalArgumentException("璁惧鎺у埗缃戝叧鎷掔粷鎸囦护锛岀姸鎬佺爜锛�" + response.code());
+ }
+ result.put("dispatchMode", "gateway");
+ result.put("gatewayAcknowledged", true);
+ result.put("gatewayStatus", response.code());
+ return result;
+ } catch (IllegalArgumentException e) {
+ throw e;
+ } catch (Exception e) {
+ log.error("璁惧鎺у埗缃戝叧璋冪敤澶辫触锛宒eviceId={}, endpoint={}", device.getId(), endpoint, e);
+ throw new IllegalArgumentException("璁惧鎺у埗缃戝叧杩炴帴澶辫触锛岃妫�鏌ヨ澶囩綉缁�");
+ }
+ }
+
+ private String trimToNull(String value) {
+ return value == null || value.trim().isEmpty() ? null : value.trim();
+ }
+
@Override
public List<DeviceMonitorVo> alarms() {
List<DeviceMonitorVo> all = monitorList();
--
Gitblit v1.9.3