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.DeviceTelemetryService;
|
import lombok.extern.slf4j.Slf4j;
|
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.stream.Collectors;
|
|
/**
|
* 设备遥测服务。
|
*
|
* 说明：现场传感器通过 PLC/网关接入，本服务提供统一的采集入口与阈值判定、
|
* 报警与远程控制能力。collectOnce 会基于设备最近一次数据做平滑推演，
|
* 接入真实网关后只需替换 sample() 的数据来源即可，其余业务逻辑保持不变。
|
*/
|
@Slf4j
|
@Service
|
public class DeviceTelemetryServiceImpl extends ServiceImpl<DeviceTelemetryMapper, DeviceTelemetry>
|
implements DeviceTelemetryService {
|
|
@Resource
|
private DeviceLedgerMapper deviceLedgerMapper;
|
|
@Resource
|
private DeviceTelemetryMapper telemetryMapper;
|
|
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<>();
|
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 (!batch.isEmpty()) {
|
saveBatch(batch);
|
}
|
return monitorList();
|
}
|
|
/**
|
* 采样：接入真实网关时替换此处数据来源（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);
|
}
|
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.put("message", "running".equals(status) ? "启动指令已下发" : "停止指令已下发");
|
log.info("[远程控制] deviceId={}, action={}, status={}", deviceId, action, status);
|
return result;
|
}
|
|
@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;
|
}
|
}
|
