package com.ruoyi.unattended.telemetry.service.impl;
|
|
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
import com.ruoyi.device.mapper.DeviceLedgerMapper;
|
import com.ruoyi.device.pojo.DeviceLedger;
|
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;
|
import java.math.BigDecimal;
|
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;
|
|
/**
|
* 设备遥测服务。
|
*
|
* 说明：现场传感器通过 PLC/网关接入，本服务提供统一的采集入口与阈值判定、
|
* 报警与远程控制能力。collectOnce 会基于设备最近一次数据做平滑推演，
|
* 接入真实网关后只需替换 sample() 的数据来源即可，其余业务逻辑保持不变。
|
*/
|
@Slf4j
|
@Service
|
public class DeviceTelemetryServiceImpl extends ServiceImpl<DeviceTelemetryMapper, DeviceTelemetry>
|
implements DeviceTelemetryService {
|
|
@Resource
|
private DeviceLedgerMapper deviceLedgerMapper;
|
|
@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");
|
private static final BigDecimal DEFAULT_PRESSURE_MAX = new BigDecimal("1.50");
|
private static final BigDecimal DEFAULT_VIBRATION_MAX = new BigDecimal("5.00");
|
|
@Override
|
public List<DeviceMonitorVo> collectOnce() {
|
List<DeviceLedger> devices = deviceLedgerMapper.selectList(
|
Wrappers.<DeviceLedger>lambdaQuery()
|
.and(w -> w.eq(DeviceLedger::getTelemetryEnabled, 1).or().isNull(DeviceLedger::getTelemetryEnabled)));
|
if (devices == null || devices.isEmpty()) {
|
// 若台账未配置遥测开关，则回退为全部设备
|
devices = deviceLedgerMapper.selectList(null);
|
}
|
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()
|
.eq(DeviceTelemetry::getDeviceLedgerId, device.getId())
|
.orderByDesc(DeviceTelemetry::getCollectTime)
|
.last("limit 1"));
|
DeviceTelemetry telemetry = sample(device, last);
|
telemetry.setCollectTime(now);
|
telemetry.setDeviceLedgerId(device.getId());
|
telemetry.setDeviceName(device.getDeviceName());
|
judge(device, telemetry);
|
batch.add(telemetry);
|
if (shouldPush(last, telemetry)) {
|
alarmDeviceIds.add(device.getId());
|
}
|
}
|
if (!batch.isEmpty()) {
|
saveBatch(batch);
|
}
|
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);
|
}
|
|
/**
|
* 采样：接入真实网关时替换此处数据来源（Modbus/MQTT/OPC-UA）
|
*/
|
private DeviceTelemetry sample(DeviceLedger device, DeviceTelemetry last) {
|
DeviceTelemetry t = new DeviceTelemetry();
|
Random random = new Random();
|
boolean running = !"stopped".equalsIgnoreCase(device.getControlStatus());
|
if (!running) {
|
t.setTemperature(scale(22 + random.nextDouble() * 3));
|
t.setPressure(scale(0.3 + random.nextDouble() * 0.1));
|
t.setVibration(scale(random.nextDouble() * 0.5));
|
t.setEnergy(BigDecimal.ZERO);
|
return t;
|
}
|
double baseTemp = last != null && last.getTemperature() != null
|
? last.getTemperature().doubleValue() : 45 + random.nextDouble() * 10;
|
double basePress = last != null && last.getPressure() != null
|
? last.getPressure().doubleValue() : 0.6 + random.nextDouble() * 0.3;
|
double baseVib = last != null && last.getVibration() != null
|
? last.getVibration().doubleValue() : 1.2 + random.nextDouble() * 0.8;
|
|
double temp = clamp(baseTemp + (random.nextDouble() - 0.5) * 1.6, 15, 95);
|
double press = clamp(basePress + (random.nextDouble() - 0.5) * 0.08, 0.1, 1.8);
|
double vib = clamp(baseVib + (random.nextDouble() - 0.5) * 0.25, 0.1, 8);
|
// 偶发越界，用于演示报警链路
|
if (random.nextDouble() < 0.04) temp = temp + 30;
|
if (random.nextDouble() < 0.03) press = press + 0.9;
|
if (random.nextDouble() < 0.03) vib = vib + 4;
|
|
t.setTemperature(scale(temp));
|
t.setPressure(scale(press));
|
t.setVibration(scale(vib));
|
t.setEnergy(scale(8 + random.nextDouble() * 6));
|
return t;
|
}
|
|
/**
|
* 阈值判定：越界 -> warning，严重越界 -> alarm
|
*/
|
private void judge(DeviceLedger device, DeviceTelemetry t) {
|
BigDecimal tempMin = nvl(device.getTempMin(), DEFAULT_TEMP_MIN);
|
BigDecimal tempMax = nvl(device.getTempMax(), DEFAULT_TEMP_MAX);
|
BigDecimal pressMin = nvl(device.getPressureMin(), DEFAULT_PRESSURE_MIN);
|
BigDecimal pressMax = nvl(device.getPressureMax(), DEFAULT_PRESSURE_MAX);
|
BigDecimal vibMax = nvl(device.getVibrationMax(), DEFAULT_VIBRATION_MAX);
|
|
List<String> alarms = new ArrayList<>();
|
String status = "normal";
|
|
if (t.getTemperature() != null) {
|
if (t.getTemperature().compareTo(tempMax) > 0 || t.getTemperature().compareTo(tempMin) < 0) {
|
alarms.add("温度越界:" + t.getTemperature() + "℃ (正常 " + tempMin + "~" + tempMax + ")");
|
status = "warning";
|
}
|
}
|
if (t.getPressure() != null) {
|
if (t.getPressure().compareTo(pressMax) > 0 || t.getPressure().compareTo(pressMin) < 0) {
|
alarms.add("压力越界:" + t.getPressure() + "MPa (正常 " + pressMin + "~" + pressMax + ")");
|
status = "warning";
|
}
|
}
|
if (t.getVibration() != null && t.getVibration().compareTo(vibMax) > 0) {
|
alarms.add("振动越界:" + t.getVibration() + "mm/s (上限 " + vibMax + ")");
|
status = "alarm";
|
}
|
// 多项同时越界升级为 alarm
|
if (alarms.size() >= 2) {
|
status = "alarm";
|
}
|
t.setStatus(status);
|
t.setAlarmContent(alarms.isEmpty() ? null : String.join("；", alarms));
|
}
|
|
@Override
|
public List<DeviceMonitorVo> monitorList() {
|
List<DeviceLedger> devices = deviceLedgerMapper.selectList(null);
|
if (devices == null || devices.isEmpty()) {
|
return Collections.emptyList();
|
}
|
List<DeviceMonitorVo> result = new ArrayList<>();
|
for (DeviceLedger device : devices) {
|
List<DeviceTelemetry> recent = telemetryMapper.selectList(
|
Wrappers.<DeviceTelemetry>lambdaQuery()
|
.eq(DeviceTelemetry::getDeviceLedgerId, device.getId())
|
.orderByDesc(DeviceTelemetry::getCollectTime)
|
.last("limit 30"));
|
DeviceMonitorVo vo = new DeviceMonitorVo();
|
vo.setDeviceId(device.getId());
|
vo.setDeviceName(device.getDeviceName());
|
vo.setDeviceModel(device.getDeviceModel());
|
vo.setControlStatus(device.getControlStatus() == null ? "running" : device.getControlStatus());
|
vo.setThreshold(buildThreshold(device));
|
if (!recent.isEmpty()) {
|
DeviceTelemetry latest = recent.get(0);
|
vo.setTemperature(latest.getTemperature());
|
vo.setPressure(latest.getPressure());
|
vo.setVibration(latest.getVibration());
|
vo.setEnergy(latest.getEnergy());
|
vo.setCollectTime(latest.getCollectTime());
|
vo.setStatus(latest.getStatus());
|
vo.setAlarmContent(latest.getAlarmContent());
|
} else {
|
vo.setStatus("offline");
|
}
|
Collections.reverse(recent);
|
vo.setSeries(recent);
|
result.add(vo);
|
}
|
return result;
|
}
|
|
private DeviceMonitorVo.Threshold buildThreshold(DeviceLedger device) {
|
DeviceMonitorVo.Threshold th = new DeviceMonitorVo.Threshold();
|
th.setTempMin(nvl(device.getTempMin(), DEFAULT_TEMP_MIN));
|
th.setTempMax(nvl(device.getTempMax(), DEFAULT_TEMP_MAX));
|
th.setPressureMin(nvl(device.getPressureMin(), DEFAULT_PRESSURE_MIN));
|
th.setPressureMax(nvl(device.getPressureMax(), DEFAULT_PRESSURE_MAX));
|
th.setVibrationMax(nvl(device.getVibrationMax(), DEFAULT_VIBRATION_MAX));
|
return th;
|
}
|
|
@Override
|
public List<DeviceTelemetry> history(Long deviceId, Integer limit) {
|
int size = limit == null || limit <= 0 ? 100 : Math.min(limit, 1000);
|
List<DeviceTelemetry> list = telemetryMapper.selectList(
|
Wrappers.<DeviceTelemetry>lambdaQuery()
|
.eq(deviceId != null, DeviceTelemetry::getDeviceLedgerId, deviceId)
|
.orderByDesc(DeviceTelemetry::getCollectTime)
|
.last("limit " + size));
|
Collections.reverse(list);
|
return list;
|
}
|
|
@Override
|
public Map<String, Object> control(Long deviceId, String action) {
|
DeviceLedger device = deviceLedgerMapper.selectById(deviceId);
|
if (device == null) {
|
throw new IllegalArgumentException("设备不存在");
|
}
|
String status;
|
if ("start".equalsIgnoreCase(action)) {
|
status = "running";
|
} else if ("stop".equalsIgnoreCase(action)) {
|
status = "stopped";
|
} else {
|
throw new IllegalArgumentException("不支持的操作:" + action);
|
}
|
Map<String, Object> dispatch = dispatchControl(device, action);
|
DeviceLedger update = new DeviceLedger();
|
update.setId(deviceId);
|
update.setControlStatus(status);
|
deviceLedgerMapper.updateById(update);
|
|
Map<String, Object> result = new LinkedHashMap<>();
|
result.put("deviceId", deviceId);
|
result.put("deviceName", device.getDeviceName());
|
result.put("controlStatus", status);
|
result.put("commandTime", LocalDateTime.now().toString());
|
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();
|
return all.stream()
|
.filter(v -> "warning".equals(v.getStatus()) || "alarm".equals(v.getStatus()))
|
.sorted((a, b) -> {
|
int ra = "alarm".equals(a.getStatus()) ? 0 : 1;
|
int rb = "alarm".equals(b.getStatus()) ? 0 : 1;
|
if (ra != rb) return ra - rb;
|
if (a.getCollectTime() == null) return 1;
|
if (b.getCollectTime() == null) return -1;
|
return b.getCollectTime().compareTo(a.getCollectTime());
|
})
|
.collect(Collectors.toList());
|
}
|
|
private static BigDecimal scale(double v) {
|
return BigDecimal.valueOf(v).setScale(2, RoundingMode.HALF_UP);
|
}
|
|
private static double clamp(double v, double min, double max) {
|
return Math.max(min, Math.min(max, v));
|
}
|
|
private static BigDecimal nvl(BigDecimal value, BigDecimal fallback) {
|
return value == null ? fallback : value;
|
}
|
}
|
