java/Robotaxi.git

			@@ -1,5 +1,16 @@
			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;
			@@ -11,36 +22,213 @@
			@Service
			public class KafkaConsumerService {




			/**
			* 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
			*
			* @param message 消息内容
			*/
			@KafkaListener(topics = "topic-prod-robotaxi-xpu-high", groupId = "group-robotaxi-xpu")
			public void consumeAutomaticPilotMessage(String message) {
			try {
			log.info("收到XPU上报高优先级主题消息: {}", message);
			} catch (Exception e) {
			log.error("处理XPU上报高优先级主题消息失败: {}", message, e);
			throw e;
			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（低优先级）
			*
			* @param message 消息内容
			*/
			@KafkaListener(topics = "topic-prod-robotaxi-xpu-normal", groupId = "group-robotaxi-xpu")
			public void consumeVehicleStatusMessage(String message) {
			try {
			log.info("收到XPU上报低优先级主题消息: {}", message);
			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);
			}

			} catch (Exception e) {
			log.error("处理XPU上报低优先级主题消息失败: {}", message, e);
			throw e;

			}

			}


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



			}

			}