From a68a670ecbf26ece1fa7f528cad02ea87f90b642 Mon Sep 17 00:00:00 2001 From: rqian <1206436827@qq.com> Date: Wed, 17 Jun 2026 09:14:42 +0800 Subject: [PATCH] update --- .../src/main/resources/application-dev.yml | 23 +- .../java/com/maibu/core/business/Device.java | 7 +- .../com/maibu/core/business/WorkRecord.java | 3 +- .../business/device}/DeviceStatusDetail.java | 2 +- .../core/business/device/NettyDevice.java | 8 +- .../java/com/maibu/influxdb/DeviceData.java | 29 -- .../com/maibu/influxdb/MowerRealTimeData.java | 155 ++++++++ .../main/java/com/maibu/influxdb/test.java | 65 ---- .../com/maibu/influxdb/util/InfluxDBUtil.java | 22 +- .../maibu/influxdb/util/InfluxSqlBuilder.java | 126 ++++-- .../com/maibu/influxdb/util/LambdaUtils.java | 14 + .../java/com/maibu/memory/GlobalMemory.java | 33 +- .../java/com/maibu/memory/SiteMemory.java | 25 +- .../java/com/maibu/mqtt/MqttClientUtil.java | 135 +++++++ .../main/java/com/maibu/mqtt/MqttTopic.java | 10 + .../com/maibu/dto/DeviceTaskQueryDTO.java | 9 +- .../com/maibu/dto/WebStatusMessageDTO.java | 1 + .../main/java/com/maibu/init/InitThread.java | 111 ++++-- .../netty/handler/DataToDataBaseHandler.java | 361 ++++++++++++++---- .../netty/handler/DeviceConnectHandler.java | 43 +-- .../com/maibu/scheduled/ScheduledTask.java | 1 + .../com/maibu/service/DeviceTaskService.java | 54 ++- .../maibu/service/TransferDeviceService.java | 14 +- .../controller/WorkRecordController.java | 34 +- .../maibu/service/impl/DeviceServiceImpl.java | 100 ++--- .../maibu/service/impl/WorkRecordService.java | 43 ++- .../maibu/netty/handler/ClientHandler.java | 17 +- 27 files changed, 1053 insertions(+), 392 deletions(-) rename {maibu-web-middleware/src/main/java/com/maibu/dto => maibu-common/src/main/java/com/maibu/core/business/device}/DeviceStatusDetail.java (77%) delete mode 100644 maibu-common/src/main/java/com/maibu/influxdb/DeviceData.java create mode 100644 maibu-common/src/main/java/com/maibu/influxdb/MowerRealTimeData.java delete mode 100644 maibu-common/src/main/java/com/maibu/influxdb/test.java create mode 100644 maibu-common/src/main/java/com/maibu/mqtt/MqttClientUtil.java create mode 100644 maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java rename {maibu-web-middleware => maibu-netty-server}/src/main/java/com/maibu/dto/WebStatusMessageDTO.java (79%) diff --git a/maibu-admin/src/main/resources/application-dev.yml b/maibu-admin/src/main/resources/application-dev.yml index 3c728ac..c51c9d8 100644 --- a/maibu-admin/src/main/resources/application-dev.yml +++ b/maibu-admin/src/main/resources/application-dev.yml @@ -20,22 +20,24 @@ spring: max-idle: 8 # 连接池中的最大空闲连接 max-active: 8 # 连接池的最大数据库连接数 max-wait: -1ms # 连接池最大阻塞等待时间（使用负值表示没有限制） - # mqtt 配置 - mqtt: - username: fastbee # 账号 - password: fastbee # 密码 - host-url: tcp://localhost:1883 # mqtt连接tcp地址 - client-id: ${random.int} # 客户端Id，不能相同，采用随机数 ${random.value} - default-topic: test # 默认主题 - timeout: 30 # 超时时间 - keepalive: 30 # 保持连接 - clearSession: true # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息) servlet: multipart: enabled: true max-file-size: 10MB max-request-size: 10MB + + # mqtt 配置 +mqtt: + username: maibu # 账号 + password: jsmbzn520 # 密码 + host-url: tcp://localhost:1883 # mqtt连接tcp地址 + client-id: maibu-mqtt-client # 客户端Id，不能相同，采用随机数 ${random.value} + default-topic: test # 默认主题 + timeout: 30 # 超时时间 + keepalive: 30 # 保持连接 + clearSession: true # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息) + logging: level: com.maibu: debug @@ -81,3 +83,4 @@ influxdb: bucket: device_data org: maibu token: maibu-token + url: http://localhost:8086 diff --git a/maibu-common/src/main/java/com/maibu/core/business/Device.java b/maibu-common/src/main/java/com/maibu/core/business/Device.java index 8e5e495..1eb3d74 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/Device.java +++ b/maibu-common/src/main/java/com/maibu/core/business/Device.java @@ -5,6 +5,8 @@ import com.fasterxml.jackson.annotation.JsonFormat; import com.fasterxml.jackson.annotation.JsonInclude; import com.maibu.annotation.Excel; import com.maibu.core.domain.BaseDO; +import com.maibu.influxdb.MowerRealTimeData; + import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; import lombok.Data; @@ -220,9 +222,8 @@ public class Device extends BaseDO { @TableField(exist = false) private DeviceRunningStatusHistory lastRunningStatus; - public DeviceRunningStatusHistory getLastRunningStatus() { - return lastRunningStatus != null ? lastRunningStatus : new DeviceRunningStatusHistory(); - } + @TableField(exist = false) + private MowerRealTimeData lastMowerRealTimeData; @TableField(exist = false) private Map statusStatistics; diff --git a/maibu-common/src/main/java/com/maibu/core/business/WorkRecord.java b/maibu-common/src/main/java/com/maibu/core/business/WorkRecord.java index 46dbfa5..bca8cfe 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/WorkRecord.java +++ b/maibu-common/src/main/java/com/maibu/core/business/WorkRecord.java @@ -6,11 +6,10 @@ import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler; import com.maibu.core.domain.OrgBaseDO; + import lombok.Data; import lombok.EqualsAndHashCode; -import java.io.Serializable; - @Data @EqualsAndHashCode(callSuper = true) @TableName(value = "iot_work_record", autoResultMap = true) diff --git a/maibu-web-middleware/src/main/java/com/maibu/dto/DeviceStatusDetail.java b/maibu-common/src/main/java/com/maibu/core/business/device/DeviceStatusDetail.java similarity index 77% rename from maibu-web-middleware/src/main/java/com/maibu/dto/DeviceStatusDetail.java rename to maibu-common/src/main/java/com/maibu/core/business/device/DeviceStatusDetail.java index 5a87e7d..500d953 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/dto/DeviceStatusDetail.java +++ b/maibu-common/src/main/java/com/maibu/core/business/device/DeviceStatusDetail.java @@ -1,4 +1,4 @@ -package com.maibu.dto; +package com.maibu.core.business.device; import lombok.Data; diff --git a/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java b/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java index 4a6e132..f33cd73 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java +++ b/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java @@ -98,6 +98,8 @@ public class NettyDevice extends Connector { siteMemory.removeDevicePlanExecute(task.getDeviceId()); task = null; currentTaskId = null; + currentPoint = null; + locationQueue.clear(); executingTask = false; return finishId; } @@ -113,6 +115,7 @@ public class NettyDevice extends Connector { siteMemory.removeDevicePlanExecute(task.getDeviceId()); task = null; currentTaskId = null; + currentPoint = null; locationQueue.clear(); executingTask = false; } @@ -152,8 +155,9 @@ public class NettyDevice extends Connector { task.setTaskStaus(DeviceTaskStaus.EXECUTING); executingTask = true; sendPathCommand((byte) 0x01, (short) 3, 0.00d, 0.00d, (short) 0); - Thread.sleep(1000); - sendPathCommand((byte) 0x01, (short) 1, currentPoint.getLat(), currentPoint.getLng(), (short) 0); + // Thread.sleep(1000); + // sendPathCommand((byte) 0x01, (short) 1, currentPoint.getLat(), + // currentPoint.getLng(), (short) 0); // sendNextPoint(); return true; } else { diff --git a/maibu-common/src/main/java/com/maibu/influxdb/DeviceData.java b/maibu-common/src/main/java/com/maibu/influxdb/DeviceData.java deleted file mode 100644 index a338f5f..0000000 --- a/maibu-common/src/main/java/com/maibu/influxdb/DeviceData.java +++ /dev/null @@ -1,29 +0,0 @@ -package com.maibu.influxdb; - - -import com.maibu.influxdb.inter.InfluxField; -import com.maibu.influxdb.inter.InfluxTag; -import com.maibu.influxdb.inter.InfluxTime; -import com.maibu.influxdb.inter.InfluxMeasurement; -import lombok.Data; - -import java.time.Instant; - -//@Measurement(name = "device_data") -@Data -@InfluxMeasurement("device_data") -public class DeviceData { - -// @Column(tag = true) - @InfluxTag - private String deviceId; - - @InfluxField - private Double temperature; - - @InfluxField - private Double voltage; - - @InfluxTime - private Instant time; -} diff --git a/maibu-common/src/main/java/com/maibu/influxdb/MowerRealTimeData.java b/maibu-common/src/main/java/com/maibu/influxdb/MowerRealTimeData.java new file mode 100644 index 0000000..0a89495 --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/influxdb/MowerRealTimeData.java @@ -0,0 +1,155 @@ +package com.maibu.influxdb; + +import java.time.Instant; +import java.util.List; + +import org.springframework.util.CollectionUtils; + +import com.influxdb.query.FluxRecord; +import com.maibu.influxdb.inter.InfluxField; +import com.maibu.influxdb.inter.InfluxMeasurement; +import com.maibu.influxdb.inter.InfluxTag; +import com.maibu.influxdb.inter.InfluxTime; + +import lombok.Data; + +//@Measurement(name = "device_data") +@Data +@InfluxMeasurement("mower_realTime_data") +public class MowerRealTimeData { + + // @Column(tag = true) + @InfluxTag + private String deviceId; + + @InfluxField + private String voltage;// 电压 + + @InfluxField + private String leftTargetSpeed; // 左轮目标速度 + + @InfluxField + private String rightTargetSpeed;// 右轮目标速度 + + @InfluxField + private String leftMeasureSpeed;// 左轮测量速度 + + @InfluxField + private String rightMeasureSpeed;// 右轮测量速度 + + @InfluxField + private String leftCurrent;// 左轮电流 + + @InfluxField + private String rightCurrent;// 右轮电流 + + @InfluxField + private String leftMotorTemp;// 左轮温度 + + @InfluxField + private String rightMotorTemp;// 右轮温度 + + @InfluxField + private String chipTemp;// 芯片温度 + + @InfluxField + private String yaw;// 偏航角 + + @InfluxField + private String pitch;// 仰俯角 + + @InfluxField + private String roll;// 翻滚角 + + @InfluxField + private String satelliteCnt;// 跟踪的卫星数 + + @InfluxField + private String qual;// 定位质量 + + @InfluxField + private String headingStatus;// 航向角状态 + + @InfluxField + private String latitude;// 纬度 + + @InfluxField + private String longitude;// 经度 + + @InfluxField + private String cuttingSpeed;// 割刀速度 + + @InfluxField + private String controlMode;// 控制模式 + + @InfluxField + private String battery;// 电池电量 + + @InfluxField + private String workingArea;// 作业面积 + + @InfluxField + private String obstacleSign;// 障碍物标志位 + + @InfluxTime + private Instant time; + + public static MowerRealTimeData transferLatestInfluxData(FluxRecord record) { + + if (record == null || record.getValue() == null) { + return null; + } + + MowerRealTimeData data = new MowerRealTimeData(); + + data.setDeviceId((String) record.getValueByKey("deviceId")); + + data.setVoltage((String) record.getValueByKey("voltage")); + + data.setLeftTargetSpeed((String) record.getValueByKey("leftTargetSpeed")); + + data.setRightTargetSpeed((String) record.getValueByKey("rightTargetSpeed")); + + data.setLeftMeasureSpeed((String) record.getValueByKey("leftMeasureSpeed")); + + data.setRightMeasureSpeed((String) record.getValueByKey("rightMeasureSpeed")); + + data.setLeftMotorTemp((String) record.getValueByKey("leftMotorTemp")); + + data.setRightMotorTemp((String) record.getValueByKey("rightMotorTemp")); + + data.setLeftCurrent((String) record.getValueByKey("leftCurrent")); + + data.setRightCurrent((String) record.getValueByKey("rightCurrent")); + + data.setChipTemp((String) record.getValueByKey("chipTemp")); + + data.setYaw((String) record.getValueByKey("yaw")); + + data.setPitch((String) record.getValueByKey("pitch")); + + data.setRoll((String) record.getValueByKey("roll")); + + data.setSatelliteCnt((String) record.getValueByKey("satelliteCnt")); + + data.setQual((String) record.getValueByKey("qual")); + + data.setHeadingStatus((String) record.getValueByKey("headingStatus")); + + data.setLatitude((String) record.getValueByKey("latitude")); + + data.setLongitude((String) record.getValueByKey("longitude")); + + data.setCuttingSpeed((String) record.getValueByKey("cuttingSpeed")); + + data.setControlMode((String) record.getValueByKey("controlMode")); + + data.setBattery((String) record.getValueByKey("battery")); + + data.setWorkingArea((String) record.getValueByKey("workingArea")); + + data.setObstacleSign((String) record.getValueByKey("obstacleSign")); + + return data; + } +} diff --git a/maibu-common/src/main/java/com/maibu/influxdb/test.java b/maibu-common/src/main/java/com/maibu/influxdb/test.java deleted file mode 100644 index 8393eaf..0000000 --- a/maibu-common/src/main/java/com/maibu/influxdb/test.java +++ /dev/null @@ -1,65 +0,0 @@ -package com.maibu.influxdb; - -import com.influxdb.query.FluxRecord; -import com.influxdb.query.FluxTable; -import com.maibu.influxdb.util.InfluxDBUtil; -import com.maibu.influxdb.util.InfluxSqlBuilder; -import org.springframework.beans.factory.annotation.Value; - -import java.time.Instant; -import java.util.List; -import java.util.Map; - -public class test { - - - @Value("${ioTDA.serverIp}") - private String serverIp; - -// @Value("${ioTDA.deviceId}") -// private String deviceId; - - @Value("${ioTDA.secret}") - private String secret; - - @Value("${ioTDA.serviceId}") - private String serviceId; - - public static void main(String[] args) { - - String url = "http://localhost:8086"; - String token = "maibu-token"; - String org = "maibu"; - String bucket = "device_data"; -// -// InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket); -// -// -// WriteApiBlocking writeApi = client.getWriteApiBlocking(); - - InfluxDBUtil dbUtil = new InfluxDBUtil(url,token,org,bucket); - - - DeviceData data = new DeviceData(); - data.setDeviceId("AGV_001"); - data.setTemperature(36.522); - data.setVoltage(48.222); - data.setTime(Instant.now()); - - String line = InfluxSqlBuilder.buildLine(data); - dbUtil.write(line); - - - - Map map = InfluxSqlBuilder.buildTagMap(data,DeviceData::getDeviceId); - String flux = InfluxSqlBuilder.buildQuery(bucket,bucket,10,map,"temperature"); - - List tables = dbUtil.query(flux); - - for (FluxTable table : tables) { - for (FluxRecord record : table.getRecords()) { - System.out.println(record.getValue()); - } - } - } -} diff --git a/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxDBUtil.java b/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxDBUtil.java index e88d792..755d321 100644 --- a/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxDBUtil.java +++ b/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxDBUtil.java @@ -14,15 +14,15 @@ public class InfluxDBUtil { private InfluxDBClient client; private WriteApi writeApi; - private String bucket; - private String org; + public String bucket; + public String org; public InfluxDBUtil(String url, String token, String org, String bucket) { this.client = InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket); this.bucket = bucket; this.org = org; - // ✅ 异步批量写入（核心） + // 异步批量写入（核心） this.writeApi = client.makeWriteApi( WriteOptions.builder() .batchSize(5000) // 每5000条写一次 @@ -32,20 +32,20 @@ public class InfluxDBUtil { .build() ); - // ✅ 监听写入成功 + // 监听写入成功 this.writeApi.listenEvents(WriteSuccessEvent.class, event -> { // 可关闭或打debug日志 // System.out.println("写入成功"); }); - // ✅ 监听异常 + // 监听异常 this.writeApi.listenEvents(WriteErrorEvent.class, event -> { System.err.println("InfluxDB写入失败: " + event.getThrowable().getMessage()); }); } // ========================= - // ✅ 批量写入（推荐 Line Protocol） + // 批量写入（推荐 Line Protocol） // ========================= public void writeBatch(List lines) { for (String line : lines) { @@ -54,14 +54,14 @@ public class InfluxDBUtil { } // ========================= - // ✅ 单条写入 + // 单条写入 // ========================= public void write(String line) { writeApi.writeRecord(bucket, org, WritePrecision.MS, line); } // ========================= - // ✅ 查询（返回原始结果） + // 查询（返回原始结果） // ========================= public List query(String flux) { QueryApi queryApi = client.getQueryApi(); @@ -69,7 +69,7 @@ public class InfluxDBUtil { } // ========================= - // ✅ 查询（转List） + // 查询（转List） // ========================= public List queryValues(String flux) { List resultList = new ArrayList<>(); @@ -86,14 +86,14 @@ public class InfluxDBUtil { } // ========================= - // ✅ 手动flush（重要） + // 手动flush（重要） // ========================= public void flush() { writeApi.flush(); } // ========================= - // ✅ 关闭资源（必须） + // 关闭资源（必须） // ========================= public void close() { try { diff --git a/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxSqlBuilder.java b/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxSqlBuilder.java index a3a1617..2ad6fbf 100644 --- a/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxSqlBuilder.java +++ b/maibu-common/src/main/java/com/maibu/influxdb/util/InfluxSqlBuilder.java @@ -6,6 +6,7 @@ import com.maibu.influxdb.inter.InfluxTime; import com.maibu.influxdb.inter.InfluxMeasurement; import java.lang.reflect.Field; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.StringJoiner; @@ -13,6 +14,7 @@ import java.util.StringJoiner; public class InfluxSqlBuilder { /** * 插入语句拼接 + * * @param obj * @return */ @@ -41,7 +43,8 @@ public class InfluxSqlBuilder { field.setAccessible(true); Object value = field.get(obj); - if (value == null) continue; + if (value == null) + continue; String key = escapeKey(field.getName()); @@ -89,49 +92,130 @@ public class InfluxSqlBuilder { } } - // ========================= // ✅ 构建 Flux 查询 // ========================= + // public static String buildQuery( + // String bucket, + // String measurement, + // long startHoursAgo, + // Map tagFilters, + // String field) { + + // StringBuilder flux = new StringBuilder(); + + // flux.append("from(bucket:\"").append(bucket).append("\")") + // .append(" |> range(start: -").append(startHoursAgo).append("h)") + // .append(" |> filter(fn: (r) => r._measurement == \"") + // .append(measurement).append("\")"); + + // // tag过滤 + // if (tagFilters != null && !tagFilters.isEmpty()) { + // for (Map.Entry entry : tagFilters.entrySet()) { + // flux.append(" |> filter(fn: (r) => r.") + // .append(entry.getKey()) + // .append(" == \"") + // .append(entry.getValue()) + // .append("\")"); + // } + // } + + // // field过滤 + // if (field != null) { + // flux.append(" |> filter(fn: (r) => r._field == \"") + // .append(field) + // .append("\")"); + // } + + // // 按时间倒序 + // flux.append(" |> sort(columns:[\"_time\"], desc:true)"); + + // return flux.toString(); + // } + public static String buildQuery( String bucket, String measurement, - long startSecondsAgo, - Map tagFilters, - String field - ) { + long startHoursAgo, + Map> tagFilters, + String field, + boolean latestPerDevice, + String groupColumn) { StringBuilder flux = new StringBuilder(); - flux.append("from(bucket:\"").append(bucket).append("\")") - .append(" |> range(start: -").append(startSecondsAgo).append("h)") + flux.append("from(bucket:\"") + .append(bucket) + .append("\")") + .append(" |> range(start: -") + .append(startHoursAgo) + .append("h)") .append(" |> filter(fn: (r) => r._measurement == \"") - .append(measurement).append("\")"); + .append(measurement) + .append("\")"); - // tag过滤 + // Tag过滤 if (tagFilters != null && !tagFilters.isEmpty()) { - for (Map.Entry entry : tagFilters.entrySet()) { - flux.append(" |> filter(fn: (r) => r.") - .append(entry.getKey()) - .append(" == \"") - .append(entry.getValue()) - .append("\")"); + + for (Map.Entry> entry : tagFilters.entrySet()) { + + Collection values = entry.getValue(); + + if (values == null || values.isEmpty()) { + continue; + } + + flux.append(" |> filter(fn: (r) => "); + + boolean first = true; + + for (String value : values) { + + if (!first) { + flux.append(" or "); + } + + flux.append("r.") + .append(entry.getKey()) + .append(" == \"") + .append(value) + .append("\""); + + first = false; + } + + flux.append(")"); } } - // field过滤 - if (field != null) { + // Field过滤 + if (field != null && !field.isEmpty()) { flux.append(" |> filter(fn: (r) => r._field == \"") .append(field) .append("\")"); } + // 每个设备最新一条 + if (latestPerDevice) { + + if (groupColumn == null || groupColumn.isEmpty()) { + groupColumn = "deviceId"; + } + + flux.append(" |> group(columns:[\"") + .append(groupColumn) + .append("\"])") + .append(" |> last()"); + } else { + flux.append(" |> sort(columns:[\"_time\"], desc:true)"); + } + return flux.toString(); } - /** * 类的某些属性转换成map + * * @param obj * @param functions * @return @@ -139,8 +223,7 @@ public class InfluxSqlBuilder { */ public static Map buildTagMap( T obj, - SFunction... functions - ) { + SFunction... functions) { Map map = new HashMap<>(); try { @@ -190,5 +273,4 @@ public class InfluxSqlBuilder { return value.toString(); // double/float } - } \ No newline at end of file diff --git a/maibu-common/src/main/java/com/maibu/influxdb/util/LambdaUtils.java b/maibu-common/src/main/java/com/maibu/influxdb/util/LambdaUtils.java index ebcf6e0..0709b63 100644 --- a/maibu-common/src/main/java/com/maibu/influxdb/util/LambdaUtils.java +++ b/maibu-common/src/main/java/com/maibu/influxdb/util/LambdaUtils.java @@ -4,6 +4,8 @@ import java.beans.Introspector; import java.lang.invoke.SerializedLambda; import java.lang.reflect.Method; +import com.maibu.influxdb.inter.InfluxMeasurement; + public class LambdaUtils { public static String getFieldName(SFunction fn) { @@ -26,4 +28,16 @@ public class LambdaUtils { throw new RuntimeException("解析字段失败", e); } } + + public static String getMeasurement(Class clazz) { + + InfluxMeasurement annotation = clazz.getAnnotation(InfluxMeasurement.class); + + if (annotation == null) { + throw new IllegalArgumentException( + clazz.getName() + " 未配置 @InfluxMeasurement"); + } + + return annotation.value(); + } } diff --git a/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java b/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java index 4425b67..9dab0a5 100644 --- a/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java +++ b/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java @@ -1,30 +1,26 @@ package com.maibu.memory; - -import com.maibu.core.business.Device; -import com.maibu.core.business.IoTCommonDevice; -import com.maibu.core.business.IoTCommonProduct; -import com.maibu.core.business.uav.IotUAVDevice; -import lombok.Data; -import org.springframework.stereotype.Component; -import org.springframework.util.CollectionUtils; - import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + +import com.maibu.core.business.Device; +import com.maibu.core.business.IoTCommonDevice; +import com.maibu.core.business.IoTCommonProduct; +import com.maibu.influxdb.util.InfluxDBUtil; +import com.maibu.mqtt.MqttClientUtil; + +import lombok.Data; + @Component @Data public class GlobalMemory { - - // orgId - siteId - deviceId -// public static ConcurrentHashMap unRegisterCommonDeviceMap = new ConcurrentHashMap<>(); -// -// public static ConcurrentHashMap unRegisterDeviceMap = new ConcurrentHashMap<>(); - // orgId - siteId - deviceId public static ConcurrentHashMap unRegisterCommonDeviceMap = new ConcurrentHashMap<>(); @@ -35,6 +31,9 @@ public class GlobalMemory { public static ConcurrentHashMap productMap = new ConcurrentHashMap<>(); + public static InfluxDBUtil influxDBUtil; + + public static MqttClientUtil mqttClientUtil; public static SiteMemory getSiteMemory(Long orgId, Long siteId) { if (orgId != null && siteId != null) { @@ -56,11 +55,12 @@ public class GlobalMemory { /** * 新增场站内存 + * * @param orgId * @param siteId * @param siteMemory */ - //todo 初始化 + // todo 初始化 public static void addSiteMemory(Long orgId, Long siteId, SiteMemory siteMemory) { if (orgId != null && siteId != null) { @@ -75,6 +75,7 @@ public class GlobalMemory { /** * 删除 + * * @param orgId * @param siteId */ diff --git a/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java b/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java index b6f2c74..72e9841 100644 --- a/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java +++ b/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java @@ -1,22 +1,23 @@ package com.maibu.memory; -import cn.hutool.core.util.ObjectUtil; -import com.maibu.core.business.*; -import com.maibu.core.business.device.Connector; -import com.maibu.core.business.device.NettyDevice; -import com.maibu.core.enums.ConnectorStatus; -import com.maibu.core.enums.ConnectorType; -import io.netty.channel.Channel; -import lombok.Data; -import lombok.extern.slf4j.Slf4j; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; + import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.stream.Collectors; +import com.maibu.core.business.Device; +import com.maibu.core.business.DevicePlan; +import com.maibu.core.business.DevicePlanTask; +import com.maibu.core.business.ErrorIdentificationStandard; +import com.maibu.core.business.IoTCommonDevice; + +import lombok.Data; +import lombok.extern.slf4j.Slf4j; @Component @Data diff --git a/maibu-common/src/main/java/com/maibu/mqtt/MqttClientUtil.java b/maibu-common/src/main/java/com/maibu/mqtt/MqttClientUtil.java new file mode 100644 index 0000000..c4dcf9f --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/mqtt/MqttClientUtil.java @@ -0,0 +1,135 @@ +package com.maibu.mqtt; + + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +@Slf4j +public class MqttClientUtil { + + private final MqttClient client; + + public MqttClientUtil( + String broker, + String clientId, + String username, + String password) throws MqttException { + + this.client = new MqttClient( + broker, + clientId, + new MemoryPersistence()); + + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(false); + options.setAutomaticReconnect(true); + options.setConnectionTimeout(10); + options.setKeepAliveInterval(60); + + if (username != null) { + options.setUserName(username); + } + + if (password != null) { + options.setPassword(password.toCharArray()); + } + + client.connect(options); + + log.info("MQTT连接成功: {}", broker); + } + + /** + * 订阅 + */ + public void subscribe(String topic) throws MqttException { + + client.subscribe(topic); + + log.info("订阅成功: {}", topic); + } + + /** + * 订阅并监听 + */ + public void subscribe( + String topic, + IMqttMessageListener listener) + throws MqttException { + + client.subscribe(topic, listener); + + log.info("订阅成功: {}", topic); + } + + /** + * 发布消息 + */ + public void publish( + String topic, + String payload) + throws MqttException { + + publish(topic, payload, 1); + } + + /** + * 发布消息 + */ + public void publish( + String topic, + String payload, + int qos) + throws MqttException { + + MqttMessage message = + new MqttMessage(payload.getBytes()); + + message.setQos(qos); + + client.publish(topic, message); + } + + /** + * 设置全局监听器 + */ + public void setCallback() { + + client.setCallback(new MqttCallback() { + + @Override + public void connectionLost(Throwable cause) { + + log.error("MQTT断开连接", cause); + } + + @Override + public void messageArrived( + String topic, + MqttMessage message) { + + log.info( + "收到消息 topic={} payload={}", + topic, + new String(message.getPayload())); + } + + @Override + public void deliveryComplete( + IMqttDeliveryToken token) { + + log.debug("消息发送成功"); + } + }); + } + + public boolean isConnected() { + return client.isConnected(); + } + + public void disconnect() throws MqttException { + client.disconnect(); + client.close(); + } +} \ No newline at end of file diff --git a/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java new file mode 100644 index 0000000..4b2aa3a --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java @@ -0,0 +1,10 @@ +package com.maibu.mqtt; + +public class MqttTopic { + + public static final String DEVICE_STATUS_TOPIC = "device/%s/realTimeMessage"; // 实时消息 + + public static final String DEVICE_TASK_STATUS_TOPIC = "task/%s/status"; // 任务状态 完成、取消、暂停、继续 + + public static final String DEVICE_TASK_ARRIVE_TOPIC = "task/%s/arrive"; // 任务到达 +} diff --git a/maibu-netty-server/src/main/java/com/maibu/dto/DeviceTaskQueryDTO.java b/maibu-netty-server/src/main/java/com/maibu/dto/DeviceTaskQueryDTO.java index c850759..106d5a1 100644 --- a/maibu-netty-server/src/main/java/com/maibu/dto/DeviceTaskQueryDTO.java +++ b/maibu-netty-server/src/main/java/com/maibu/dto/DeviceTaskQueryDTO.java @@ -1,5 +1,8 @@ package com.maibu.dto; +import java.time.LocalDateTime; + +import com.fasterxml.jackson.annotation.JsonFormat; import com.maibu.core.domain.BaseEntity; import com.maibu.core.enums.DeviceTaskStaus; @@ -11,8 +14,10 @@ import lombok.EqualsAndHashCode; public class DeviceTaskQueryDTO extends BaseEntity { private Long userId; -// private LocalDateTime startTime; -// private LocalDateTime endTime; + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private LocalDateTime startTime; + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private LocalDateTime endTime; private DeviceTaskStaus taskStaus; private String planName; private Long planId; diff --git a/maibu-web-middleware/src/main/java/com/maibu/dto/WebStatusMessageDTO.java b/maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java similarity index 79% rename from maibu-web-middleware/src/main/java/com/maibu/dto/WebStatusMessageDTO.java rename to maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java index fe2cf7c..68525fa 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/dto/WebStatusMessageDTO.java +++ b/maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java @@ -4,6 +4,7 @@ import lombok.Data; import java.util.List; +import com.maibu.core.business.device.DeviceStatusDetail; import com.maibu.core.enums.MesType; @Data diff --git a/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java b/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java index b5087a7..3edad51 100644 --- a/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java +++ b/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java @@ -1,20 +1,13 @@ package com.maibu.init; -import com.maibu.constant.FastBeeConstant; -import com.maibu.core.business.DevicePlan; -import com.maibu.core.business.DevicePlanTask; -import com.maibu.core.domain.entity.SysSite; -import com.maibu.core.enums.DeviceTaskStaus; -import com.maibu.mapper.DevicePlanMapper; -import com.maibu.mapper.DevicePlanTaskMapper; -import com.maibu.mapper.SysSiteMapper; -import com.maibu.mapper.SysUserClientMapper; -import com.maibu.memory.GlobalMemory; -import com.maibu.memory.SiteMemory; -import com.maibu.mybatis.LambdaQueryWrapperX; -import com.maibu.service.DevicePlanTaskMonitorService; -import com.maibu.service.DeviceThreadService; -import lombok.extern.slf4j.Slf4j; +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Executor; + +import javax.annotation.Resource; + +import org.eclipse.paho.client.mqttv3.MqttException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; @@ -22,19 +15,29 @@ import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; -import javax.annotation.Resource; -import java.io.IOException; -import java.util.Arrays; -import java.util.List; -import java.util.concurrent.Executor; - +import com.maibu.constant.FastBeeConstant; +import com.maibu.core.business.DevicePlan; +import com.maibu.core.business.DevicePlanTask; +import com.maibu.core.domain.entity.SysSite; +import com.maibu.core.enums.DeviceTaskStaus; +import com.maibu.influxdb.util.InfluxDBUtil; +import com.maibu.mapper.DevicePlanMapper; +import com.maibu.mapper.DevicePlanTaskMapper; +import com.maibu.mapper.SysSiteMapper; +import com.maibu.memory.GlobalMemory; +import com.maibu.memory.SiteMemory; +import com.maibu.mqtt.MqttClientUtil; +import com.maibu.mybatis.LambdaQueryWrapperX; +import com.maibu.service.DevicePlanTaskMonitorService; +import com.maibu.service.DeviceThreadService; +import org.springframework.beans.factory.annotation.Value; +import lombok.extern.slf4j.Slf4j; @Slf4j @Component @Order(20) public class InitThread implements ApplicationRunner { - @Resource(name = FastBeeConstant.TASK.DEVICE_ERROR_MONITOR) private Executor deviceErrorMonitorExecutor; @@ -56,15 +59,48 @@ public class InitThread implements ApplicationRunner { @Autowired private SysSiteMapper sysSiteMapper; + @Value("${influxdb.url}") + private String influxDBUrl; + + @Value("${influxdb.username}") + private String influxDBUsername; + + @Value("${influxdb.password}") + private String influxDBPassword; + + @Value("${influxdb.bucket}") + private String influxDBBucket; + + @Value("${influxdb.org}") + private String influxDBOrg; + + @Value("${influxdb.token}") + private String influxDBToken; + + @Value("${mqtt.username}") + private String mqttUsername; + + @Value("${mqtt.password}") + private String mqttPassword; + + @Value("${mqtt.host-url}") + private String mqttHostUrl; + + @Value("${mqtt.client-id}") + private String mqttClientId; + @Override - public void run(ApplicationArguments args) { + public void run(ApplicationArguments args) throws MqttException { recoverMemory(); init(); } - - private void init() { - //设备错误监听线程 + private void init() throws MqttException { + // 初始化InfluxDBUtil + GlobalMemory.influxDBUtil = new InfluxDBUtil(influxDBUrl, influxDBToken, influxDBOrg, influxDBBucket); + // 初始化MqttClientUtil + GlobalMemory.mqttClientUtil = new MqttClientUtil(mqttHostUrl, mqttClientId, mqttUsername, mqttPassword); + // 设备错误监听线程 deviceErrorMonitorExecutor.execute(() -> { try { deviceThreadService.deviceErrorMonitor(); @@ -72,14 +108,14 @@ public class InitThread implements ApplicationRunner { throw new RuntimeException(e); } }); - //设备任务处理线程 + // 设备任务处理线程 deviceTaskHandlerExecutor.execute(() -> { - //任务下发 + // 任务下发 try { devicePlanTaskMonitorService.generatePlanTask(); - //执行准备队列 + // 执行准备队列 devicePlanTaskMonitorService.doLoop(); - //执行任务 + // 执行任务 devicePlanTaskMonitorService.doExecute(); } catch (InterruptedException e) { throw new RuntimeException(e); @@ -91,9 +127,9 @@ public class InitThread implements ApplicationRunner { * 恢复内存 */ public void recoverMemory() { - //todo 启动时清除当前登录记录 -// System.out.println("开始清除登录信息"); -// sysUserClientMapper.clearDeviceName(); + // todo 启动时清除当前登录记录 + // System.out.println("开始清除登录信息"); + // sysUserClientMapper.clearDeviceName(); List siteList = sysSiteMapper.selectList(); if (!CollectionUtils.isEmpty(siteList)) { siteList.forEach(x -> { @@ -108,7 +144,8 @@ public class InitThread implements ApplicationRunner { } // 未执行完的重复计划 LambdaQueryWrapperX query = new LambdaQueryWrapperX<>(); - List list = Arrays.asList(DeviceTaskStaus.NEW.getCode(), DeviceTaskStaus.EXECUTING.getCode(), DeviceTaskStaus.PAUSE.getCode()); + List list = Arrays.asList(DeviceTaskStaus.NEW.getCode(), DeviceTaskStaus.EXECUTING.getCode(), + DeviceTaskStaus.PAUSE.getCode()); query.in(DevicePlan::getTaskStaus, list); List devicePlanList = devicePlanMapper.selectList(query); if (!CollectionUtils.isEmpty(devicePlanList)) { @@ -116,7 +153,7 @@ public class InitThread implements ApplicationRunner { Long orgId = x.getOrgId(); Long siteId = x.getSiteId(); SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId); - if(siteMemory!=null){ + if (siteMemory != null) { siteMemory.addDevicePlan(x); } }); @@ -131,7 +168,7 @@ public class InitThread implements ApplicationRunner { Long orgId = x.getOrgId(); Long siteId = x.getSiteId(); SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId); - if(siteMemory!=null){ + if (siteMemory != null) { siteMemory.addDevicePlanPrepare(x); } }); @@ -145,7 +182,7 @@ public class InitThread implements ApplicationRunner { Long orgId = x.getOrgId(); Long siteId = x.getSiteId(); SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId); - if(siteMemory!=null){ + if (siteMemory != null) { siteMemory.addDevicePlanExecute(x); } }); diff --git a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java index 9862891..309ae71 100644 --- a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java +++ b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java @@ -1,102 +1,333 @@ package com.maibu.netty.handler; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; + +import org.apache.commons.codec.binary.Hex; +import org.apache.commons.lang3.StringUtils; +import org.eclipse.paho.client.mqttv3.MqttException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + import com.maibu.common.CommandConstant; import com.maibu.common.NettyCacheKey; import com.maibu.core.business.DeviceRunStatistics; import com.maibu.core.business.DeviceRunningStatusHistory; import com.maibu.core.business.DeviceStatusRecordDTO; -import com.maibu.core.redis.RedisCache; +import com.maibu.core.business.device.DeviceStatusDetail; import com.maibu.core.business.device.NettyDevice; +import com.maibu.core.business.inter.WebsocketMesDispather; +import com.maibu.core.enums.MesType; +import com.maibu.core.redis.RedisCache; +import com.maibu.dto.WebStatusMessageDTO; +import com.maibu.influxdb.MowerRealTimeData; +import com.maibu.influxdb.util.InfluxSqlBuilder; import com.maibu.manager.DeviceSessionManager; +import com.maibu.memory.GlobalMemory; +import com.maibu.mqtt.MqttTopic; +import com.maibu.utils.json.JsonUtils; + import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; -import org.apache.commons.codec.binary.Hex; -import org.apache.commons.lang3.StringUtils; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Component; - -import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.Date; -import java.util.List; @Component @ChannelHandler.Sharable public class DataToDataBaseHandler extends SimpleChannelInboundHandler { + private static final Logger logger = LoggerFactory.getLogger(DataToDataBaseHandler.class); + @Autowired private DeviceSessionManager deviceSessionManager; @Autowired private RedisCache redisCache; - private static final Long interval = 5000L; + @Autowired + private WebsocketMesDispather websocketMesDispather; + private static final Long interval = 5000L; @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { - Channel channel = ctx.channel(); - String deviceID = deviceSessionManager.getDeviceIdByChannel(channel); - if (StringUtils.isEmpty(deviceID)) return; + try { + Channel channel = ctx.channel(); + String deviceID = deviceSessionManager.getDeviceIdByChannel(channel); + if (StringUtils.isEmpty(deviceID)) + return; - byte[] bytes = Hex.decodeHex(msg.toCharArray()); - if (bytes[2] == CommandConstant.status) { - String key = NettyCacheKey.deviceRunningStatusKey + deviceID; - DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(key); - boolean record = false; - List histories; - if (recordDTO == null) { - recordDTO = new DeviceStatusRecordDTO(); - histories = new ArrayList<>(); - record = true; - } else { - long lastRecordTime = recordDTO.getLastRecordTime(); - histories = recordDTO.getHistories(); - if (System.currentTimeMillis() - lastRecordTime >= interval) { - record = true; - } - } - if (record) { - String deviceStatus = new String(bytes, 2, bytes.length - 4, StandardCharsets.UTF_8).trim(); - String[] split = deviceStatus.split(","); - - DeviceRunningStatusHistory deviceRunningStatus = getDeviceRunningStatusHistory(deviceID, split); - if (histories == null) { + byte[] bytes = Hex.decodeHex(msg.toCharArray()); + if (bytes[2] == CommandConstant.status) { + String key = NettyCacheKey.deviceRunningStatusKey + deviceID; + DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(key); + boolean record = false; + List histories; + if (recordDTO == null) { + recordDTO = new DeviceStatusRecordDTO(); histories = new ArrayList<>(); + record = true; + } else { + long lastRecordTime = recordDTO.getLastRecordTime(); + histories = recordDTO.getHistories(); + if (System.currentTimeMillis() - lastRecordTime >= interval) { + record = true; + } } - histories.add(deviceRunningStatus); - recordDTO.setHistories(histories); - recordDTO.setLastRecordTime(System.currentTimeMillis()); + if (record) { + // 推送websocket实时消息给前端 todo mqtt推送 - redisCache.setCacheObject(key, recordDTO); - // 判断整点 记录到数据库 -// deviceStatusHistoryMapper.insert(deviceRunningStatus); - NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceID); - String workArea = deviceRunningStatus.getWorkingArea(); - DeviceRunStatistics latestStatus; - if(nettyDevice.getLatestStatus() == null){ - latestStatus = new DeviceRunStatistics(); - latestStatus.setStartTime(LocalDateTime.now()); - latestStatus.setDeviceId(deviceID); - }else { - latestStatus = nettyDevice.getLatestStatus(); - // 使用Duration.between计算两个时间点之间的持续时间 - Duration duration = Duration.between(latestStatus.getStartTime(), LocalDateTime.now()); - long minutes = duration.toMinutes(); - latestStatus.setTime(minutes); + String deviceStatus = new String(bytes, 2, bytes.length - 4, StandardCharsets.UTF_8).trim(); + String[] split = deviceStatus.split(","); + WebStatusMessageDTO dto = createWebDeviceStatusMessage(split); + try { + GlobalMemory.mqttClientUtil.publish( + String.format(MqttTopic.DEVICE_STATUS_TOPIC, deviceID), + JsonUtils.toJsonString(dto)); + } catch (MqttException e) { + logger.error("发送实时状态数据错误: {}", e.getMessage()); + } + List controlMasters = deviceSessionManager + .getAllSlaveControl(deviceID); + if (!CollectionUtils.isEmpty(controlMasters)) { + controlMasters.forEach(x -> { + websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto)); + }); + } + DeviceRunningStatusHistory deviceRunningStatus = getDeviceRunningStatusHistory(deviceID, split); + if (histories == null) { + histories = new ArrayList<>(); + } + histories.add(deviceRunningStatus); + recordDTO.setHistories(histories); + recordDTO.setLastRecordTime(System.currentTimeMillis()); + redisCache.setCacheObject(key, recordDTO); + + // todo 记录到InfluxDB 是否删除redis中的记录 目前先保留最近一次的记录在redis中 + MowerRealTimeData data = processInfluxData(deviceID, split); + String line = InfluxSqlBuilder.buildLine(data); + GlobalMemory.influxDBUtil.write(line); + + // 判断整点 记录到数据库 + // deviceStatusHistoryMapper.insert(deviceRunningStatus); + NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceID); + String workArea = deviceRunningStatus.getWorkingArea(); + DeviceRunStatistics latestStatus; + if (nettyDevice.getLatestStatus() == null) { + latestStatus = new DeviceRunStatistics(); + latestStatus.setStartTime(LocalDateTime.now()); + latestStatus.setDeviceId(deviceID); + } else { + latestStatus = nettyDevice.getLatestStatus(); + // 使用Duration.between计算两个时间点之间的持续时间 + Duration duration = Duration.between(latestStatus.getStartTime(), LocalDateTime.now()); + long minutes = duration.toMinutes(); + latestStatus.setTime(minutes); + } + latestStatus.setWorkArea(Double.valueOf(workArea)); + nettyDevice.setLatestStatus(latestStatus); } - latestStatus.setWorkArea(Double.valueOf(workArea)); - nettyDevice.setLatestStatus(latestStatus); } -// else{ -// System.out.println(false); -// } + ctx.fireChannelRead(msg); + } catch (Exception e) { + logger.error("处理数据异常: {}", e.getMessage()); } - ctx.fireChannelRead(msg); + } + + public WebStatusMessageDTO createWebDeviceStatusMessage(String[] split) { + WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO(); + List transferData = transferStatusData(split); + webStatusMessageDTO.setData(transferData); + webStatusMessageDTO.setType(MesType.deviceInfo); + return webStatusMessageDTO; + } + + private List transferStatusData(String[] split) { + List transferData = new ArrayList<>(); + + DeviceStatusDetail detail0 = new DeviceStatusDetail(); + detail0.setName("voltage"); + detail0.setValue(split[0]); + detail0.setUnit(""); + transferData.add(detail0); + + DeviceStatusDetail detail1 = new DeviceStatusDetail(); + detail1.setName("leftTargetSpeed"); + detail1.setValue(split[1]); + detail1.setUnit(""); + transferData.add(detail1); + + DeviceStatusDetail detail2 = new DeviceStatusDetail(); + detail2.setName("rightTargetSpeed"); + detail2.setValue(split[2]); + detail2.setUnit(""); + transferData.add(detail2); + + DeviceStatusDetail detail3 = new DeviceStatusDetail(); + detail3.setName("leftMeasureSpeed"); + detail3.setValue(split[3]); + detail3.setUnit(""); + transferData.add(detail3); + + DeviceStatusDetail detail4 = new DeviceStatusDetail(); + detail4.setName("rightMeasureSpeed"); + detail4.setValue(split[4]); + detail4.setUnit(""); + transferData.add(detail4); + + DeviceStatusDetail detail5 = new DeviceStatusDetail(); + detail5.setName("leftCurrent"); + detail5.setValue(split[5]); + detail5.setUnit(""); + transferData.add(detail5); + + DeviceStatusDetail detail6 = new DeviceStatusDetail(); + detail6.setName("rightCurrent"); + detail6.setValue(split[6]); + detail6.setUnit(""); + transferData.add(detail6); + + DeviceStatusDetail detail7 = new DeviceStatusDetail(); + detail7.setName("leftMotorTemp"); + detail7.setValue(split[7]); + detail7.setUnit(""); + transferData.add(detail7); + + DeviceStatusDetail detail8 = new DeviceStatusDetail(); + detail8.setName("rightMotorTemp"); + detail8.setValue(split[8]); + detail8.setUnit(""); + transferData.add(detail8); + + DeviceStatusDetail detail9 = new DeviceStatusDetail(); + detail9.setName("chipTemp"); + detail9.setValue(split[9]); + detail9.setUnit(""); + transferData.add(detail9); + + DeviceStatusDetail detail10 = new DeviceStatusDetail(); + detail10.setName("yaw"); + detail10.setValue(split[10]); + detail10.setUnit(""); + transferData.add(detail10); + + DeviceStatusDetail detail11 = new DeviceStatusDetail(); + detail11.setName("pitch"); + detail11.setValue(split[11]); + detail11.setUnit(""); + transferData.add(detail11); + + DeviceStatusDetail detail12 = new DeviceStatusDetail(); + detail12.setName("roll"); + detail12.setValue(split[12]); + detail12.setUnit(""); + transferData.add(detail12); + + DeviceStatusDetail detail13 = new DeviceStatusDetail(); + detail13.setName("satelliteCnt"); + detail13.setValue(split[13]); + detail13.setUnit(""); + transferData.add(detail13); + + DeviceStatusDetail detail14 = new DeviceStatusDetail(); + detail14.setName("qual"); + detail14.setValue(split[14]); + detail14.setUnit(""); + transferData.add(detail14); + + DeviceStatusDetail detail15 = new DeviceStatusDetail(); + detail15.setName("headingStatus"); + detail15.setValue(split[15]); + detail15.setUnit(""); + transferData.add(detail15); + + DeviceStatusDetail detail16 = new DeviceStatusDetail(); + detail16.setName("latitude"); + detail16.setValue(split[16]); + detail16.setUnit(""); + transferData.add(detail16); + + DeviceStatusDetail detail17 = new DeviceStatusDetail(); + detail17.setName("longitude"); + detail17.setValue(split[17]); + detail17.setUnit(""); + transferData.add(detail17); + + // 18 时间暂未给值 + + DeviceStatusDetail detail19 = new DeviceStatusDetail(); + detail19.setName("cuttingSpeed"); + detail19.setValue(split[19]); + detail19.setUnit(""); + transferData.add(detail19); + + DeviceStatusDetail detail20 = new DeviceStatusDetail(); + detail20.setName("controlMode"); + detail20.setValue(split[20]); + detail20.setUnit(""); + transferData.add(detail20); + + DeviceStatusDetail detail21 = new DeviceStatusDetail(); + detail21.setName("battery"); + detail21.setValue(split[21]); + detail21.setUnit(""); + transferData.add(detail21); + + DeviceStatusDetail detail22 = new DeviceStatusDetail(); + detail22.setName("workingArea"); + detail22.setValue(split[22]); + detail22.setUnit(""); + transferData.add(detail22); + + DeviceStatusDetail detail23 = new DeviceStatusDetail(); + detail23.setName("obstacleSign"); + detail23.setValue(split[23]); + detail23.setUnit(""); + transferData.add(detail23); + + return transferData; + } + + private MowerRealTimeData processInfluxData(String deviceID, String[] split) { + MowerRealTimeData data = new MowerRealTimeData(); + data.setDeviceId(deviceID); + data.setVoltage(split[0]); + data.setLeftTargetSpeed(split[1]); + data.setRightTargetSpeed(split[2]); + data.setLeftMeasureSpeed(split[3]); + data.setRightMeasureSpeed(split[4]); + data.setLeftCurrent(split[5]); + data.setRightCurrent(split[6]); + data.setLeftMotorTemp(split[7]); + data.setRightMotorTemp(split[8]); + data.setChipTemp(split[9]); + data.setYaw(split[10]); + data.setPitch(split[11]); + data.setRoll(split[12]); + data.setSatelliteCnt(split[13]); + data.setQual(split[14]); + data.setHeadingStatus(split[15]); + data.setLatitude(split[16]); + data.setLongitude(split[17]); + + // 18 时间暂未给值 + + data.setCuttingSpeed(split[19]); + data.setControlMode(split[20]); + data.setBattery(split[21]); + data.setWorkingArea(split[22]); + data.setObstacleSign(split[23]); + + data.setTime(Instant.now()); + return data; } private DeviceRunningStatusHistory getDeviceRunningStatusHistory(String deviceID, String[] split) { @@ -121,7 +352,7 @@ public class DataToDataBaseHandler extends SimpleChannelInboundHandler { deviceRunningStatus.setLatitude(split[16]); deviceRunningStatus.setLongitude(split[17]); - //18 时间暂未给值 + // 18 时间暂未给值 deviceRunningStatus.setCuttingSpeed(split[19]); deviceRunningStatus.setControlMode(split[20]); diff --git a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java index 92b8ad4..1f1ff00 100644 --- a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java +++ b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java @@ -4,17 +4,16 @@ import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.time.LocalDateTime; import java.util.List; -import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import javax.annotation.Resource; import org.apache.commons.codec.binary.Hex; import org.apache.commons.lang3.StringUtils; +import org.eclipse.paho.client.mqttv3.MqttException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; @@ -22,9 +21,7 @@ import com.maibu.common.CommandConstant; import com.maibu.common.Constant; import com.maibu.core.business.Device; import com.maibu.core.business.device.NettyDevice; -import com.maibu.core.business.inter.DeviceWebSocketInterace; import com.maibu.core.business.inter.WebsocketMesDispather; -import com.maibu.core.business.path.LatAndLngEntity; import com.maibu.core.domain.entity.SysUser; import com.maibu.core.domain.entity.UserClient; import com.maibu.core.enums.CommandRequestType; @@ -39,10 +36,11 @@ import com.maibu.dto.DeviceLoginResponseDTO; import com.maibu.dto.DeviceRequestDTO; import com.maibu.dto.DeviceRespondDTO; import com.maibu.dto.DeviceStatusChangeDTO; -import com.maibu.dto.DeviceTaskReportDTO; import com.maibu.dto.DeviceTaskStatusMessageDTO; import com.maibu.manager.DeviceSessionManager; import com.maibu.mapper.DeviceRunStatisticsMapper; +import com.maibu.memory.GlobalMemory; +import com.maibu.mqtt.MqttTopic; import com.maibu.service.IDeviceService; import com.maibu.service.ISysUserClientService; import com.maibu.service.ISysUserService; @@ -363,9 +361,13 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { changeDTO.setStatus(DeviceTaskStaus.FINISH.toString()); changeDTO.setType(MesType.device_task_change); changeDTO.setTaskId(taskId); - // ByteBuf buf = ctx.alloc().buffer(); - // ByteBuf buf = Unpooled.buffer(); - // CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction); + + try { + GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId), + JsonUtils.toJsonString(changeDTO)); + } catch (MqttException e) { + logger.error("任务完成状态推送失败 task:{},device:{}error:{}", taskId, deviceId, e.getMessage()); + } List controlMasters = sessionManager .getAllSlaveControl(deviceId); if (!CollectionUtils.isEmpty(controlMasters)) { @@ -373,14 +375,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(changeDTO)); // String content = buf.toString(CharsetUtil.UTF_8); - logger.info("任务完成状态推送 task:{},device:{}，content:{}", taskId, deviceId, + logger.info("任务完成状态推送 task:{},device:{},content:{}", taskId, deviceId, JsonUtils.toJsonString(changeDTO)); - // if (x.getChannel() != null && x.getChannel().isActive()) { - // x.getChannel().writeAndFlush(buf); - // String content = buf.toString(CharsetUtil.UTF_8); - // logger.info("任务完成状态推送 task:{},device:{}，content:{}", taskId, deviceId, - // content); - // } }); } } @@ -393,9 +389,14 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { reportDTO.setTaskId(taskId); reportDTO.setEntity(device.getCurrentPoint()); reportDTO.setType(MesType.device_task_arrive_point); - // ByteBuf buf = ctx.alloc().buffer(); - // ByteBuf buf = Unpooled.buffer(); - // CommandUtils.buildCommand(buf, reportDTO, CommandConstant.interaction); + try { + + GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_TASK_ARRIVE_TOPIC, taskId), + JsonUtils.toJsonString(reportDTO)); + } catch (MqttException e) { + logger.error("任务点位到达推送失败 task:{},device:{}error:{}", taskId, deviceId, e.getMessage()); + } + List controlMasters = sessionManager .getAllSlaveControl(deviceId); if (!CollectionUtils.isEmpty(controlMasters)) { @@ -404,12 +405,6 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { JsonUtils.toJsonString(reportDTO)); logger.info("任务点位到达推送 task:{},device:{},content:{}", taskId, deviceId, JsonUtils.toJsonString(reportDTO)); - // if (x.getChannel() != null && x.getChannel().isActive()) { - // x.getChannel().writeAndFlush(buf); - // String content = buf.toString(CharsetUtil.UTF_8); - // logger.info("任务点位到达推送 task:{},device:{},content:{}", taskId, deviceId, - // content); - // } }); } } diff --git a/maibu-netty-server/src/main/java/com/maibu/scheduled/ScheduledTask.java b/maibu-netty-server/src/main/java/com/maibu/scheduled/ScheduledTask.java index 4645a35..f2780e7 100644 --- a/maibu-netty-server/src/main/java/com/maibu/scheduled/ScheduledTask.java +++ b/maibu-netty-server/src/main/java/com/maibu/scheduled/ScheduledTask.java @@ -44,6 +44,7 @@ public class ScheduledTask { // recordDTO.getHistories().forEach(x -> { // deviceStatusHistoryMapper.insert(x); // }); + deviceStatusHistoryMapper.insertBatch(recordDTO.getHistories()); } redisCache.deleteObject(k); diff --git a/maibu-netty-server/src/main/java/com/maibu/service/DeviceTaskService.java b/maibu-netty-server/src/main/java/com/maibu/service/DeviceTaskService.java index 94b624b..185c65b 100644 --- a/maibu-netty-server/src/main/java/com/maibu/service/DeviceTaskService.java +++ b/maibu-netty-server/src/main/java/com/maibu/service/DeviceTaskService.java @@ -375,6 +375,18 @@ public class DeviceTaskService { if (!StringUtils.isEmpty(dto.getDeviceId())) { list = list.stream().filter(x -> x.getDeviceId().equals(dto.getDeviceId())).collect(Collectors.toList()); } + if (dto.getStartTime() != null) { + list = list.stream() + .filter(x -> x.getCreateTime() != null + && !x.getCreateTime().isBefore(dto.getStartTime())) + .collect(Collectors.toList()); + } + if (dto.getEndTime() != null) { + list = list.stream() + .filter(x -> x.getCreateTime() != null + && !x.getCreateTime().isAfter(dto.getEndTime())) + .collect(Collectors.toList()); + } return transferDeviceTaskData(list); } @@ -395,6 +407,12 @@ public class DeviceTaskService { if (!StringUtils.isEmpty(dto.getDeviceId())) { queryWrapper.eq(DevicePlanTask::getDeviceId, dto.getDeviceId()); } + if (dto.getStartTime() != null) { + queryWrapper.ge(DevicePlanTask::getCreateTime, dto.getStartTime()); + } + if (dto.getEndTime() != null) { + queryWrapper.le(DevicePlanTask::getCreateTime, dto.getEndTime()); + } return transferDeviceTaskData(devicePlanTaskMapper.selectList(queryWrapper)); } @@ -446,24 +464,32 @@ public class DeviceTaskService { } public boolean cancelTask(DeviceTaskCommandDTO dto, String username) { - if (!StringUtils.isEmpty(dto.getDeviceId())) { - NettyDevice device = deviceSessionManager.getDevice(dto.getDeviceId()); - if (device != null) { - return device.cancelTask(); - } - } else { + String deviceId = dto.getDeviceId(); + DevicePlanTask devicePlanTask = null; + if (StringUtils.isEmpty(dto.getDeviceId())) { Long taskId = dto.getTaskId(); - // Long siteId = dto.getSiteId(); - // Long orgId = dto.getOrgId(); - DevicePlanTask devicePlanTask = devicePlanTaskMapper.selectById(taskId); + devicePlanTask = devicePlanTaskMapper.selectById(taskId); if (devicePlanTask != null) { - NettyDevice device = deviceSessionManager.getDevice(devicePlanTask.getDeviceId()); - if (device != null) { - return device.cancelTask(); - } + deviceId = devicePlanTask.getDeviceId(); + } + } + NettyDevice device = deviceSessionManager.getDevice(deviceId); + if (device != null && device.getTask() != null) { + return device.cancelTask(); + } else { + if (devicePlanTask != null) { + devicePlanTask.setTaskStaus(DeviceTaskStaus.CANCELED); + devicePlanTask.setUpdateTime(LocalDateTime.now()); + devicePlanTaskMapper.saveOrUpdate(devicePlanTask); + SiteMemory siteMemory = GlobalMemory.getSiteMemory(devicePlanTask.getOrgId(), + devicePlanTask.getSiteId()); + siteMemory.removeDevicePlanPrepare(devicePlanTask); + siteMemory.removeDevicePlanExecute(devicePlanTask.getDeviceId()); + return true; + } else { + return false; } } - return false; } // todo 完善 diff --git a/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java b/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java index 2967922..01d0255 100644 --- a/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java +++ b/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java @@ -136,13 +136,13 @@ public class TransferDeviceService { if (device == null) { return AjaxResult.error("设备不存在"); } else { - Long tenantId = device.getTenantId(); - if (tenantId == -1) { - return AjaxResult.error("设备未绑定"); - } - if (!tenantId.equals(user.getUserId())) { - return AjaxResult.error("无操作权限"); - } + // Long tenantId = device.getTenantId(); + // if (tenantId == -1) { + // return AjaxResult.error("设备未绑定"); + // } + // if (!tenantId.equals(user.getUserId())) { + // return AjaxResult.error("无操作权限"); + // } device.setDeviceAlias(bindDTO.getDeviceAlias()); device.setUpdateTime(LocalDateTime.now()); device.setCreateBy(user.getUsername()); diff --git a/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/WorkRecordController.java b/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/WorkRecordController.java index c6f842e..723fa29 100644 --- a/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/WorkRecordController.java +++ b/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/WorkRecordController.java @@ -1,21 +1,24 @@ package com.maibu.controller; - import com.maibu.core.business.WorkRecord; +import com.maibu.core.business.path.LatAndLngEntity; +import com.maibu.core.controller.BaseController; import com.maibu.core.domain.R; import com.maibu.service.impl.WorkRecordService; import com.maibu.utils.MinioUtil; +import com.maibu.utils.json.JsonUtils; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; +import org.springframework.http.MediaType; import java.io.InputStream; import java.util.List; - @RestController @RequestMapping("/iot/workRecord") -public class WorkRecordController { +public class WorkRecordController extends BaseController { @Autowired private MinioUtil minioUtil; @@ -23,15 +26,16 @@ public class WorkRecordController { @Autowired private WorkRecordService workRecordService; - - @PostMapping("/add") + @PostMapping(value = "/add", consumes = MediaType.MULTIPART_FORM_DATA_VALUE) public R addWorkRecord( - @RequestPart("file") MultipartFile file, // 接收上传的图片 + @RequestPart("file") MultipartFile file, // 接收上传的图片 @RequestPart("workRecord") WorkRecord workRecord // 接收表单里的 JSON 对象 ) { int i = -1; - WorkRecord workRecord1 = workRecordService.selectWorkRecordByWorkName(workRecord.getWorkName()); + // WorkRecord workRecordObj = JsonUtils.parseObject(workRecord, WorkRecord.class); + + WorkRecord workRecord1 = workRecordService.selectWorkRecordByWorkNameAndSiteId(workRecord.getWorkName(), workRecord.getSiteId()); if (workRecord1 != null) { return R.fail("该任务已存在"); } @@ -49,7 +53,8 @@ public class WorkRecordController { // 这里可以把 URL 存到 workRecord 里 workRecord.setImgUrl(url); - + workRecord.setUserId(getUserId()); + workRecord.setOrgId(getLoginUser().getOrgId()); i = workRecordService.addWorkRecord(workRecord); } catch (Exception e) { return R.fail("新增失败: " + e.getMessage()); @@ -58,7 +63,6 @@ public class WorkRecordController { return i <= 0 ? R.fail("新增失败") : R.ok("新增成功"); } - @GetMapping("/selectByUserId") public R selectWorkRecordByUserId(String userId) { List workRecords = workRecordService.selectWorkRecordByUserId(userId); @@ -89,4 +93,16 @@ public class WorkRecordController { return R.ok(workRecords); } + @GetMapping("/selectById") + public R selectById(@RequestParam Long id) { + WorkRecord workRecord = workRecordService.selectById(id); + return R.ok(workRecord); + } + + @GetMapping("/selectByTaskId") + public R selectByTaskId(@RequestParam Long taskId) { + List path = workRecordService.selectByTaskId(taskId); + return R.ok(path); + } + } diff --git a/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceServiceImpl.java b/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceServiceImpl.java index 6692574..f5e2ed3 100644 --- a/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceServiceImpl.java +++ b/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceServiceImpl.java @@ -1,7 +1,30 @@ package com.maibu.service.impl; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import javax.annotation.Resource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cache.annotation.Cacheable; +import org.springframework.stereotype.Service; +import org.springframework.util.CollectionUtils; + import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; -import com.maibu.core.business.*; +import com.influxdb.query.FluxRecord; +import com.influxdb.query.FluxTable; +import com.maibu.core.business.Device; +import com.maibu.core.business.DevicePlan; +import com.maibu.core.business.DeviceRunningStatusHistory; +import com.maibu.core.business.DeviceStatusRecordDTO; import com.maibu.core.business.dto.DeviceBindDTO; import com.maibu.core.domain.AjaxResult; import com.maibu.core.domain.entity.SysRole; @@ -9,6 +32,9 @@ import com.maibu.core.domain.entity.SysSite; import com.maibu.core.domain.model.LoginUser; import com.maibu.core.enums.DeviceTaskStaus; import com.maibu.core.redis.RedisCache; +import com.maibu.influxdb.MowerRealTimeData; +import com.maibu.influxdb.util.InfluxSqlBuilder; +import com.maibu.influxdb.util.LambdaUtils; import com.maibu.mapper.DeviceMapper; import com.maibu.mapper.DevicePlanMapper; import com.maibu.mapper.DeviceStatusHistoryMapper; @@ -19,19 +45,6 @@ import com.maibu.service.IDeviceService; import com.maibu.service.ISysRoleService; import com.maibu.utils.SecurityUtils; import com.maibu.utils.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.cache.annotation.Cacheable; -import org.springframework.stereotype.Service; -import org.springframework.util.CollectionUtils; - -import javax.annotation.Resource; -import java.time.LocalDateTime; -import java.util.*; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.stream.Collectors; - /** * 设备Service业务层处理 @@ -59,7 +72,6 @@ public class DeviceServiceImpl implements IDeviceService { @Autowired private SysSiteMapper sysSiteMapper; - /** * 根据设备编号查询设备 * @@ -88,7 +100,6 @@ public class DeviceServiceImpl implements IDeviceService { return device; } - /** * 查询设备列表 * @@ -130,46 +141,38 @@ public class DeviceServiceImpl implements IDeviceService { .map(DevicePlan::getDeviceId) .collect(Collectors.toSet()); - Map> statusMap = - allStatusList.stream() - .collect(Collectors.groupingBy(DeviceRunningStatusHistory::getDeviceId)); + Map> statusMap = allStatusList.stream() + .collect(Collectors.groupingBy(DeviceRunningStatusHistory::getDeviceId)); deviceList.forEach(x -> { Integer s = x.getOnlineStatus() == null ? 0 : x.getOnlineStatus(); if (0 == s) { - x.setFeStatus(3); //离线 + x.setFeStatus(3); // 离线 } else { if (!CollectionUtils.isEmpty(taskDeviceIds) && taskDeviceIds.contains(x.getDeviceName())) { x.setFeStatus(1); } else { - x.setFeStatus(2); //空闲 + x.setFeStatus(2); // 空闲 } -// if (!CollectionUtils.isEmpty(planTasks)) { -// long count = planTasks.stream().filter(y -> y.getDeviceId().equals(x.getDeviceName())).count(); -// if (count > 0) { -// x.setFeStatus(1); //任务中 -// } else { -// x.setFeStatus(2); //空闲 -// } -// } else { -// x.setFeStatus(2); //空闲 -// } } - //经纬度 + // 经纬度 String key = "running_status:" + x.getDeviceName(); DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(key); DeviceRunningStatusHistory lastRunningStatus = null; if (recordDTO != null && !CollectionUtils.isEmpty(recordDTO.getHistories())) { - List list = recordDTO.getHistories().stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList()); + List list = recordDTO.getHistories().stream() + .filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList()); if (!CollectionUtils.isEmpty(list)) { lastRunningStatus = list.get(list.size() - 1); } } else { if (!CollectionUtils.isEmpty(allStatusList)) { List list = statusMap.get(x.getDeviceName()); -// List list = allStatusList.stream().filter(z->z.getDeviceId().equals(x.getDeviceName())).collect(Collectors.toList()); + // List list = + // allStatusList.stream().filter(z->z.getDeviceId().equals(x.getDeviceName())).collect(Collectors.toList()); if (!CollectionUtils.isEmpty(list)) { - List l = list.stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList()); + List l = list.stream().filter(y -> !"0".equals(y.getQual())) + .collect(Collectors.toList()); if (!CollectionUtils.isEmpty(l)) { lastRunningStatus = l.get(l.size() - 1); } @@ -183,7 +186,6 @@ public class DeviceServiceImpl implements IDeviceService { return deviceList; } - @Override public List selectSerialNumberByProductId(Long productId) { return deviceMapper.selectSerialNumberByProductId(productId); @@ -205,7 +207,6 @@ public class DeviceServiceImpl implements IDeviceService { return deviceMapper.getDeviceNumsByProductId(productId); } - /** * 重置设备状态 * @@ -227,7 +228,6 @@ public class DeviceServiceImpl implements IDeviceService { return deviceMapper.selectSerialNumbersByProductId(productId); } - @Override public boolean bind(DeviceBindDTO bindDTO, String username) { try { @@ -288,7 +288,7 @@ public class DeviceServiceImpl implements IDeviceService { String roleKey = sysRole.getRoleKey(); if (!StringUtils.isEmpty(roleKey)) { if ("manager".equals(roleKey) || "siteManager".equals(roleKey) || "admin".equals(roleKey)) { -// device.setSiteId(siteId); + // device.setSiteId(siteId); } else { device.setTenantId(loginUser.getUserId()); } @@ -321,19 +321,20 @@ public class DeviceServiceImpl implements IDeviceService { DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(key); DeviceRunningStatusHistory lastRunningStatus = null; if (recordDTO != null && !CollectionUtils.isEmpty(recordDTO.getHistories())) { - List list = recordDTO.getHistories().stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList()); + List list = recordDTO.getHistories().stream() + .filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList()); if (!CollectionUtils.isEmpty(list)) { lastRunningStatus = list.get(list.size() - 1); } } if (lastRunningStatus == null) { if (!CollectionUtils.isEmpty(allStatusList)) { - Map> statusMap = - allStatusList.stream() - .collect(Collectors.groupingBy(DeviceRunningStatusHistory::getDeviceId)); + Map> statusMap = allStatusList.stream() + .collect(Collectors.groupingBy(DeviceRunningStatusHistory::getDeviceId)); List list = statusMap.get(x.getDeviceName()); if (!CollectionUtils.isEmpty(list)) { - List l = list.stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList()); + List l = list.stream().filter(y -> !"0".equals(y.getQual())) + .collect(Collectors.toList()); if (!CollectionUtils.isEmpty(l)) { lastRunningStatus = l.get(l.size() - 1); } @@ -367,16 +368,15 @@ public class DeviceServiceImpl implements IDeviceService { } else if (!"admin".equals(roleKey)) { device.setTenantId(loginUser.getUserId()); } -// if ("admin".equals(roleKey)) { -// deviceList = deviceMapper.selectList(); -// } else { -// deviceList = deviceMapper.selectDeviceList(device); -// } + // if ("admin".equals(roleKey)) { + // deviceList = deviceMapper.selectList(); + // } else { + // deviceList = deviceMapper.selectDeviceList(device); + // } deviceList = deviceMapper.selectDeviceList(device); } } return deviceList; } - } diff --git a/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/WorkRecordService.java b/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/WorkRecordService.java index 1d59655..6e8db3b 100644 --- a/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/WorkRecordService.java +++ b/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/WorkRecordService.java @@ -1,10 +1,14 @@ package com.maibu.service.impl; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.maibu.core.business.DevicePlanTask; import com.maibu.core.business.WorkRecord; +import com.maibu.core.business.path.LatAndLngEntity; +import com.maibu.mapper.DevicePlanTaskMapper; import com.maibu.mapper.WorkRecordMapper; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import org.springframework.util.CollectionUtils; import java.util.List; @@ -14,10 +18,12 @@ public class WorkRecordService { @Autowired private WorkRecordMapper workRecordMapper; + @Autowired + private DevicePlanTaskMapper devicePlanTaskMapper; public int addWorkRecord(WorkRecord workRecord) throws Exception { int i = workRecordMapper.insert(workRecord); -// int i = workRecordMapper.insertWorkRecord(workRecord); + // int i = workRecordMapper.insertWorkRecord(workRecord); return i; } @@ -27,28 +33,55 @@ public class WorkRecordService { return workRecordMapper.selectOne(queryWrapper); } + public WorkRecord selectWorkRecordByWorkNameAndSiteId(String workName, Long siteId) { + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(WorkRecord::getWorkName, workName); + queryWrapper.eq(WorkRecord::getSiteId, siteId); + return workRecordMapper.selectOne(queryWrapper); + } + public List selectWorkRecordByUserId(String userId) { LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); queryWrapper.eq(WorkRecord::getUserId, userId); return workRecordMapper.selectList(queryWrapper); } - public int deleteWorkRecordByWorkName(String workName) { LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); queryWrapper.eq(WorkRecord::getWorkName, workName); return workRecordMapper.delete(queryWrapper); } - public boolean updateWorkRecord(WorkRecord workRecord) { return workRecordMapper.saveOrUpdate(workRecord); } - public List selectBySiteId(Long siteId) { LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); queryWrapper.eq(WorkRecord::getSiteId, siteId); - return workRecordMapper.selectList(queryWrapper); + List list = workRecordMapper.selectList(queryWrapper); + if (!CollectionUtils.isEmpty(list)) { + list.forEach(workRecord -> { + workRecord.setJsonData(null); + }); + } + return list; + } + + public WorkRecord selectById(Long id) { + return workRecordMapper.selectById(id); + } + + public List selectByTaskId(Long taskId) { + DevicePlanTask devicePlanTask = devicePlanTaskMapper.selectById(taskId); // 先查询设备任务，确认任务存在 + if (devicePlanTask == null) { + return null; + } + Long routeId = devicePlanTask.getRouteId(); // 获取任务关联的路线ID + WorkRecord workRecord = workRecordMapper.selectById(routeId); // 查询路线ID对应的工作记录 + if (workRecord != null && workRecord.getJsonData() != null) { + return workRecord.getJsonData().getPath(); // 返回路线数据 + } + return null; // 如果没有找到对应的工作记录或路线数据，返回 null } } diff --git a/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java b/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java index 8fdd8cc..2a80e47 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java +++ b/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java @@ -3,6 +3,8 @@ package com.maibu.netty.handler; import cn.hutool.json.JSONObject; import com.maibu.common.MiddleCommandConstant; import com.maibu.common.MiddleConstant; +import com.maibu.core.business.device.DeviceStatusDetail; +import com.maibu.core.business.inter.WebsocketMesDispather; import com.maibu.core.business.path.LatAndLngEntity; import com.maibu.core.enums.MesType; import com.maibu.core.enums.RespondCode; @@ -37,6 +39,8 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { private static final HttpService httpService = SpringUtils.getBean(HttpService.class); + private final WebsocketMesDispather websocketMesDispather = SpringUtils.getBean(WebsocketMesDispather.class); + @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { @@ -113,7 +117,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { dto.setTaskId(Long.valueOf(taskId)); dto.setStatus(status); WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto)); - }else if ("device_task_arrive_point".equals(eventType)) { + } else if ("device_task_arrive_point".equals(eventType)) { String taskId = obj.get("taskId").toString(); LatAndLngEntity position = JsonUtils.parseObject(obj.get("entity").toString(), LatAndLngEntity.class); @@ -179,13 +183,14 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { } } else if (isStatusDate(data)) { // 状态消息推送给前端 - WebStatusMessageDTO dto = createWebDeviceStatusMessage(data); - WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto)); + // WebStatusMessageDTO dto = createWebDeviceStatusMessage(data); + // WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto)); + + // String[] k = key.split(":"); + // String deviceId = k[0] + ":" + k[1]; + // websocketMesDispather.dispather(deviceId,JsonUtils.toJsonString(dto)); } else if (isPath(data)) { // 已到达 - if (data[5] == (byte) 0x01) { - - } } } catch (Exception e) { logger.error("channelRead error:{}", e.getMessage());