lingyunxsh/mqtt_power
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

单片机互通调试初步完成

Browse Source
This commit is contained in:
lingyunxsh
2025-05-27 16:44:45 +08:00
parent 22e1109d81
commit 6aaae06bac
22 changed files with 1678 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
springboot-init-main/src/main/java/com/yupi/project/aop/MqttPublishLogAspect.java
View File

@@ -48,6 +48,7 @@ public class MqttPublishLogAspect {
String commandTypeArg = null;
String payloadJsonArg = null;
Long sessionIdArg = null;
Long associatedTaskIdArg = null;
Long taskIdForLog = null;
if (args.length >= 3) {
@@ -57,6 +58,9 @@ public class MqttPublishLogAspect {
if (args.length >= 4 && args[3] instanceof Long) {
sessionIdArg = (Long) args[3];
}
if (args.length >= 5 && args[4] instanceof Long) {
associatedTaskIdArg = (Long) args[4];
}
}
String topic = null;
@@ -82,7 +86,10 @@ public class MqttPublishLogAspect {
throw throwable;
}
if (robotIdArg != null && sessionIdArg !=null) {
if (associatedTaskIdArg != null) {
taskIdForLog = associatedTaskIdArg;
} else if (robotIdArg != null && sessionIdArg !=null) {
log.warn("associatedTaskIdArg was null in MqttPublishLogAspect for robot {}, session {}. Falling back to DB query for taskId.", robotIdArg, sessionIdArg);
RobotTask latestTask = robotTaskService.findLatestTaskByRobotIdAndSessionId(robotIdArg,sessionIdArg);
if(latestTask != null){
taskIdForLog = latestTask.getId();

4
springboot-init-main/src/main/java/com/yupi/project/service/MqttService.java
View File

@@ -14,10 +14,12 @@ public interface MqttService {
* @param commandType 指令类型
* @param payloadJson 指令的参数内容 (JSON格式字符串)。对于无参数指令，可以为 null 或空字符串。
* @param sessionId 关联的充电会话ID (可选, 主要用于充电相关指令)
* @param associatedTaskId (可选) 如果此指令关联到一个已由调用方创建的RobotTask，则传入其ID。
* 如果为null，MqttService内部可能会尝试创建新任务（取决于实现策略，但当前策略是外部创建）。
* @return 如果指令成功进入发送流程（任务已创建并尝试发布），返回 true；
* 如果机器人正忙或创建任务失败等原因导致无法发送，返回 false。
* @throws Exception 如果MQTT发布过程中发生异常 (例如 MqttException)
*/
boolean sendCommand(String robotId, CommandTypeEnum commandType, String payloadJson, Long sessionId) throws Exception;
boolean sendCommand(String robotId, CommandTypeEnum commandType, String payloadJson, Long sessionId, Long associatedTaskId) throws Exception;
}

9
springboot-init-main/src/main/java/com/yupi/project/service/RobotTaskService.java
View File

@@ -22,6 +22,15 @@ public interface RobotTaskService extends IService<RobotTask> {
*/
boolean hasPendingOrSentTask(String robotId);
/**
* 检查指定机器人是否有正在处理（PENDING 或 SENT）的任务, 排除指定的任务ID。
*
* @param robotId 机器人ID
* @param taskIdToExclude 要排除的任务ID (可以为 null)
* @return 如果有其他待处理或已发送任务，则返回 true，否则 false
*/
boolean hasOtherPendingOrSentTask(String robotId, Long taskIdToExclude);
/**
* 创建一个新的机器人指令任务。
*

40
springboot-init-main/src/main/java/com/yupi/project/service/impl/ChargingSessionServiceImpl.java
View File

@@ -153,9 +153,15 @@ public class ChargingSessionServiceImpl extends ServiceImpl<ChargingSessionMappe
// 5. 发送MQTT指令给机器人
try {
mqttService.sendCommand(assignedRobot.getRobotUid(), CommandTypeEnum.MOVE_TO_SPOT, moveToSpotPayload, moveTask.getId());
robotTaskService.markTaskAsSent(moveTask.getId(), new Date());
log.info("已向机器人 {} 发送移动指令 for session {}, 任务ID: {}", assignedRobot.getRobotUid(), sessionId, moveTask.getId());
boolean commandPublished = mqttService.sendCommand(assignedRobot.getRobotUid(), CommandTypeEnum.MOVE_TO_SPOT, moveToSpotPayload, sessionId, moveTask.getId());
if (!commandPublished) {
log.error("MQTT移动指令发送失败 (sendCommand返回false) for session {}, 任务ID: {}. 将任务标记为失败。", sessionId, moveTask.getId());
robotTaskService.markTaskAsFailed(moveTask.getId(), "MQTT 指令未成功发布 (sendCommand返回false)", new Date());
throw new BusinessException(ErrorCode.OPERATION_ERROR, "发送MQTT指令失败 (未发布)");
}
log.info("已成功请求向机器人 {} 发送移动指令 for session {}, 任务ID: {}", assignedRobot.getRobotUid(), sessionId, moveTask.getId());
} catch (Exception e) {
log.error("发送MQTT移动指令失败 for session {}, 任务ID: {}: ", sessionId, moveTask.getId(), e);
// 此处可能需要更复杂的错误处理，如重试、标记任务失败等
@@ -182,25 +188,29 @@ public class ChargingSessionServiceImpl extends ServiceImpl<ChargingSessionMappe
return false;
}
// 校验会话状态和关联的任务ID
// 原始状态应该是 ROBOT_ASSIGNED，表示移动任务已发出
if (!ChargingSessionStatusEnum.ROBOT_ASSIGNED.getValue().equals(session.getStatus()) ||
!robotTaskId.equals(session.getRelatedRobotTaskId())) {
log.warn("处理机器人到达事件失败，会话状态 ({}) 或任务ID ({}) 不匹配 (期望任务ID: {}) for session {}",
(session.getRelatedRobotTaskId() != null && !robotTaskId.equals(session.getRelatedRobotTaskId())) ) { // 确保关联的任务ID也匹配
log.warn("处理机器人到达事件失败 (或视为开始充电事件)，会话状态 ({}) 或任务ID ({}) 不匹配 (期望任务ID: {}) for session {}",
session.getStatus(), robotTaskId, session.getRelatedRobotTaskId(), sessionId);
return false;
}
session.setStatus(ChargingSessionStatusEnum.ROBOT_ARRIVED.getValue());
session.setRobotArrivalTime(new Date());
// 更新会话状态为 CHARGING_STARTED，并记录相关信息
session.setStatus(ChargingSessionStatusEnum.CHARGING_STARTED.getValue());
session.setRobotArrivalTime(new Date()); // 机器人到达时间 (即充电开始时间点)
session.setChargeStartTime(new Date()); // 明确记录充电开始时间
session.setUpdateTime(new Date());
// 接下来可以创建"开始充电"的机器人任务，或等待用户指令/自动开始
// 此处简化，认为到达后即可准备开始充电，相关任务由MQTT回调触发创建
boolean updated = this.updateById(session);
if(updated) {
log.info("机器人已到达，会话 {} 状态更新为 ROBOT_ARRIVED. 关联任务ID: {}", sessionId, robotTaskId);
// 可选：更新机器人状态为IDLE（在车位旁等待）或 CHARGING_READY
chargingRobotService.updateRobotStatus(session.getRobotUidSnapshot(), RobotStatusEnum.IDLE, null, null, null, new Date());
// 可选：更新车位状态为 CHARGING (如果机器人到达即代表车位被用于充电)
parkingSpotService.updateSpotStatus(session.getSpotUidSnapshot(), ParkingSpotStatusEnum.CHARGING, session.getId());
log.info("机器人已到达，会话 {} 状态更新为 CHARGING_STARTED (视为充电开始). 关联任务ID: {}", sessionId, robotTaskId);
// 更新机器人DB状态为 CHARGING
chargingRobotService.updateRobotStatus(session.getRobotUidSnapshot(), RobotStatusEnum.CHARGING, null, null, robotTaskId, new Date());
// 确保车位状态为 CHARGING (这一步之前已经有了，保持即可)
parkingSpotService.updateSpotStatus(session.getSpotUidSnapshot(), ParkingSpotStatusEnum.CHARGING, session.getId());
}
return updated;
}
@@ -543,7 +553,7 @@ public class ChargingSessionServiceImpl extends ServiceImpl<ChargingSessionMappe
boolean mqttSent = false;
try {
mqttSent = mqttService.sendCommand(session.getRobotUidSnapshot(), CommandTypeEnum.STOP_CHARGE, stopChargePayload, stopTask.getId());
mqttSent = mqttService.sendCommand(session.getRobotUidSnapshot(), CommandTypeEnum.STOP_CHARGE, stopChargePayload, sessionId, stopTask.getId());
} catch (Exception e) {
log.error("发送MQTT停止充电指令失败 for session {}, 任务ID: {}: {}", sessionId, stopTask.getId(), e.getMessage(), e);
robotTaskService.markTaskAsFailed(stopTask.getId(), "MQTT发送失败: " + e.getMessage(), new Date());

120
springboot-init-main/src/main/java/com/yupi/project/service/impl/MqttServiceImpl.java
View File

@@ -3,6 +3,7 @@ package com.yupi.project.service.impl;
import com.yupi.project.config.properties.MqttProperties;
import com.yupi.project.model.entity.RobotTask;
import com.yupi.project.model.enums.CommandTypeEnum;
import com.yupi.project.model.enums.RobotTaskStatusEnum;
import com.yupi.project.service.MqttService;
import com.yupi.project.service.RobotTaskService;
import lombok.RequiredArgsConstructor;
@@ -16,6 +17,9 @@ import org.springframework.transaction.annotation.Transactional;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.Map;
/**
* MQTT 服务实现类
@@ -28,14 +32,17 @@ public class MqttServiceImpl implements MqttService {
private final MqttClient mqttClient; // 由Spring从MqttConfig注入的MQTT客户端实例
private final MqttProperties mqttProperties; // MQTT相关的配置属性
private final RobotTaskService robotTaskService; // 机器人任务服务，用于创建和更新任务状态
private final ObjectMapper objectMapper; // Jackson ObjectMapper for JSON manipulation
// 构造函数注入依赖
public MqttServiceImpl(@Qualifier("mqttClientBean") MqttClient mqttClient,
MqttProperties mqttProperties,
RobotTaskService robotTaskService) {
RobotTaskService robotTaskService,
ObjectMapper objectMapper) {
this.mqttClient = mqttClient;
this.mqttProperties = mqttProperties;
this.robotTaskService = robotTaskService;
this.objectMapper = objectMapper;
}
/**
@@ -45,32 +52,93 @@ public class MqttServiceImpl implements MqttService {
* @param commandType 指令类型 (枚举)
* @param payloadJson 指令的JSON格式负载
* @param sessionId (可选) 与此指令关联的充电会话ID
* @param associatedTaskId (可选) 此指令关联的、已由调用方创建的RobotTask ID。
* @return 指令是否发送成功 (注意：这仅表示MQTT消息已发布，不代表设备已执行)
* @throws Exception 如果发送过程中发生MQTT异常或其他意外异常
*/
@Override
@Transactional(rollbackFor = Exception.class) // 如果MQTT发布失败或后续操作失败，确保事务回滚（例如任务创建）
public boolean sendCommand(String robotId, CommandTypeEnum commandType, String payloadJson, Long sessionId) throws Exception {
log.info("尝试向机器人 {} 发送指令 {}，负载: {}", commandType, robotId, payloadJson);
@Transactional(rollbackFor = Exception.class)
public boolean sendCommand(String robotId, CommandTypeEnum commandType, String payloadJson, Long sessionId, Long associatedTaskId) throws Exception {
log.info("尝试向机器人 {} 发送指令 {} (关联任务ID: {})，负载: {}", commandType, robotId, associatedTaskId == null ? "N/A" : associatedTaskId, payloadJson);
// 步骤1：首先检查机器人是否已有正在处理(PENDING)或已发送(SENT)的任务
if (robotTaskService.hasPendingOrSentTask(robotId)) {
// 特殊处理：STOP_CHARGE (停止充电) 指令具有高优先级，可以覆盖其他任务
if (associatedTaskId == null) {
log.error("MqttServiceImpl.sendCommand 错误: associatedTaskId 不能为 null。任务应由调用者预先创建。机器人 {}，指令 {}", robotId, commandType);
// 根据当前设计，associatedTaskId 是必需的，因为 MqttServiceImpl 不再负责创建任务。
throw new IllegalArgumentException("associatedTaskId cannot be null. Task must be pre-created by caller.");
}
// 步骤1：检查机器人是否已有 *其他* 正在处理(PENDING)或已发送(SENT)的任务
if (robotTaskService.hasOtherPendingOrSentTask(robotId, associatedTaskId)) {
if (CommandTypeEnum.STOP_CHARGE.equals(commandType)) {
log.info("机器人 {} 有正在处理的任务，但 STOP_CHARGE 是优先指令，继续执行。", robotId);
log.info("机器人 {} 有其他正在处理的任务，但 STOP_CHARGE 是优先指令，继续执行任务 {}。", robotId, associatedTaskId);
} else {
log.warn("机器人 {} 正忙 (存在PENDING或SENT状态的任务)，指令 {} 已中止。", robotId, commandType);
return false; // 机器人忙，不发送新指令 (除非是停止指令)
log.warn("机器人 {} 正忙 (存在其他PENDING或SENT状态的任务)，指令 {} (任务 {}) 已中止。", robotId, commandType, associatedTaskId);
// 注意：即使中止，预先创建的 associatedTaskId 任务仍然是 PENDING。调用者需要处理此情况。
return false;
}
}
// 步骤2：如果机器人不忙（或指令是STOP_CHARGE），再创建新任务
RobotTask task = robotTaskService.createTask(robotId, commandType, payloadJson, sessionId);
if (task == null || task.getId() == null) {
log.error("为机器人 {} 的指令 {} 创建RobotTask失败。发送操作中止。", commandType, robotId);
return false; // 任务创建失败，不继续发送
// 步骤2：获取预先创建的任务 (associatedTaskId)
RobotTask taskToProcess = robotTaskService.getById(associatedTaskId);
if (taskToProcess == null) {
log.error("为机器人 {} 的指令 {} 查找预创建的 RobotTask (ID: {}) 失败。发送操作中止。", commandType, robotId, associatedTaskId);
return false; // 任务未找到
}
log.info("已为机器人 {} 的指令 {} 创建 PENDING 状态的 RobotTask，任务ID: {}。", commandType, robotId, task.getId());
// 校验任务状态是否为 PENDING，因为后续会将其标记为 SENT
if (taskToProcess.getStatus() != RobotTaskStatusEnum.PENDING) {
log.warn("预创建的任务 {} (机器人 {}, 指令 {}) 状态为 {}，而非预期的 PENDING。可能已被处理或状态错误。发送操作中止。",
associatedTaskId, robotId, commandType, taskToProcess.getStatus());
return false;
}
log.info("处理预创建的 PENDING 状态 RobotTask ID: {}，机器人 {}，指令 {}。", associatedTaskId, robotId, commandType);
// <<<< START MODIFICATION: Add/Ensure taskId in payloadJson using associatedTaskId >>>>
String finalPayloadJson;
try {
// Parse the original payloadJson
// If payloadJson is null or empty, create a new JSON object
// Otherwise, parse it and add/overwrite the taskId
ObjectNode payloadNode;
if (payloadJson == null || payloadJson.trim().isEmpty()) {
payloadNode = objectMapper.createObjectNode();
} else {
// Check if it's a valid JSON, otherwise treat as non-JSON or handle error
try {
payloadNode = (ObjectNode) objectMapper.readTree(payloadJson);
} catch (Exception e) {
// If original payload is not a valid JSON, we might log a warning
// and create a new JSON object just for the taskId, or wrap the original string.
// For now, let's assume if it's not empty, it should be valid JSON.
// Or, simpler: create a Map, add taskId, then add all from original parsed map.
log.warn("Original payloadJson for robot {} command {} is not a valid JSON object: {}. Proceeding by creating a new JSON with taskId.", robotId, commandType, payloadJson, e);
// Fallback: create a new node. This might lose original non-JSON payload if that was intended.
// A safer approach for non-JSON or mixed payloads might be needed if that's a use case.
payloadNode = objectMapper.createObjectNode();
// If you expect non-JSON payloads and want to preserve them, this part needs more thought.
// For now, assuming command payloads are always JSON.
}
}
payloadNode.put("taskId", String.valueOf(associatedTaskId)); // NEW: use associatedTaskId
// For START_CHARGE, ensure sessionId from method argument is also in the payload if not already
// This is based on ESP32 code expecting sessionId in payload for START_CHARGE
if (CommandTypeEnum.START_CHARGE.equals(commandType) && sessionId != null) {
if (!payloadNode.has("sessionId") || payloadNode.get("sessionId") == null || payloadNode.get("sessionId").isNull()) {
payloadNode.put("sessionId", sessionId); // Add sessionId if not present
log.info("Added sessionId {} to START_CHARGE command payload for task {}.", sessionId, associatedTaskId);
}
}
finalPayloadJson = objectMapper.writeValueAsString(payloadNode);
} catch (Exception e) {
log.error("为机器人 {} 的指令 {} (任务ID: {}) 添加 taskId 到 payload 时发生JSON处理异常: {}", commandType, robotId, associatedTaskId, e.getMessage(), e);
// Mark task as failed because we cannot construct a valid payload
robotTaskService.markTaskAsFailed(associatedTaskId, "Payload JSON processing error: " + e.getMessage(), new Date());
return false; // Payload construction failed
}
// <<<< END MODIFICATION >>>>
// 3. 构建目标MQTT主题
// 主题格式: [mqttProperties.commandTopicBase]/[robotId]
@@ -78,38 +146,38 @@ public class MqttServiceImpl implements MqttService {
String topic = mqttProperties.getCommandTopicBase() + "/" + robotId;
// 4. 准备MQTT消息
String effectivePayload = (payloadJson == null) ? "" : payloadJson; // 确保payload不为null
MqttMessage mqttMessage = new MqttMessage(effectivePayload.getBytes(StandardCharsets.UTF_8));
MqttMessage mqttMessage = new MqttMessage(finalPayloadJson.getBytes(StandardCharsets.UTF_8));
mqttMessage.setQos(mqttProperties.getDefaultQos()); // 设置QoS级别，从配置中读取
// mqttMessage.setRetained(false); // 保留消息标志，默认为false，通常指令不需要保留
try {
// 5. 发布MQTT消息
if (!mqttClient.isConnected()) {
log.error("MQTT客户端未连接。无法向机器人 {} 发送指令 {}。任务ID: {}", commandType, robotId, task.getId());
log.error("MQTT客户端未连接。无法向机器人 {} 发送指令 {}。任务ID: {}", commandType, robotId, associatedTaskId);
// 任务将保持PENDING状态。后续可由定时任务重试或标记为错误。
return false;
}
log.debug("正在向主题发布消息: {}, QoS: {}, 负载: {}", topic, mqttMessage.getQos(), effectivePayload);
log.debug("正在向主题发布消息: {}, QoS: {}, 负载: {}", topic, mqttMessage.getQos(), finalPayloadJson);
mqttClient.publish(topic, mqttMessage);
log.info("成功向机器人 {} 在主题 {} 上发布指令 {}。任务ID: {}", commandType, robotId, topic, task.getId());
log.info("成功向机器人 {} 在主题 {} 上发布指令 {}。任务ID: {}", commandType, robotId, topic, associatedTaskId);
// 6. 将机器人任务的状态标记为 SENT (已发送)
boolean markedAsSent = robotTaskService.markTaskAsSent(task.getId(), new Date());
boolean markedAsSent = robotTaskService.markTaskAsSent(associatedTaskId, new Date());
if (!markedAsSent) {
log.warn("指令 {} 已发布给机器人 {}，但将任务 {} 标记为 SENT 失败。系统可能处于不一致状态。",
commandType, robotId, task.getId());
commandType, robotId, associatedTaskId);
// 这是一个严重警告。指令已发送，但数据库状态未正确反映。
// 事务应回滚。
}
return true; // 指令已发送，并尝试更新了任务状态
} catch (MqttException e) {
log.error("向机器人 {} 发布指令 {} (任务ID: {}) 时发生MqttException。错误: {}",
commandType, robotId, task.getId(), e.getMessage(), e);
// RobotTask 将保持 PENDING 状态。重试机制或超时任务后续可以处理此问题。
commandType, robotId, associatedTaskId, e.getMessage(), e);
// RobotTask (associatedTaskId) 将保持 PENDING 状态 (或它之前的状态)。事务应回滚。
throw e; // 重新抛出异常，以便Spring的@Transactional进行事务回滚
} catch (Exception e) {
log.error("向机器人 {} 发送指令 {} (任务ID: {}) 时发生意外异常。错误: {}",
commandType, robotId, task.getId(), e.getMessage(), e);
commandType, robotId, associatedTaskId, e.getMessage(), e);
throw e; // 重新抛出异常，以便事务回滚
}
}

20
springboot-init-main/src/main/java/com/yupi/project/service/impl/RobotTaskServiceImpl.java
View File

@@ -44,6 +44,26 @@ public class RobotTaskServiceImpl extends ServiceImpl<RobotTaskMapper, RobotTask
return false;
}
@Override
public boolean hasOtherPendingOrSentTask(String robotId, Long taskIdToExclude) {
if (robotId == null) {
log.warn("Cannot check for other pending/sent tasks: robotId is null.");
return false;
}
QueryWrapper<RobotTask> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("robot_id", robotId)
.in("status", RobotTaskStatusEnum.PENDING, RobotTaskStatusEnum.SENT);
if (taskIdToExclude != null) {
queryWrapper.ne("id", taskIdToExclude); // Exclude the current task ID
}
long count = this.count(queryWrapper);
if (count > 0) {
log.info("Robot {} has {} other PENDING or SENT tasks (excluding task ID: {}).", robotId, count, taskIdToExclude == null ? "N/A" : taskIdToExclude);
return true;
}
return false;
}
@Override
@Transactional
public RobotTask createTask(String robotId, CommandTypeEnum commandType, String payloadJson, Long sessionId) {