feat: 设备遥测实时采集与报警推送

liu

6 天以前 722f3d8cfd48e3db2836578b36cff648dee1f72a

 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("设备控制网关调用失败，deviceId={}, 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();