package com.xiaopeng.automaticPilot.service;
|
|
import cn.hutool.core.lang.Snowflake;
|
import cn.hutool.core.util.IdUtil;
|
import com.alibaba.fastjson.JSONObject;
|
import com.alibaba.fastjson2.JSON;
|
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
|
import com.xiaopeng.api.dto.operation.LonLatDto;
|
import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO;
|
import com.xiaopeng.api.dto.pushMessage.SendMessageDTO;
|
import com.xiaopeng.api.feignClient.map.XpMapClient;
|
import com.xiaopeng.api.feignClient.order.XpOrderClient;
|
import com.xiaopeng.api.feignClient.pushMessage.XpPushMessageClient;
|
import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
|
import com.xiaopeng.api.model.order.XpOrder;
|
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
|
import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
|
import com.xiaopeng.api.vo.automaticPilot.proto.down.GeoPoint;
|
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.*;
|
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.*;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
|
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
|
import com.xiaopeng.api.vo.operation.RouteVo;
|
import com.xiaopeng.automaticPilot.util.XPUUtil;
|
import com.xiaopeng.common.core.domain.R;
|
import com.xiaopeng.common.core.utils.StringUtils;
|
import com.xiaopeng.common.core.utils.uuid.UUID;
|
import com.xiaopeng.common.redis.service.RedisService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.data.mongodb.core.MongoTemplate;
|
import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
|
import org.springframework.data.mongodb.core.query.Criteria;
|
import org.springframework.data.mongodb.core.query.Query;
|
import org.springframework.data.redis.core.RedisTemplate;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.Objects;
|
|
/**
|
* Kafka 消费者服务
|
*/
|
@Slf4j
|
@Service
|
public class KafkaConsumerService {
|
|
@Resource
|
private XpMapClient xpMapClient;
|
|
@Resource
|
private XpOrderClient xpOrderClient;
|
|
@Resource
|
private MongoTemplate mongoTemplate;
|
|
@Resource
|
private XpVehicleClient xpVehicleClient;
|
|
@Resource
|
private XPUUtil xpuUtil;
|
|
@Autowired
|
public RedisTemplate redisTemplate;
|
|
@Resource
|
private XpPushMessageClient xpPushMessageClient;
|
|
|
/**
|
* 消费消息 - 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);
|
systemStatusXpu(msgContent);
|
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发布系统运行状态---车辆实时位置
|
* @param msgContent
|
*/
|
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();
|
redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);
|
|
//保存车辆实时信息到mongodb
|
VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class);
|
if(null == vehiclePosition){
|
vehiclePosition = new VehiclePosition();
|
}
|
vehiclePosition.setVin(vin);
|
vehiclePosition.setPoint(new GeoJsonPoint(longitude, latitude));
|
vehiclePosition.setCourse(course);
|
vehiclePosition.setAlt(alt);
|
VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicle_status();
|
//车辆动态信息
|
VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamic_status();
|
dynamicStatus.setSpeed(dynamicStatus.getSpeed());
|
vehiclePosition.setCurrentElectricity(dynamicStatus.getBattery_level());
|
//订单信息
|
OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunning_order_status_info();
|
Long orderId = null;
|
String order_id = runningOrderStatusInfo.getOrder_id();
|
vehiclePosition.setHasOrder(0);
|
if(StringUtils.isNotEmpty(order_id)){
|
if(order_id.contains("XU")){
|
vehiclePosition.setHasOrder(1);
|
orderId = Long.parseLong(order_id.substring(2));
|
}else{
|
orderId = Long.parseLong(order_id);
|
}
|
}
|
vehiclePosition.setOrderId(orderId);
|
mongoTemplate.save(vehiclePosition);
|
//调用推送服务接口
|
if(null != orderId){
|
RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info();
|
PushOrderRouteDTO dto = new PushOrderRouteDTO();
|
dto.setOrderId(orderId);
|
dto.setRouteType(order_id.contains("XU") ? 2 : 1);
|
dto.setLongitude(longitude);
|
dto.setLatitude(latitude);
|
dto.setDuration(rtTripInfo.getRemaining_time());
|
dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
|
xpPushMessageClient.pushOrderRoute(dto);
|
}
|
}
|
|
|
|
|
|
/**
|
* 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("驶往上车点的行程");
|
buildNaviRequest(orderId, taskId, orderPhase);
|
break;
|
case ROUTE_TO_DESTINATION:
|
log.info("驶往下车点的行程");
|
break;
|
case ROUTE_TO_CRUISE:
|
log.info("巡游阶段路线");
|
break;
|
}
|
}
|
|
|
/**
|
* 构建导航请求
|
* @param orderId
|
* @param taskId
|
* @param orderPhase
|
*/
|
public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase){
|
//订单类型 1：乘客订单，2=巡游订单
|
Integer orderType = 1;
|
LonLatDto destination = new LonLatDto();
|
LonLatDto origin = new LonLatDto();
|
if(orderId.contains("XU")){
|
orderType = 2;
|
// todo 下发巡游单的导航数据
|
//巡游单
|
if(orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)){
|
origin.setLongitude(116.40528);
|
origin.setLatitude(39.90498);
|
destination.setLongitude(116.40528);
|
destination.setLatitude(39.90498);
|
}
|
}else{
|
//正常单
|
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
|
if(200 != xpOrderR.getCode()){
|
throw new RuntimeException("订单不存在");
|
}
|
XpOrder xpOrder = xpOrderR.getData();
|
//驶往上车点的行程
|
if(orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)){
|
XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
|
VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
|
if(Objects.isNull(vehiclePosition)){
|
throw new RuntimeException("车辆位置信息不存在");
|
}
|
//查询司机当前位置到起点的线路
|
origin.setLongitude(vehiclePosition.getPoint().getX());
|
origin.setLatitude(vehiclePosition.getPoint().getY());
|
destination.setLongitude(xpOrder.getStartLon());
|
destination.setLatitude(xpOrder.getStartLat());
|
}
|
//驶往下车点的行程
|
if(orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)){
|
//查询起点到终点的线路
|
origin.setLongitude(xpOrder.getStartLon());
|
origin.setLatitude(xpOrder.getStartLat());
|
destination.setLongitude(xpOrder.getEndLon());
|
destination.setLatitude(xpOrder.getEndLat());
|
}
|
OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
|
dto.setOrderId(Long.parseLong(orderId));
|
dto.setOrigin(origin);
|
dto.setDestination(destination);
|
R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
|
if(200 != routeVoR.getCode()){
|
throw new RuntimeException("获取订单线路和运费失败");
|
}
|
RouteVo routeVo = routeVoR.getData();
|
//构建导航下发请求对象
|
cloudOrderOperation(orderType, orderId, taskId, routeVo);
|
}
|
}
|
|
|
/**
|
* 云端订单操作----下发导航数据
|
* @param routeVo
|
*/
|
public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo){
|
//构建消息体
|
OrderOperation orderOperation = new OrderOperation();
|
orderOperation.setOrder_id(orderId);
|
orderOperation.setTask_id(taskId);
|
orderOperation.setOperation_type(OrderOperationType.CLOUD_UPDATE_WAYPOINTS);
|
|
//构建航点
|
List<Waypoint> waypoints = new ArrayList<>();
|
List<LonLatDto> polyline = routeVo.getPolyline();
|
for (int i = 0; i < polyline.size(); i++) {
|
LonLatDto lonLatDto = polyline.get(i);
|
Waypoint waypoint = new Waypoint();
|
waypoint.setWaypoint_id(UUID.randomUUID().toString());
|
if(0 == i){
|
waypoint.setPoint_type(PointType.START_POINT);
|
//订单类型不同，点位的动作类型不同
|
waypoint.setAction_type(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED);
|
}else if(polyline.size() - 1 == i){
|
waypoint.setPoint_type(PointType.END_POINT);
|
waypoint.setAction_type(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED);
|
}else{
|
waypoint.setPoint_type(PointType.WAY_POINT);
|
//订单类型不同，点位的动作类型不同
|
waypoint.setAction_type(ActionType.ACTION_UNSPECIFIED);
|
}
|
//构建经纬度
|
GeoPoint geoPoint = new GeoPoint();
|
geoPoint.setLongitude(lonLatDto.getLongitude());
|
geoPoint.setLatitude(lonLatDto.getLatitude());
|
waypoint.setPoint(geoPoint);
|
waypoint.setSequence(i + 1);
|
waypoints.add(waypoint);
|
}
|
orderOperation.setWaypoints(waypoints);
|
//开发发送请求到XPU
|
xpuUtil.cloudOrderOperation(orderOperation);
|
}
|
|
|
public void c(MsgContent msgContent){
|
|
|
|
}
|
|
}
|
