package com.xiaopeng.automaticPilot.util; import cn.hutool.core.lang.Snowflake; import cn.hutool.core.util.IdUtil; import com.alibaba.fastjson2.JSON; import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto; import com.xiaopeng.api.feignClient.automaticPilot.VmpOpenServiceClient; import com.xiaopeng.api.feignClient.automaticPilot.XPUClient; import com.xiaopeng.api.vo.automaticPilot.MqttPublishVo; import com.xiaopeng.api.vo.automaticPilot.protobuf.*; import com.xiaopeng.common.core.utils.StringUtils; import com.xiaopeng.common.core.utils.uuid.UUID; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.util.Collections; import java.util.List; import java.util.Map; /** * 自动驾驶接口对接文档 */ @Slf4j @Component public class XPUUtil { @Value("${xpu.topic.msg.prefix:Robotaxi/Msg/Xpu}") private String topicPrefix; @Resource private XPUClient xpuClient; @Resource private VmpOpenServiceClient vmpOpenServiceClient; // 查询车管绑定类型,1:XPUID,2:待扩展 public static final String XPU_ID_TYPE = "1"; // 查询车管 车辆xpuid信息返回字段,用于按需返回字段 public static final List TYPE_LIST = Collections.singletonList("1"); private static final com.google.protobuf.util.JsonFormat.Printer jsonPrinter = com.google.protobuf.util.JsonFormat.printer().printingEnumsAsInts(); /** * Robotaxi创建订单推送给XPU * * @param taskList * @return */ public MqttPublishDto cloudCreateOrder(RtTaskList taskList, String vin, Long sequence) { MsgContent.Builder msgContentBuilder = MsgContent.newBuilder(); MsgHeader.Builder headerBuilder = MsgHeader.newBuilder(); headerBuilder.setMsgId(UUID.randomUUID().toString()); headerBuilder.setTimestamp(System.currentTimeMillis()); headerBuilder.setCmd(RTCmdType.CLOUD_CREATE_ORDER.getValue()); headerBuilder.setSequence(sequence); headerBuilder.setType(MessageType.MESSAGE_TYPE_REQUEST); headerBuilder.setSource(SourceType.CLOUD); msgContentBuilder.setHeader(headerBuilder.build()); msgContentBuilder.setRtTaskList(taskList); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopicByVin(vin)); try { vo.setPayload(jsonPrinter.print(msgContentBuilder.build())); } catch (Exception e) { log.error("Protobuf转JSON失败", e); throw new RuntimeException("Protobuf转JSON失败", e); } log.info("云端-->XPU 创建订单:{}", JSON.toJSONString(vo)); try { MqttPublishDto dto = xpuClient.publish(vo); log.info("XPU返回:{}", JSON.toJSONString(dto)); if(200 != dto.getCode()){ dto = againPushMessage(vo); } return dto; }catch (Exception e){ throw new RuntimeException("云端-->XPU 创建订单失败:", e); } } /** * 取消订单、修改途经点、终点 * @param operation * @return */ public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin, Long sequence) { MsgContent.Builder msgContentBuilder = MsgContent.newBuilder(); MsgHeader.Builder headerBuilder = MsgHeader.newBuilder(); headerBuilder.setMsgId(UUID.randomUUID().toString()); headerBuilder.setTimestamp(System.currentTimeMillis()); headerBuilder.setCmd(RTCmdType.CLOUD_ORDER_OPERATION.getValue()); headerBuilder.setSequence(sequence); headerBuilder.setType(MessageType.MESSAGE_TYPE_REQUEST); headerBuilder.setSource(SourceType.CLOUD); msgContentBuilder.setHeader(headerBuilder.build()); msgContentBuilder.setOrderOperation(operation); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopicByVin(vin)); try { vo.setPayload(jsonPrinter.print(msgContentBuilder.build())); } catch (Exception e) { log.error("Protobuf转JSON失败", e); throw new RuntimeException("Protobuf转JSON失败", e); } log.info("云端-->XPU 修改订单:{}", vo); try { MqttPublishDto dto = xpuClient.publish(vo); log.info("XPU返回:{}", JSON.toJSONString(dto)); if(200 != dto.getCode()){ dto = againPushMessage(vo); } return dto; }catch (Exception e){ throw new RuntimeException("云端-->XPU 修改订单失败:", e); } } /** * 下发导航地址到车端 * * @param navigationData * @return */ public MqttPublishDto cloudPushNavigation(RestoreNavigationData navigationData, String xpuId, Long sequence) { MsgContent.Builder msgContentBuilder = MsgContent.newBuilder(); MsgHeader.Builder headerBuilder = MsgHeader.newBuilder(); headerBuilder.setMsgId(UUID.randomUUID().toString()); headerBuilder.setTimestamp(System.currentTimeMillis()); headerBuilder.setCmd(RTCmdType.CLOUD_NAVIGATION_RESPONSE.getValue()); headerBuilder.setSequence(sequence); headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE); headerBuilder.setSource(SourceType.CLOUD); msgContentBuilder.setHeader(headerBuilder.build()); msgContentBuilder.setRestoreNavigationData(navigationData); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopicByXpuId(xpuId)); try { vo.setPayload(jsonPrinter.print(msgContentBuilder.build())); } catch (Exception e) { log.error("Protobuf转JSON失败", e); throw new RuntimeException("Protobuf转JSON失败", e); } log.info("云端-->XPU 下发导航信息:{}", vo); try { MqttPublishDto dto = xpuClient.publish(vo); log.info("XPU返回:{}", JSON.toJSONString(dto)); if(200 != dto.getCode()){ dto = againPushMessage(vo); } return dto; }catch (Exception e){ throw new RuntimeException("云端-->XPU 下发导航信息失败:", e); } } /** * 响应消息 * * @param responseMessage * @return */ public MqttPublishDto responseMessage(ResponseMessage responseMessage, String vin) { MsgContent.Builder msgContentBuilder = MsgContent.newBuilder(); MsgHeader.Builder headerBuilder = MsgHeader.newBuilder(); headerBuilder.setMsgId(UUID.randomUUID().toString()); headerBuilder.setTimestamp(System.currentTimeMillis()); headerBuilder.setCmd(RTCmdType.CLOUD_TASK_CONTROL.getValue()); Snowflake snowflake = IdUtil.getSnowflake(0, 0); headerBuilder.setSequence(snowflake.nextId()); headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE); headerBuilder.setSource(SourceType.CLOUD); msgContentBuilder.setHeader(headerBuilder.build()); msgContentBuilder.setResponse(responseMessage); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopicByVin(vin)); try { vo.setPayload(jsonPrinter.print(msgContentBuilder.build())); } catch (Exception e) { log.error("Protobuf转JSON失败", e); throw new RuntimeException("Protobuf转JSON失败", e); } log.info("云端-->XPU 响应消息:{}", vo); return xpuClient.publish(vo); } /** * @description: 根据vin查询xpuId * @param: vin * @return: xpuId */ public String getXpuIdByVin(String vin) { Map xpuIdMapping = vmpOpenServiceClient.getVehicleBindInfoByVin(vin, TYPE_LIST); log.info("getMqttPublishTopic vin,{},xpuIdMapping,{}", vin, JSON.toJSONString(xpuIdMapping)); if (xpuIdMapping == null || StringUtils.isEmpty(xpuIdMapping.get(XPU_ID_TYPE)) ) { log.error("根据vin查询车辆xpuid信息为空,vin={}", vin); throw new RuntimeException("根据vin查询车辆xpuid信息为空"); } return xpuIdMapping.get(XPU_ID_TYPE); } private String buildTopicByVin(String vin) { String xpuId = getXpuIdByVin(vin); return this.buildTopicByXpuId(xpuId); } private String buildTopicByXpuId(String xpuId) { String prefix = topicPrefix == null ? "" : topicPrefix; if (!prefix.endsWith("/")) { prefix = prefix + "/"; } String value = xpuId; if (xpuId.length() > 10) { int start = 10; int end = Math.min(24, xpuId.length()); value = xpuId.substring(start, end); } return prefix + value + "00000000"; } /** * 消息下发XPU重试机制 * @param message * @return */ public MqttPublishDto againPushMessage(MqttPublishVo message) { log.info("云端-->XPU 消息重试:{}", message); MqttPublishDto dto = null; for (int i = 0; i < 3; i++) { try { dto = xpuClient.publish(message); log.info("云端-->XPU 消息重试{}次:{}", (i+1), message); if (dto.getCode() == 200) { break; } }catch (Exception e){ log.error("云端-->XPU 推送失败:{}", message); } } return dto; } }