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){ } }