IOT/MiddlePlatform
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
rqian
2026-06-17 09:14:42 +08:00
parent 3879cd0067
commit a68a670ecb
27 changed files with 1053 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
maibu-admin/src/main/resources/application-dev.yml
View File

@@ -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

7
maibu-common/src/main/java/com/maibu/core/business/Device.java
View File

@@ -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<Integer,Integer> statusStatistics;

3
maibu-common/src/main/java/com/maibu/core/business/WorkRecord.java
View File

@@ -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)

2
maibu-web-middleware/src/main/java/com/maibu/dto/DeviceStatusDetail.java → maibu-common/src/main/java/com/maibu/core/business/device/DeviceStatusDetail.java
View File

@@ -1,4 +1,4 @@
package com.maibu.dto;
package com.maibu.core.business.device;
import lombok.Data;

8
maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java
View File

@@ -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 {

29
maibu-common/src/main/java/com/maibu/influxdb/DeviceData.java
View File

@@ -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;
}

155
maibu-common/src/main/java/com/maibu/influxdb/MowerRealTimeData.java Normal file
View File

@@ -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;
}
}

65
maibu-common/src/main/java/com/maibu/influxdb/test.java
View File

@@ -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<String, String> map = InfluxSqlBuilder.buildTagMap(data,DeviceData::getDeviceId);
String flux = InfluxSqlBuilder.buildQuery(bucket,bucket,10,map,"temperature");
List<FluxTable> tables = dbUtil.query(flux);
for (FluxTable table : tables) {
for (FluxRecord record : table.getRecords()) {
System.out.println(record.getValue());
}
}
}
}

22
maibu-common/src/main/java/com/maibu/influxdb/util/InfluxDBUtil.java
View File

@@ -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<String> 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<FluxTable> query(String flux) {
QueryApi queryApi = client.getQueryApi();
@@ -69,7 +69,7 @@ public class InfluxDBUtil {
}
// =========================
// ✅ 查询（转List<Map>）
// 查询（转List<Map>）
// =========================
public List<String> queryValues(String flux) {
List<String> resultList = new ArrayList<>();
@@ -86,14 +86,14 @@ public class InfluxDBUtil {
}
// =========================
// ✅ 手动flush（重要）
// 手动flush（重要）
// =========================
public void flush() {
writeApi.flush();
}
// =========================
// ✅ 关闭资源（必须）
// 关闭资源（必须）
// =========================
public void close() {
try {

122
maibu-common/src/main/java/com/maibu/influxdb/util/InfluxSqlBuilder.java
View File

@@ -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<String, String> 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<String, String> 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<String, String> tagFilters,
String field
) {
long startHoursAgo,
Map<String, ? extends Collection<String>> 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<String, String> entry : tagFilters.entrySet()) {
flux.append(" |> filter(fn: (r) => r.")
for (Map.Entry<String, ? extends Collection<String>> entry : tagFilters.entrySet()) {
Collection<String> 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(entry.getValue())
.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 <T> Map<String, String> buildTagMap(
T obj,
SFunction<T, ?>... functions
) {
SFunction<T, ?>... functions) {
Map<String, String> map = new HashMap<>();
try {
@@ -190,5 +273,4 @@ public class InfluxSqlBuilder {
return value.toString(); // double/float
}
}

14
maibu-common/src/main/java/com/maibu/influxdb/util/LambdaUtils.java
View File

@@ -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 <T> String getFieldName(SFunction<T, ?> 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();
}
}

31
maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java
View File

@@ -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<Long, IoTCommonDevice> unRegisterCommonDeviceMap = new ConcurrentHashMap<>();
//
// public static ConcurrentHashMap<Long, Device> unRegisterDeviceMap = new ConcurrentHashMap<>();
// orgId - siteId - deviceId
public static ConcurrentHashMap<String, IoTCommonDevice> unRegisterCommonDeviceMap = new ConcurrentHashMap<>();
@@ -35,6 +31,9 @@ public class GlobalMemory {
public static ConcurrentHashMap<Long, IoTCommonProduct> 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,6 +55,7 @@ public class GlobalMemory {
/**
* 新增场站内存
*
* @param orgId
* @param siteId
* @param siteMemory
@@ -75,6 +75,7 @@ public class GlobalMemory {
/**
* 删除
*
* @param orgId
* @param siteId
*/

25
maibu-common/src/main/java/com/maibu/memory/SiteMemory.java
View File

@@ -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

135
maibu-common/src/main/java/com/maibu/mqtt/MqttClientUtil.java Normal file
View File

@@ -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();
}
}

10
maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java Normal file
View File

@@ -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"; // 任务到达
}

9
maibu-netty-server/src/main/java/com/maibu/dto/DeviceTaskQueryDTO.java
View File

@@ -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;

1
maibu-web-middleware/src/main/java/com/maibu/dto/WebStatusMessageDTO.java → maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java
View File

@@ -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

89
maibu-netty-server/src/main/java/com/maibu/init/InitThread.java
View File

@@ -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,14 +59,47 @@ 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 {
@@ -108,7 +144,8 @@ public class InitThread implements ApplicationRunner {
}
// 未执行完的重复计划
LambdaQueryWrapperX<DevicePlan> query = new LambdaQueryWrapperX<>();
List<String> list = Arrays.asList(DeviceTaskStaus.NEW.getCode(), DeviceTaskStaus.EXECUTING.getCode(), DeviceTaskStaus.PAUSE.getCode());
List<String> list = Arrays.asList(DeviceTaskStaus.NEW.getCode(), DeviceTaskStaus.EXECUTING.getCode(),
DeviceTaskStaus.PAUSE.getCode());
query.in(DevicePlan::getTaskStaus, list);
List<DevicePlan> devicePlanList = devicePlanMapper.selectList(query);
if (!CollectionUtils.isEmpty(devicePlanList)) {

269
maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java
View File

@@ -1,47 +1,69 @@
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<String> {
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 {
try {
Channel channel = ctx.channel();
String deviceID = deviceSessionManager.getDeviceIdByChannel(channel);
if (StringUtils.isEmpty(deviceID)) return;
if (StringUtils.isEmpty(deviceID))
return;
byte[] bytes = Hex.decodeHex(msg.toCharArray());
if (bytes[2] == CommandConstant.status) {
@@ -61,9 +83,25 @@ public class DataToDataBaseHandler extends SimpleChannelInboundHandler<String> {
}
}
if (record) {
// 推送websocket实时消息给前端 todo mqtt推送
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<NettyDevice> 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<>();
@@ -71,8 +109,13 @@ public class DataToDataBaseHandler extends SimpleChannelInboundHandler<String> {
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);
@@ -92,11 +135,199 @@ public class DataToDataBaseHandler extends SimpleChannelInboundHandler<String> {
latestStatus.setWorkArea(Double.valueOf(workArea));
nettyDevice.setLatestStatus(latestStatus);
}
// else{
// System.out.println(false);
// }
}
ctx.fireChannelRead(msg);
} catch (Exception e) {
logger.error("处理数据异常: {}", e.getMessage());
}
}
public WebStatusMessageDTO createWebDeviceStatusMessage(String[] split) {
WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO();
List<DeviceStatusDetail> transferData = transferStatusData(split);
webStatusMessageDTO.setData(transferData);
webStatusMessageDTO.setType(MesType.deviceInfo);
return webStatusMessageDTO;
}
private List<DeviceStatusDetail> transferStatusData(String[] split) {
List<DeviceStatusDetail> 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) {

43
maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java
View File

@@ -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<String> {
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<NettyDevice> controlMasters = sessionManager
.getAllSlaveControl(deviceId);
if (!CollectionUtils.isEmpty(controlMasters)) {
@@ -373,14 +375,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
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<String> {
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<NettyDevice> controlMasters = sessionManager
.getAllSlaveControl(deviceId);
if (!CollectionUtils.isEmpty(controlMasters)) {
@@ -404,12 +405,6 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
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);
// }
});
}
}

1
maibu-netty-server/src/main/java/com/maibu/scheduled/ScheduledTask.java
View File

@@ -44,6 +44,7 @@ public class ScheduledTask {
// recordDTO.getHistories().forEach(x -> {
// deviceStatusHistoryMapper.insert(x);
// });
deviceStatusHistoryMapper.insertBatch(recordDTO.getHistories());
}
redisCache.deleteObject(k);

54
maibu-netty-server/src/main/java/com/maibu/service/DeviceTaskService.java
View File

@@ -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,25 +464,33 @@ 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) {
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;
}
}
}
// todo 完善
public boolean pauseTask(DeviceTaskCommandDTO dto, String username) {

14
maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java
View File

@@ -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());

32
maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/WorkRecordController.java
View File

@@ -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("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<WorkRecord> 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<LatAndLngEntity> path = workRecordService.selectByTaskId(taskId);
return R.ok(path);
}
}

78
maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceServiceImpl.java
View File

@@ -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,8 +141,7 @@ public class DeviceServiceImpl implements IDeviceService {
.map(DevicePlan::getDeviceId)
.collect(Collectors.toSet());
Map<String, List<DeviceRunningStatusHistory>> statusMap =
allStatusList.stream()
Map<String, List<DeviceRunningStatusHistory>> statusMap = allStatusList.stream()
.collect(Collectors.groupingBy(DeviceRunningStatusHistory::getDeviceId));
deviceList.forEach(x -> {
@@ -144,32 +154,25 @@ public class DeviceServiceImpl implements IDeviceService {
} else {
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<DeviceRunningStatusHistory> list = recordDTO.getHistories().stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList());
List<DeviceRunningStatusHistory> 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<DeviceRunningStatusHistory> list = statusMap.get(x.getDeviceName());
// List<DeviceRunningStatusHistory> list = allStatusList.stream().filter(z->z.getDeviceId().equals(x.getDeviceName())).collect(Collectors.toList());
// List<DeviceRunningStatusHistory> list =
// allStatusList.stream().filter(z->z.getDeviceId().equals(x.getDeviceName())).collect(Collectors.toList());
if (!CollectionUtils.isEmpty(list)) {
List<DeviceRunningStatusHistory> l = list.stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList());
List<DeviceRunningStatusHistory> 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<String> 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 {
@@ -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<DeviceRunningStatusHistory> list = recordDTO.getHistories().stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList());
List<DeviceRunningStatusHistory> 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<String, List<DeviceRunningStatusHistory>> statusMap =
allStatusList.stream()
Map<String, List<DeviceRunningStatusHistory>> statusMap = allStatusList.stream()
.collect(Collectors.groupingBy(DeviceRunningStatusHistory::getDeviceId));
List<DeviceRunningStatusHistory> list = statusMap.get(x.getDeviceName());
if (!CollectionUtils.isEmpty(list)) {
List<DeviceRunningStatusHistory> l = list.stream().filter(y -> !"0".equals(y.getQual())).collect(Collectors.toList());
List<DeviceRunningStatusHistory> l = list.stream().filter(y -> !"0".equals(y.getQual()))
.collect(Collectors.toList());
if (!CollectionUtils.isEmpty(l)) {
lastRunningStatus = l.get(l.size() - 1);
}
@@ -378,5 +379,4 @@ public class DeviceServiceImpl implements IDeviceService {
return deviceList;
}
}

41
maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/WorkRecordService.java
View File

@@ -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,6 +18,8 @@ public class WorkRecordService {
@Autowired
private WorkRecordMapper workRecordMapper;
@Autowired
private DevicePlanTaskMapper devicePlanTaskMapper;
public int addWorkRecord(WorkRecord workRecord) throws Exception {
int i = workRecordMapper.insert(workRecord);
@@ -27,28 +33,55 @@ public class WorkRecordService {
return workRecordMapper.selectOne(queryWrapper);
}
public WorkRecord selectWorkRecordByWorkNameAndSiteId(String workName, Long siteId) {
LambdaQueryWrapper<WorkRecord> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(WorkRecord::getWorkName, workName);
queryWrapper.eq(WorkRecord::getSiteId, siteId);
return workRecordMapper.selectOne(queryWrapper);
}
public List<WorkRecord> selectWorkRecordByUserId(String userId) {
LambdaQueryWrapper<WorkRecord> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(WorkRecord::getUserId, userId);
return workRecordMapper.selectList(queryWrapper);
}
public int deleteWorkRecordByWorkName(String workName) {
LambdaQueryWrapper<WorkRecord> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(WorkRecord::getWorkName, workName);
return workRecordMapper.delete(queryWrapper);
}
public boolean updateWorkRecord(WorkRecord workRecord) {
return workRecordMapper.saveOrUpdate(workRecord);
}
public List<WorkRecord> selectBySiteId(Long siteId) {
LambdaQueryWrapper<WorkRecord> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(WorkRecord::getSiteId, siteId);
return workRecordMapper.selectList(queryWrapper);
List<WorkRecord> 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<LatAndLngEntity> 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
}
}

15
maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java
View File

@@ -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 {
@@ -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());