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-08-19 11:02:52 +08:00
parent ffd9720939
commit 1b19e32aa5
30 changed files with 629 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
maibu-common/src/main/java/com/maibu/config/DeviceTask.java
View File

@@ -97,6 +97,11 @@ public class DeviceTask {
return builder(FastBeeConstant.TASK.DEVICE_ERROR_MONITOR);
}
@Bean(FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR)
public Executor deviceHeartBeatMonitor() {
return builder(FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR);
}
@Bean(FastBeeConstant.TASK.DEVICE_TASK_HANDLER)
public Executor deviceTaskHandler() {
return builder(FastBeeConstant.TASK.DEVICE_TASK_HANDLER);

6
maibu-common/src/main/java/com/maibu/constant/Constant.java
View File

@@ -21,9 +21,11 @@ public class Constant {
public static final AttributeKey<String> ATT_SITE_ID = AttributeKey.valueOf("siteId");
public static final String WheelGainCoefficients = "WheelGainCoefficients";
public static final String WheelGainCoefficients = "WheelGainCoefficients";
public static final String mqttClientPrefix = "common-device-client-";
public static final String mqttClientPrefix = "common-device-client-";
public static final String mqttHandleKey = "hostHandle";
public static final Long hostOfflineInterval = 10000L; //单位毫秒
}

5
maibu-common/src/main/java/com/maibu/constant/FastBeeConstant.java
View File

@@ -100,6 +100,11 @@ public interface FastBeeConstant {
* 设备错误
*/
String DEVICE_ERROR_MONITOR = "deviceErrorMonitor";
/**
* 设备错误
*/
String DEVICE_HEARTBEAT_MONITOR = "deviceHeartBeatMonitor";
/**
* 设备任务处理线程
*/

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

@@ -20,6 +20,8 @@ public class WorkRecord extends OrgBaseDO {
private String workName;
private Long userId;
private String mode;//导航任务模式 single_point|coverage
// private String jsonData;
@TableField(typeHandler = JacksonTypeHandler.class)

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

@@ -1,35 +1,35 @@
package com.maibu.core.business.device;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.time.LocalDateTime;
import java.util.LinkedList;
import java.util.Queue;
import com.maibu.utils.json.JsonUtils;
import org.springframework.util.CollectionUtils;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.maibu.core.business.Device;
import com.maibu.core.business.DevicePlanTask;
import com.maibu.core.business.DeviceRunParam;
import com.maibu.core.business.DeviceRunStatistics;
import com.maibu.core.business.PlanPath;
import com.maibu.core.business.WorkRecord;
import com.maibu.core.business.*;
import com.maibu.core.business.path.LatAndLngEntity;
import com.maibu.core.business.path.RoutePlanSendEntity;
import com.maibu.core.enums.ConnectorType;
import com.maibu.core.enums.DeviceTaskStaus;
import com.maibu.core.host.HostWorkRecordCreateDTO;
import com.maibu.core.host.MutiPointEntity;
import com.maibu.core.host.PointEntity;
import com.maibu.mapper.DevicePlanTaskMapper;
import com.maibu.mapper.WorkRecordMapper;
import com.maibu.memory.GlobalMemory;
import com.maibu.memory.SiteMemory;
import com.maibu.mqtt.MqttTopic;
import com.maibu.utils.StringUtils;
import com.maibu.utils.json.JsonUtils;
import com.maibu.utils.spring.SpringUtils;
import io.netty.buffer.Unpooled;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
@Slf4j
@EqualsAndHashCode(callSuper = true)
@@ -67,6 +67,8 @@ public class NettyDevice extends Connector {
@JsonIgnore
private WorkRecordMapper workRecordMapper = SpringUtils.getBean(WorkRecordMapper.class);
private static final String mqttClientId = "maibu-mqtt-client";
public NettyDevice(String connectorId) {
super(connectorId);
this.deviceType = ConnectorType.MASTER; // 默认上位机角色
@@ -77,6 +79,11 @@ public class NettyDevice extends Connector {
this.deviceType = deviceType;
}
public void status(Integer onlineStatus){
this.onlineStatus = onlineStatus;
}
// todo 如何确定命令已收到？
public void sendNextPoint() {
if (task.getTaskStaus().equals(DeviceTaskStaus.PAUSE))
@@ -109,13 +116,14 @@ public class NettyDevice extends Connector {
}
public synchronized Long finishTask() {
sendPathCommand((byte) 0x01, (short) 0, 0.00d, 0.00d, (short) 0);
// sendPathCommand((byte) 0x01, (short) 0, 0.00d, 0.00d, (short) 0);
long finishId = task.getId();
task.setTaskStaus(DeviceTaskStaus.FINISH);
task.setFinishTime(LocalDateTime.now());
devicePlanTaskMapper.saveOrUpdate(task);
SiteMemory siteMemory = GlobalMemory.getSiteMemory(task.getOrgId(), task.getSiteId());
siteMemory.removeDevicePlanExecute(task.getDeviceId());
GlobalMemory.taskPathMap.remove(task.getId());
task = null;
currentTaskId = null;
currentPoint = null;
@@ -124,15 +132,58 @@ public class NettyDevice extends Connector {
return finishId;
}
public synchronized boolean cancelTask() {
// public synchronized boolean cancelTask() {
// if (task != null) {
// sendPathCommand((byte) 0x01, (short) 0, 0.00d, 0.00d, (short) 0);
// task.setTaskStaus(DeviceTaskStaus.CANCELED);
// task.setUpdateTime(LocalDateTime.now());
// devicePlanTaskMapper.saveOrUpdate(task);
// SiteMemory siteMemory = GlobalMemory.getSiteMemory(task.getOrgId(), task.getSiteId());
// siteMemory.removeDevicePlanPrepare(task);
// siteMemory.removeDevicePlanExecute(task.getDeviceId());
// task = null;
// currentTaskId = null;
// currentPoint = null;
// locationQueue.clear();
// executingTask = false;
// }
// return true;
// }
public synchronized boolean cancelingTask() throws MqttException {
if (task != null) {
WorkRecord workRecord = workRecordMapper.selectById(task.getRouteId());
if (workRecord != null) {
HostWorkRecordCreateDTO dto = new HostWorkRecordCreateDTO();
dto.setAction("cancel");
dto.setCommand_id(task.getId() + "_" + System.currentTimeMillis());
dto.setTask_id(String.valueOf(task.getId()));
dto.setSn(task.getDeviceId());
dto.setMode(workRecord.getMode());
dto.setTimestamp(System.currentTimeMillis());
dto.setMessage_id(task.getId() + "_" + System.currentTimeMillis());
dto.setProtocol_version("1.4");
GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.MOWER_TASK_ROUTE_SET, task.getDeviceId()),
JsonUtils.toJsonString(dto));
task.setTaskStaus(DeviceTaskStaus.CANCELING);
task.setUpdateTime(LocalDateTime.now());
devicePlanTaskMapper.saveOrUpdate(task);
}
}
return true;
}
public synchronized boolean canceledTask() {
if (task != null) {
sendPathCommand((byte) 0x01, (short) 0, 0.00d, 0.00d, (short) 0);
task.setTaskStaus(DeviceTaskStaus.CANCELED);
task.setUpdateTime(LocalDateTime.now());
devicePlanTaskMapper.saveOrUpdate(task);
SiteMemory siteMemory = GlobalMemory.getSiteMemory(task.getOrgId(), task.getSiteId());
siteMemory.removeDevicePlanPrepare(task);
siteMemory.removeDevicePlanExecute(task.getDeviceId());
GlobalMemory.taskPathMap.remove(task.getId());
task = null;
currentTaskId = null;
currentPoint = null;
@@ -238,28 +289,96 @@ public class NettyDevice extends Connector {
}
}
// public void startTask() {
// try {
// if (executingTask)
// return;
// if (task != null) {
// Long routeId = task.getRouteId();
// WorkRecord workRecord = workRecordMapper.selectById(routeId);
// if (workRecord != null) {
// PlanPath planPath = workRecord.getJsonData();
// if (planPath != null && !CollectionUtils.isEmpty(planPath.getPath())) {
// planPath.getPath().forEach(x -> {
// locationQueue.offer(x);
// });
// }
// }
// task.setStartTime(LocalDateTime.now());
// }
// executingTask = true;
// sendNextPoint();
// } catch (Exception e) {
// throw new RuntimeException(e);
// }
// }
public void startTask() {
try {
if (executingTask)
return;
if (task != null) {
Long routeId = task.getRouteId();
WorkRecord workRecord = workRecordMapper.selectById(routeId);
if (workRecord != null) {
PlanPath planPath = workRecord.getJsonData();
if (planPath != null && !CollectionUtils.isEmpty(planPath.getPath())) {
planPath.getPath().forEach(x -> {
locationQueue.offer(x);
});
}
}
task.setStartTime(LocalDateTime.now());
executingTask = true;
sendNavigationCommand();
}
executingTask = true;
sendNextPoint();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public void sendNavigationCommand() throws MqttException {
log.debug("发送开始任务 taskStatus:{},routeId:{}", task.getTaskStaus(), task.getRouteId());
if (!task.getTaskStaus().equals(DeviceTaskStaus.EXECUTING))
return;
WorkRecord workRecord = workRecordMapper.selectById(task.getRouteId());
log.debug("发送开始任务 workRecord {}", workRecord);
if (workRecord != null) {
HostWorkRecordCreateDTO dto = new HostWorkRecordCreateDTO();
dto.setAction("start");
dto.setCommand_id(task.getId() + "_" + System.currentTimeMillis());
dto.setTask_id(String.valueOf(task.getId()));
dto.setSn(task.getDeviceId());
dto.setMode(workRecord.getMode());
dto.setTimestamp(System.currentTimeMillis());
dto.setMessage_id(task.getId() + "_" + System.currentTimeMillis());
dto.setProtocol_version("1.4");
PointEntity point = new PointEntity();
if (!StringUtils.isEmpty(workRecord.getMode())) {
if (workRecord.getMode().equals("single_point")) {
point.setLongitude(workRecord.getJsonData().getPath().get(0).getLng());
point.setLatitude(workRecord.getJsonData().getPath().get(0).getLat());
}
if (workRecord.getMode().equals("coverage")) {
List<MutiPointEntity> boundary_points = new ArrayList<>();
// workRecord.getJsonData().getPath().forEach(x -> {
// MutiPointEntity mutiPoint = new MutiPointEntity();
// mutiPoint.setLongitude(x.getLng());
// mutiPoint.setLatitude(x.getLat());
// mutiPoint.setIndex();
// boundary_points.add(mutiPoint);
// });
List<LatAndLngEntity> path = workRecord.getJsonData().getPath();
for (int i = 0; i < path.size(); i++) {
LatAndLngEntity x = path.get(i);
MutiPointEntity mutiPoint = new MutiPointEntity();
mutiPoint.setLongitude(x.getLng());
mutiPoint.setLatitude(x.getLat());
mutiPoint.setIndex(i);
boundary_points.add(mutiPoint);
}
point.setBoundary_points(boundary_points);
}
}
dto.setData(point);
GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.MOWER_TASK_ROUTE_SET, task.getDeviceId()),
JsonUtils.toJsonString(dto));
}
}
}

2
maibu-common/src/main/java/com/maibu/core/business/dto/DeviceTaskCommandDTO.java
View File

@@ -20,4 +20,6 @@ public class DeviceTaskCommandDTO {
private Long taskId;
private boolean forceCancel = false; //强制取消 不等回复
}

13
maibu-common/src/main/java/com/maibu/core/enums/DeviceTaskStaus.java
View File

@@ -7,12 +7,13 @@ import lombok.Getter;
@AllArgsConstructor
public enum DeviceTaskStaus {
NEW("NEW","新建"),
EXECUTING("EXECUTING","执行中"),
PAUSE("PAUSE","暂停中"),
CANCELED("CANCELED","已取消"),
FINISH("FINISH","执行成功"),
FAILED("FAILED","执行失败");
NEW("NEW", "新建"),
EXECUTING("EXECUTING", "执行中"),
PAUSE("PAUSE", "暂停中"),
CANCELED("CANCELED", "已取消"),
CANCELING("CANCELING", "取消中"),
FINISH("FINISH", "执行成功"),
FAILED("FAILED", "执行失败");
private final String code;

12
maibu-common/src/main/java/com/maibu/core/host/HeartBeatDTO.java Normal file
View File

@@ -0,0 +1,12 @@
package com.maibu.core.host;
import lombok.Data;
import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@Data
public class HeartBeatDTO extends HostBase {
private String status;
private Long sysTimestamp; //服务器的时间
}

13
maibu-common/src/main/java/com/maibu/core/host/HostBase.java Normal file
View File

@@ -0,0 +1,13 @@
package com.maibu.core.host;
import lombok.Data;
@Data
public class HostBase {
private String protocol_version;
private String message_id;
private String sn;
private Long timestamp; //上位机的时间
}

13
maibu-common/src/main/java/com/maibu/core/host/HostLocationRelTimeDTO.java Normal file
View File

@@ -0,0 +1,13 @@
package com.maibu.core.host;
import lombok.Data;
import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@Data
public class HostLocationRelTimeDTO extends HostBase {
private LocationMessage data;
}

19
maibu-common/src/main/java/com/maibu/core/host/HostTaskStatusDTO.java Normal file
View File

@@ -0,0 +1,19 @@
package com.maibu.core.host;
import lombok.Data;
import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@Data
public class HostTaskStatusDTO extends HostBase{
private String task_id; //任务唯一编号
private String command_id;//指令唯一编号
private String mode;//导航任务模式 single_point|coverage
private String status;//任务动作 accepted|running|paused|completed|cancelled|failed
private HostTaskStatusDetailDTO data;
}

12
maibu-common/src/main/java/com/maibu/core/host/HostTaskStatusDetailDTO.java Normal file
View File

@@ -0,0 +1,12 @@
package com.maibu.core.host;
import lombok.Data;
@Data
public class HostTaskStatusDetailDTO {
private float progress;
private String error_code;
private String error_message;
}

13
maibu-common/src/main/java/com/maibu/core/host/HostVehicleRelTimeDTO.java Normal file
View File

@@ -0,0 +1,13 @@
package com.maibu.core.host;
import lombok.Data;
import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@Data
public class HostVehicleRelTimeDTO extends HostBase {
private VehicleMessage data;
}

20
maibu-common/src/main/java/com/maibu/core/host/HostWorkRecordCreateDTO.java Normal file
View File

@@ -0,0 +1,20 @@
package com.maibu.core.host;
import lombok.Data;
import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@Data
public class HostWorkRecordCreateDTO extends HostBase {
private String task_id; //任务唯一编号
private String command_id;//指令唯一编号
private String mode;//导航任务模式 single_point|coverage
private String action;//任务动作 start/pause/resume/cancel
// private String frame_id; //WGS84
private PointEntity data;
}

15
maibu-common/src/main/java/com/maibu/core/host/LocationMessage.java Normal file
View File

@@ -0,0 +1,15 @@
package com.maibu.core.host;
import lombok.Data;
@Data
public class LocationMessage {
private double latitude;
private double longitude;
private double yaw;//目标航向，正北0°、顺时针正、0-360°
private String fix_status;
private double altitude;
}

7
maibu-service/maibu-iot-service/src/main/java/com/maibu/dto/host/PointEntity.java → maibu-common/src/main/java/com/maibu/core/host/MutiPointEntity.java
View File

@@ -1,13 +1,14 @@
package com.maibu.dto.host;
package com.maibu.core.host;
import lombok.Data;
@Data
public class PointEntity {
public class MutiPointEntity {
private int index;//目标点序号
private int index;
private double latitude;
private double longitude;
private double yaw;//目标航向，正北0°、顺时针正、0-360°
}

15
maibu-common/src/main/java/com/maibu/core/host/VehicleMessage.java Normal file
View File

@@ -0,0 +1,15 @@
package com.maibu.core.host;
import lombok.Data;
@Data
public class VehicleMessage {
private double speed;
private double battery_percentage;//电量
private double battery_voltage;//电压
private String motion;//moving|stopped|blocked
private boolean emergency_stop; //急停状态
}

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

@@ -3,6 +3,7 @@ package com.maibu.memory;
import com.maibu.constant.Constant;
import com.maibu.core.business.Device;
import com.maibu.core.business.IoTCommonDevice;
import com.maibu.core.business.path.LatAndLngEntity;
import com.maibu.influxdb.util.InfluxDBUtil;
import com.maibu.mqtt.*;
import lombok.Data;
@@ -24,6 +25,8 @@ public class GlobalMemory {
// key：orgId - siteId
public static ConcurrentHashMap<Long, Map<Long, SiteMemory>> orgAttrs = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Long, List<LatAndLngEntity>> taskPathMap = new ConcurrentHashMap<>();
public static InfluxDBUtil influxDBUtil;
// public static MqttClientUtil mqttClientUtil;
@@ -101,11 +104,15 @@ public class GlobalMemory {
deviceConfig.setUsername(mqttUsername);
deviceConfig.setPassword(mqttPassword);
//TODO 添加所有需要订阅的
deviceConfig.setSubscribeTopics(Arrays.asList(MqttTopic.MOWER_WILDCARD_LOCATION));
deviceConfig.setSubscribeTopics(Arrays.asList(
MqttTopic.MOWER_HEARTBEAT,//心跳
MqttTopic.MOWER_WILDCARD_LOCATION,//位置 经纬度
MqttTopic.MOWER_WILDCARD_TASK_ROUTE, //路径回复
MqttTopic.MOWER_WILDCARD_ERROR,//错误
MqttTopic.MOWER_WILDCARD_REALTIME //状态消息
));
deviceConfig.setHandlerType(Constant.mqttHandleKey); //唯一的 用于处理上位机mqtt的
customMqttDeviceMonitor.registerDevice(deviceConfig);
}
}

26
maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java
View File

@@ -1,5 +1,6 @@
package com.maibu.mqtt;
import com.maibu.core.host.HeartBeatDTO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
@@ -66,6 +67,8 @@ public class CustomMqttDeviceMonitor {
*/
private ScheduledExecutorService scheduler;
private final ConcurrentHashMap<String, HeartBeatDTO> hearBeatMap = new ConcurrentHashMap<>();
/**
* 巡检间隔（秒）
*/
@@ -74,7 +77,7 @@ public class CustomMqttDeviceMonitor {
// ==================== 消息处理器管理 ====================
public CustomMqttMessageHandler getMessageHandler(String key){
public CustomMqttMessageHandler getMessageHandler(String key) {
return handlerMap.get(key);
}
@@ -152,6 +155,25 @@ public class CustomMqttDeviceMonitor {
return client != null && client.isConnected();
}
/**
* 保存心跳
*/
public void saveHeartBeat(String sn, HeartBeatDTO dto) {
hearBeatMap.put(sn, dto);
}
/**
* 获取心跳
*/
public HeartBeatDTO getHeartBeat(String sn) {
return hearBeatMap.get(sn);
}
public List<HeartBeatDTO> getAllHeartBeat() {
return (List<HeartBeatDTO>) hearBeatMap.values();
}
// ==================== 连接管理 ====================
/**
@@ -383,7 +405,7 @@ public class CustomMqttDeviceMonitor {
MqttMessage message = new MqttMessage(payload.getBytes());
message.setQos(qos);
client.publish(topic, message);
log.debug("发布消息: deviceKey={}, topic={}", deviceKey, topic);
log.debug("发布消息: deviceKey={}, topic={},payload:{}", deviceKey, topic, payload);
}
/**

82
maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java
View File

@@ -3,7 +3,18 @@ package com.maibu.mqtt;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.maibu.constant.Constant;
import com.maibu.core.business.device.NettyDevice;
import com.maibu.core.host.HeartBeatDTO;
import com.maibu.core.host.HostLocationRelTimeDTO;
import com.maibu.core.host.HostTaskStatusDTO;
import com.maibu.core.host.HostVehicleRelTimeDTO;
import com.maibu.memory.DeviceSessionManager;
import com.maibu.memory.GlobalMemory;
import com.maibu.utils.StringUtils;
import com.maibu.utils.json.JsonUtils;
import com.maibu.utils.spring.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
/**
@@ -14,6 +25,8 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
private final ObjectMapper objectMapper = new ObjectMapper();
private final DeviceSessionManager deviceSessionManager = SpringUtils.getBean(DeviceSessionManager.class);
@Override
public String getHandlerType() {
@@ -86,26 +99,15 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
}
/**
* 处理 GPS 位置数据上报
* topic: mower/{deviceId}/property/location/post
* 定位信息 经纬度
*/
private void handleLocationPost(String deviceId, String topic, String payload) {
try {
JsonNode jsonNode = objectMapper.readTree(payload);
log.debug("【位置数据】deviceId={}, payload={}", deviceId, payload);
HostLocationRelTimeDTO locationRelTimeDTO = JsonUtils.parseObject(payload, HostLocationRelTimeDTO.class);
double latitude = getDouble(jsonNode, "latitude", "lat");
double longitude = getDouble(jsonNode, "longitude", "lng", "lon");
double speed = getDouble(jsonNode, "speed");
double heading = getDouble(jsonNode, "heading", "direction");
double altitude = getDouble(jsonNode, "altitude", "alt");
int satellites = getInt(jsonNode, "satellites", "sats");
String fixQuality = getString(jsonNode, "fixQuality", "fix_quality");
Double batteryLevel = getDoubleObj(jsonNode, "batteryLevel", "battery");
log.info("【位置数据】deviceId={}, lat={}, lng={}, speed={}, heading={}, sats={}",
deviceId, latitude, longitude, speed, heading, satellites);
onLocationUpdate(deviceId, latitude, longitude, speed, heading, altitude, satellites, fixQuality, batteryLevel);
//todo 发送位置信息
} catch (Exception e) {
log.error("位置消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage());
@@ -113,11 +115,12 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
}
/**
* 处理实时运行数据
* topic: mower/{deviceId}/property/realtime/post
* 上位机数据
*/
private void handleRealtimePost(String deviceId, String topic, String payload) {
log.debug("【实时数据】deviceId={}, payload={}", deviceId, payload);
HostVehicleRelTimeDTO vehicleRelTimeDTO = JsonUtils.parseObject(payload, HostVehicleRelTimeDTO.class);
}
/**
@@ -134,6 +137,12 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
*/
private void handleHeartbeatPost(String deviceId, String topic, String payload) {
log.debug("【心跳】deviceId={}", deviceId);
// 心跳处理
HeartBeatDTO beatDTO = JsonUtils.parseObject(payload, HeartBeatDTO.class);
if (beatDTO != null) {
beatDTO.setSysTimestamp(System.currentTimeMillis());
GlobalMemory.customMqttDeviceMonitor.saveHeartBeat(beatDTO.getSn(), beatDTO);
}
}
/**
@@ -147,8 +156,38 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
/**
* 处理任务相关上报
*/
private void handleTaskPost(String deviceId, String topic, String payload, String action) {
private void handleTaskPost(String deviceId, String topic, String payload, String action) throws MqttException {
log.info("【任务上报】deviceId={}, action={}, payload={}", deviceId, action, payload);
HostTaskStatusDTO dto = JsonUtils.parseObject(payload, HostTaskStatusDTO.class);
if (dto != null && !StringUtils.isEmpty(dto.getStatus())) {
NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
if ("completed".equals(dto.getStatus())) {
nettyDevice.finishTask();
}
if ("cancelled".equals(dto.getStatus())) {
nettyDevice.canceledTask();
}
// if("paused".equals(dto.getStatus())){
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
// nettyDevice.finishTask();
// }
// if("failed".equals(dto.getStatus())){
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
// nettyDevice.finishTask();
// }
// if("accepted".equals(dto.getStatus())){
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
// nettyDevice.finishTask();
// }
// if("running".equals(dto.getStatus())){
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
// nettyDevice.finishTask();
// }
}
}
/**
@@ -165,13 +204,6 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
log.debug("【控制响应】deviceId={}, payload={}", deviceId, payload);
}
/**
* 位置数据回调，子类可覆盖实现业务逻辑
*/
protected void onLocationUpdate(String deviceId, double latitude, double longitude,
double speed, double heading, double altitude,
int satellites, String fixQuality, Double batteryLevel) {
}
/**
* 从 Topic 中解析 deviceId

14
maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java
View File

@@ -45,8 +45,11 @@ public class MqttTopic {
// ==================== 任务管理 ====================
/** 路径规划下发（平台->设备），对应 TCP 0x01 path */
public static final String MOWER_TASK_ROUTE_SET = "mower/%s/task/route/set";
public static final String MOWER_TASK_ROUTE_POST = "mower/%s/task/route/post";
// public static final String MOWER_TASK_ROUTE_SET = "mower/%s/task/route/set";
// public static final String MOWER_TASK_ROUTE_POST = "mower/%s/task/route/post";
public static final String MOWER_TASK_ROUTE_SET = "mower/%s/task/target/set";
/** 启动任务执行（平台->设备），对应 TCP 0x12 interaction */
public static final String MOWER_TASK_STATUS_SET = "mower/%s/task/status/set";
public static final String MOWER_TASK_POINT_POST = "mower/%s/task/point/post";
@@ -87,6 +90,13 @@ public class MqttTopic {
/** 订阅所有设备的任务状态变更 */
public static final String MOWER_WILDCARD_TASK_STATUS = "mower/+/task/status/post";
public static final String MOWER_WILDCARD_TASK_ROUTE = " mower/+/task/route/post";
public static final String MOWER_HEARTBEAT = " mower/+/event/heartbeat/post";
/**
* 格式化 Topic 模板，将 %s 替换为设备ID
*

5
maibu-netty-server/src/main/java/com/maibu/controller/DeviceTaskController.java
View File

@@ -3,6 +3,7 @@ package com.maibu.controller;
import java.time.LocalDate;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.web.bind.annotation.GetMapping;
@@ -125,12 +126,12 @@ public class DeviceTaskController extends BaseController {
}
@PostMapping("/cancelTask")
public AjaxResult cancelTask(@RequestBody DeviceTaskCommandDTO dto) {
public AjaxResult cancelTask(@RequestBody DeviceTaskCommandDTO dto) throws MqttException {
String deviceId = dto.getDeviceId();
if (StringUtils.isEmpty(deviceId) && dto.getTaskId() == null) {
return AjaxResult.error("deviceId或taskId为空!");
}
return AjaxResult.success(deviceTaskService.cancelTask(dto, getUsername()));
return AjaxResult.success(deviceTaskService.cancelTask(dto));
}
@PostMapping("/pauseTask")

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

@@ -39,6 +39,9 @@ public class InitThread implements ApplicationRunner {
@Resource(name = FastBeeConstant.TASK.DEVICE_ERROR_MONITOR)
private Executor deviceErrorMonitorExecutor;
@Resource(name = FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR)
private Executor deviceHeartBeatMonitorExecutor;
@Resource(name = FastBeeConstant.TASK.COMMON_DEVICE_MONITOR)
private Executor commonDeviceMonitorExecutor;
@@ -124,6 +127,11 @@ public class InitThread implements ApplicationRunner {
GlobalMemory.initMqttDeviceMonitor(mqttHostUrl, mqttClientId, mqttUsername, mqttPassword);
// 上位机设备心跳监控
deviceHeartBeatMonitorExecutor.execute(() -> {
deviceThreadService.heartBeatMonitor();
});
// 设备错误监听线程
deviceErrorMonitorExecutor.execute(() -> {
try {

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

@@ -11,6 +11,7 @@ import java.util.stream.Collectors;
import com.maibu.memory.DeviceSessionManager;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@@ -476,7 +477,7 @@ public class DeviceTaskService {
return task.getId();
}
public boolean cancelTask(DeviceTaskCommandDTO dto, String username) {
public boolean cancelTask(DeviceTaskCommandDTO dto) throws MqttException {
String deviceId = dto.getDeviceId();
DevicePlanTask devicePlanTask = null;
if (StringUtils.isEmpty(dto.getDeviceId())) {
@@ -488,7 +489,11 @@ public class DeviceTaskService {
}
NettyDevice device = deviceSessionManager.getDevice(deviceId);
if (device != null && device.getTask() != null) {
return device.cancelTask();
if(dto.isForceCancel()){
return device.canceledTask();
}else {
return device.cancelingTask();
}
} else {
if (devicePlanTask != null) {
devicePlanTask.setTaskStaus(DeviceTaskStaus.CANCELED);

40
maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java
View File

@@ -14,12 +14,15 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import com.maibu.constant.Constant;
import com.maibu.core.host.HeartBeatDTO;
import com.maibu.memory.DeviceSessionManager;
import com.maibu.utils.json.JsonUtils;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.Constants;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.stereotype.Service;
@@ -254,4 +257,41 @@ public class DeviceThreadService {
return null;
}
/**
* 监控心跳
*/
public void heartBeatMonitor() {
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
executor.scheduleWithFixedDelay(() -> {
try {
Long nowTime = System.currentTimeMillis();
List<HeartBeatDTO> heartBeatDTOS = GlobalMemory.customMqttDeviceMonitor.getAllHeartBeat();
if (!CollectionUtils.isEmpty(heartBeatDTOS)) {
heartBeatDTOS.forEach(x -> {
Long lastActiveTime = x.getSysTimestamp();
NettyDevice device = sessionManager.getDevice(x.getSn());
if (nowTime - lastActiveTime >= Constant.hostOfflineInterval) {
//超时了判断离线
if (device != null) {
device.status(0);
//todo 推送设备离线
}
} else {
}
});
}
} catch (Exception e) {
log.error("执行监测心跳线程错误：{}", e.getMessage());
}
}, 0, 500, TimeUnit.MILLISECONDS);
}
}

184
maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/HostComputerController.java
View File

@@ -1,67 +1,181 @@
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.memory.GlobalMemory;
import com.maibu.service.impl.HostComputerService;
import com.maibu.service.impl.WorkRecordService;
import com.maibu.utils.MinioUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestPart;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.io.InputStream;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.zip.GZIPInputStream;
@Slf4j
@RestController
@RequestMapping("/iot/host")
public class HostComputerController extends BaseController {
@Autowired
private MinioUtil minioUtil;
@Autowired
private WorkRecordService workRecordService;
private static final int BATCH_SIZE = 5000;
@Autowired
private HostComputerService hostComputerService;
@PostMapping(value = "/add", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public R addWorkRecord(
@RequestPart("file") MultipartFile file, // 接收上传的图片
@RequestPart("workRecord") WorkRecord workRecord // 接收表单里的 JSON 对象
/**
* 上传 csv.gz 轨迹文件并解析为 LatAndLngEntity 列表（分批处理，每批 BATCH_SIZE 条）
*
* <p>CSV 格式：第 1 列=cell_index，第 2 列=x(lat)，第 3 列=y(lng)</p>
*
* @param file csv.gz 压缩文件
* @param taskId 任务ID
* @return 解析后的完整轨迹数据
*/
@PostMapping(value = "/upload-path-csv", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public R uploadCsv(
@RequestPart("file") MultipartFile file,
@RequestParam("taskId") Long taskId
) {
int i = -1;
WorkRecord workRecord1 = workRecordService.selectWorkRecordByWorkNameAndSiteId(workRecord.getWorkName(), workRecord.getSiteId());
if (workRecord1 != null) {
return R.fail("该任务已存在");
if (file == null || file.isEmpty()) {
return R.fail("文件不能为空");
}
if (taskId == null) {
return R.fail("taskId 不能为空");
}
try (InputStream inputStream = file.getInputStream()) {
String originalFilename = file.getOriginalFilename();
if (originalFilename == null || !originalFilename.toLowerCase().endsWith(".csv.gz")) {
return R.fail("只支持 .csv.gz 格式文件");
}
String fileName = "routePlanImage/rt" + workRecord.getWorkName() + System.currentTimeMillis() + ".png";
log.info("收到 csv.gz 文件: taskId={}, fileName={}, size={}", taskId, originalFilename, file.getSize());
// 上传到 MinIO
minioUtil.uploadFile(fileName, inputStream, "image/png");
List<LatAndLngEntity> allPath = new ArrayList<>();
List<LatAndLngEntity> batch = new ArrayList<>(BATCH_SIZE);
int batchIndex = 0;
// 获取访问 URL
String url = minioUtil.getPublicUrl(fileName);
System.out.println("✅ 图片上传成功，访问地址: " + url);
try (GZIPInputStream gzipInputStream = new GZIPInputStream(file.getInputStream());
BufferedReader reader = new BufferedReader(new InputStreamReader(gzipInputStream))) {
// 这里可以把 URL 存到 workRecord 里
workRecord.setImgUrl(url);
workRecord.setUserId(getUserId());
workRecord.setOrgId(getLoginUser().getOrgId());
String line;
int lineNum = 0;
while ((line = reader.readLine()) != null) {
lineNum++;
if (line.trim().isEmpty()) {
continue;
}
String[] values = line.split(",");
if (lineNum == 1) {
log.info("CSV 表头: {}", Arrays.toString(values));
continue;
}
if (values.length < 3) {
log.warn("第 {} 行列数不足3列，跳过: {}", lineNum, line);
continue;
}
LatAndLngEntity entity = parseLatLng(values);
if (entity != null) {
batch.add(entity);
}
if (batch.size() >= BATCH_SIZE) {
batchIndex++;
processBatch(taskId, batch, batchIndex);
allPath.addAll(batch);
batch.clear();
log.info("csv 解析进度: taskId={}, 第 {} 批({}条), 累计 {} 条",
taskId, batchIndex, BATCH_SIZE, allPath.size());
}
}
if (!batch.isEmpty()) {
batchIndex++;
processBatch(taskId, batch, batchIndex);
allPath.addAll(batch);
batch.clear();
}
log.info("csv 解析完成: taskId={}, 共 {} 个轨迹点, 分 {} 批处理", taskId, allPath.size(), batchIndex);
GlobalMemory.taskPathMap.put(taskId, allPath);
return R.ok();
i = hostComputerService.saveWorkRecord(workRecord);
} catch (Exception e) {
return R.fail("新增失败: " + e.getMessage());
log.error("csv.gz 解析失败: taskId={}, error={}", taskId, e.getMessage(), e);
return R.fail("文件解析失败: " + e.getMessage());
}
}
/**
* 从 CSV 行解析 LatAndLngEntity
* 第 2 列(x) → lat，第 3 列(y) → lng
*/
private LatAndLngEntity parseLatLng(String[] values) {
try {
String xStr = values[1].trim();
String yStr = values[2].trim();
if (xStr.isEmpty() || yStr.isEmpty()) {
return null;
}
LatAndLngEntity entity = new LatAndLngEntity();
entity.setLat(Double.parseDouble(xStr));
entity.setLng(Double.parseDouble(yStr));
return entity;
} catch (NumberFormatException e) {
log.warn("解析轨迹点失败: values={}, error={}", Arrays.toString(values), e.getMessage());
return null;
}
}
/**
* 分批处理轨迹数据
*
* <p>每批 BATCH_SIZE 条数据触发一次，可用于：</p>
* <ul>
* <li>批量入库（insertBatch）</li>
* <li>发布 MQTT 消息到上位机</li>
* <li>数据校验/统计</li>
* </ul>
*
* @param taskId 任务ID
* @param path 当前批次的轨迹点列表
* @param batchIndex 批次序号（从 1 开始）
*/
protected void processBatch(Long taskId, List<LatAndLngEntity> path, int batchIndex) {
if (path == null || path.isEmpty()) {
return;
}
return i <= 0 ? R.fail("新增失败") : R.ok("新增成功");
int size = path.size();
double minLat = Double.MAX_VALUE, maxLat = -Double.MAX_VALUE;
double minLng = Double.MAX_VALUE, maxLng = -Double.MAX_VALUE;
for (LatAndLngEntity p : path) {
if (p.getLat() < minLat) minLat = p.getLat();
if (p.getLat() > maxLat) maxLat = p.getLat();
if (p.getLng() < minLng) minLng = p.getLng();
if (p.getLng() > maxLng) maxLng = p.getLng();
}
log.info("【第 {} 批处理】taskId={}, 本批 {} 条, lat范围=[{}, {}], lng范围=[{}, {}]",
batchIndex, taskId, size,
String.format("%.6f", minLat), String.format("%.6f", maxLat),
String.format("%.6f", minLng), String.format("%.6f", maxLng));
}
}

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

@@ -3,6 +3,8 @@ package com.maibu.controller;
import java.io.InputStream;
import java.util.List;
import com.maibu.core.domain.AjaxResult;
import com.maibu.memory.GlobalMemory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
@@ -110,4 +112,9 @@ public class WorkRecordController extends BaseController {
return R.ok(path);
}
@GetMapping("getTaskPath")
public AjaxResult getTaskPath(@RequestParam Long taskId) {
return AjaxResult.success(GlobalMemory.taskPathMap.get(taskId));
}
}

14
maibu-service/maibu-iot-service/src/main/java/com/maibu/dto/host/Coverage.java
View File

@@ -1,14 +0,0 @@
package com.maibu.dto.host;
import lombok.Data;
@Data
public class Coverage {
private double cell_size;//覆盖分辨率，单位m
private double angle;//覆盖作业方向，单位度
private int start_point_index;//覆盖起始边界点序号
}

21
maibu-service/maibu-iot-service/src/main/java/com/maibu/dto/host/HostWorkRecordCreateDTO.java
View File

@@ -1,21 +0,0 @@
package com.maibu.dto.host;
import lombok.Data;
import java.util.List;
@Data
public class HostWorkRecordCreateDTO {
private String task_id; //任务唯一编号
private String command_id;//指令唯一编号
private String mode;//导航任务模式
private String action;//任务动作 start/pause/resume/cancel
private String frame_id; //WGS84
private List<PointEntity> points;
private Coverage coverage;
}

9
maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/HostComputerService.java
View File

@@ -1,6 +1,7 @@
package com.maibu.service.impl;
import com.maibu.core.business.WorkRecord;
import com.maibu.memory.GlobalMemory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@@ -13,13 +14,5 @@ public class HostComputerService {
private WorkRecordService workRecordService;
public int saveWorkRecord(WorkRecord workRecord) throws Exception {
// 发送 mqtt 消息到上位机
return workRecordService.addWorkRecord(workRecord);
}
}