package com.xiaopeng.automaticPilot.service;
|
|
import com.alibaba.fastjson2.JSON;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.MessageType;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.MsgContent;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.RTCmdType;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.RtMsgHeader;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.RealtimePose;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.SystemRuntimeStatus;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.VehicleStaticInfo;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.Wgs;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.kafka.annotation.KafkaListener;
|
import org.springframework.stereotype.Service;
|
|
/**
|
* Kafka 消费者服务
|
*/
|
@Slf4j
|
@Service
|
public class KafkaConsumerService {
|
|
|
|
|
/**
|
* 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
|
*
|
*/
|
public void topicProdRobotaxiXpuHigh(String message) {
|
MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
|
RtMsgHeader header = msgContent.getHeader();
|
MessageType type = header.getType();
|
//处理请求消息
|
if (MessageType.MESSAGE_TYPE_REQUEST == type) {
|
//区分业务类型
|
RTCmdType cmd = header.getCmd();
|
switch (cmd) {
|
case SYSTEM_STATUS_XPU:
|
log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
|
break;
|
case XPU_NAVIGATION_REQUEST:
|
log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
|
xpuNavigationRequest(msgContent);
|
break;
|
case TRIP_STATUS_UPDATE_EVENT:
|
log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case PULLOVER_STATUS_EVENT:
|
log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case DRIVING_MODE_UPDATE_EVENT:
|
log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case SYSTEM_STATUS_UPDATE_EVENT:
|
log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case TASK_STATE_UPDATE_EVENT:
|
log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case ORDER_STATE_UPDATE_EVENT:
|
log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case POINT_ARRIVAL_STATUS_EVENT:
|
log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case PICKUP_CONFIRMATION_EVENT:
|
log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case DROPOFF_INFO_EVENT:
|
log.info("用户下车信息(XPU→云端)：{}", message);
|
break;
|
case NAVIGATION_DEVIATION_EVENT:
|
log.info("导航偏航(XPU→云端)：{}", message);
|
break;
|
case EMERGENCY_STOP_EVENT:
|
log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case NAVIGATION_UPDATE_EVENT:
|
log.info("导航信息变化(XPU→云端)：{}", message);
|
break;
|
case TASK_QUEUE_UPDATE_EVENT:
|
log.info("XPU待执行任务队列(XPU→云端)：{}", message);
|
break;
|
case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
|
log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
|
break;
|
case ORDER_STATE_UPDATE_TO_CDU_EVENT:
|
log.info("更新订单状态到CDU (XPU→云端)：{}", message);
|
break;
|
default:
|
log.info("消息不在处理范围内，忽略：{}", message);
|
}
|
|
|
|
}
|
|
}
|
|
/**
|
* 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
|
*/
|
public void topicProdRobotaxiXpuNormal(String message) {
|
MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
|
RtMsgHeader header = msgContent.getHeader();
|
MessageType type = header.getType();
|
//处理请求消息
|
if (MessageType.MESSAGE_TYPE_REQUEST == type) {
|
//区分业务类型
|
RTCmdType cmd = header.getCmd();
|
switch (cmd) {
|
case SYSTEM_STATUS_XPU:
|
log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
|
break;
|
case XPU_NAVIGATION_REQUEST:
|
log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
|
xpuNavigationRequest(msgContent);
|
break;
|
case TRIP_STATUS_UPDATE_EVENT:
|
log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case PULLOVER_STATUS_EVENT:
|
log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case DRIVING_MODE_UPDATE_EVENT:
|
log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case SYSTEM_STATUS_UPDATE_EVENT:
|
log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case TASK_STATE_UPDATE_EVENT:
|
log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case ORDER_STATE_UPDATE_EVENT:
|
log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case POINT_ARRIVAL_STATUS_EVENT:
|
log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case PICKUP_CONFIRMATION_EVENT:
|
log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case DROPOFF_INFO_EVENT:
|
log.info("用户下车信息(XPU→云端)：{}", message);
|
break;
|
case NAVIGATION_DEVIATION_EVENT:
|
log.info("导航偏航(XPU→云端)：{}", message);
|
break;
|
case EMERGENCY_STOP_EVENT:
|
log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
|
break;
|
case NAVIGATION_UPDATE_EVENT:
|
log.info("导航信息变化(XPU→云端)：{}", message);
|
break;
|
case TASK_QUEUE_UPDATE_EVENT:
|
log.info("XPU待执行任务队列(XPU→云端)：{}", message);
|
break;
|
case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
|
log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
|
break;
|
case ORDER_STATE_UPDATE_TO_CDU_EVENT:
|
log.info("更新订单状态到CDU (XPU→云端)：{}", message);
|
break;
|
default:
|
log.info("消息不在处理范围内，忽略：{}", message);
|
}
|
|
|
}
|
|
}
|
|
|
public void systemStatusXpu(MsgContent msgContent){
|
SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
|
//获取实时位姿
|
RealtimePose pose = systemRuntimeStatus.getPose();
|
Wgs naviPose = pose.getNavi_pose();
|
Double latitude = naviPose.getLatitude();
|
Double longitude = naviPose.getLongitude();
|
Double course = naviPose.getCourse();
|
Float alt = naviPose.getAlt();
|
//获取车辆静态信息
|
VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
|
String vin = staticInfo.getVin();
|
|
|
|
}
|
|
|
|
|
|
/**
|
* XPU向云端请求导航
|
* @param msgContent
|
*/
|
public void xpuNavigationRequest(MsgContent msgContent){
|
NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
|
String orderId = naviRequest.getOrder_id();
|
String taskId = naviRequest.getTask_id();
|
OrderPhase orderPhase = naviRequest.getOrder_phase();
|
log.info("订单ID：{}", orderId);
|
log.info("任务ID：{}", taskId);
|
switch (orderPhase){
|
case IDLE:
|
log.info("订单的任务未开始执行");
|
break;
|
case ROUTE_TO_PICKUP:
|
log.info("驶往上车点的行程");
|
|
break;
|
case ROUTE_TO_DESTINATION:
|
log.info("驶往下车点的行程");
|
break;
|
case ROUTE_TO_CRUISE:
|
log.info("巡游阶段路线");
|
break;
|
}
|
}
|
|
|
|
public void c(MsgContent msgContent){
|
|
|
|
}
|
|
}
|
