4.8 态势感知

• sample代码架构
• sample代码实现
• 开发注意事项

4.8.1 sample代码架构

态势感知功能只在Pilot2上支持，机场没有可视化界面，不支持态势感知功能。

1. Pilot2与云端建立连接，并成功加载TSA态势感知模块：在Pilot2上通过JSBridge加载TSA态势感知模块后，会建立与后端的websocket连接，建立websocket连接后，Pilot2会请求设备列表的拓扑。
2. 设备上下线：云端感知到任意设备上下线后都会通过websocket协议，通知同一工作空间下的所有设备更新设备拓扑。
3. 推送位置信息：云端在收到设备上报的osd信息后，会通过态势感知功能的websocket连接，将设备的位置信息等推送到同一工作空间下的所有设备。

4.8.2 sample代码实现

1、Pilot2成功上云，并通过JSBridge成功加载TSA态势感知模块后，Pilot2会与后端建立websocket连接，成功建立websocket连接后，Pilot2会通过GET /manage/api/v1/workspaces/{workspace_id}/devices/topologies请求后端获取设备列表拓扑，对应sdk包中封装的接口为：IHttpTsaService#obtainDeviceTopologyList

@GetMapping(PREFIX + "/workspaces/{workspace_id}/devices/topologies")
HttpResultResponse<TopologyResponse> obtainDeviceTopologyList(
    @PathVariable(name = "workspace_id") String workspaceId,
    HttpServletRequest req, HttpServletResponse rsp);

在示例代码中的实现为：TopologyController#obtainDeviceTopologyList

@Override
public HttpResultResponse<TopologyResponse> obtainDeviceTopologyList(String workspaceId, HttpServletRequest req, HttpServletResponse rsp) {
    // 调用topologyService#getDeviceTopology获取该工作空间下的设备拓扑
    List<TopologyList> topologyList = topologyService.getDeviceTopology(workspaceId);
    return HttpResultResponse.success(new TopologyResponse().setList(topologyList));
}

topologyService#getDeviceTopology：

@Override
public List<TopologyList> getDeviceTopology(String workspaceId) {
    // 从Redis中获取当前工作空间下的所有网关设备
    List<DeviceDTO> gatewayList = deviceService.getDevicesByParams(
        DeviceQueryParam.builder()
        .workspaceId(workspaceId)
        .domains(List.of(DeviceDomainEnum.REMOTER_CONTROL.getDomain()))
        .build());

    List<TopologyList> topologyList = new ArrayList<>();
    // 查询网关设备所关联的子设备
    gatewayList.forEach(device -> this.getDeviceTopologyByGatewaySn(device.getDeviceSn())
        .ifPresent(topologyList::add));
    return topologyList;
}

topologyService#getDeviceTopologyByGatewaySn

public Optional<TopologyList> getDeviceTopologyByGatewaySn(String gatewaySn) {
    // 查询设备是否在线，如果不在线，则直接返回
    Optional<DeviceDTO> dtoOptional = deviceService.getDeviceBySn(gatewaySn);
    if (dtoOptional.isEmpty()) {
        return Optional.empty();
    }
    List<DeviceTopology> parents = new ArrayList<>();
    DeviceDTO device = dtoOptional.get();
    // 将查询的网关设备信息转换为另一个实体
    DeviceTopology gateway = deviceService.deviceConvertToTopologyDTO(device);
    parents.add(gateway);

    // 查询设备拓扑关系
    Optional<TopologyDeviceDTO> deviceTopo = deviceService.getDeviceTopoForPilot(device.getChildDeviceSn());
    List<DeviceTopology> deviceTopoList = new ArrayList<>();
    deviceTopo.ifPresent(deviceTopoList::add);
    return Optional.ofNullable(new TopologyList().setParents(parents).setHosts(deviceTopoList));
}

2、在设备上下线时，云端收到设备上下线的消息，会通过websocket协议推送给同一工作空间下的所有设备（可以参考4.1 设备上线），Pilot2上处于同一工作空间下的设备会收到设备上下线消息。

3、Pilot2内部会再次通过GET /manage/api/v1/workspaces/{workspace_id}/devices/topologies请求后端获取设备列表拓扑。更新Pilot2上的拓扑信息。

4、设备收到上报的osd信息会通过websocket协议传给同一工作空间下的所有设备，Pilot2收到消息后，会在地图上实时展示设备的位置信息，设备朝向等。

上线设备上报的osd消息，所有mqtt消息都会经过InboundMessageRouter#determineTargetChannels方法，在该方法中找到通过DeviceTopicEnum#find方法找到消息对应的主题，如果是 thing/product/{device_sn}/osd 主题的消息会被选择到ChannelName.INBOUND_OSD通道中。

public class InboundMessageRouter extends AbstractMessageRouter {

    @Override
    @Router(inputChannel = ChannelName.INBOUND)
    protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
        // 获取消息头
        MessageHeaders headers = message.getHeaders();
        String topic = headers.get(MqttHeaders.RECEIVED_TOPIC).toString();
        byte[] payload = (byte[])message.getPayload();
        log.debug("received topic :{} \t payload :{}", topic, new String(payload));
        // 根据消息topic不同，将消息转发到不同的channel中。
        //  thing/product/{device_sn}/osd 主题消息会被转发到ChannelName.INBOUND_OSD通道中。
        DeviceTopicEnum topicEnum = DeviceTopicEnum.find(topic);
        MessageChannel bean = (MessageChannel) SpringBeanUtils.getBean(topicEnum.getBeanName());
        return Collections.singleton(bean);
    }
}

上报固件版本号的消息被转发到ChannelName.INBOUND_OSD中后，通过

public IntegrationFlow osdRouterFlow() {
    return IntegrationFlows
        .from(ChannelName.INBOUND_OSD)
        .transform(Message.class, source -> {
            try {
                TopicOsdRequest response = Common.getObjectMapper().readValue((byte[]) source.getPayload(), new TypeReference<TopicOsdRequest>() {});
                String topic = String.valueOf(source.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
                // 填充请求消息中的gateway_sn
                return response.setFrom(topic.substring((THING_MODEL_PRE + PRODUCT).length(), topic.indexOf(OSD_SUF)));
            } catch (IOException e) {
                throw new CloudSDKException(e);
            }
        }, null)
        .<TopicOsdRequest>handle((response, headers) -> {
            // 从sdk包封装好的消息中获取gateway消息
            GatewayManager gateway = SDKManager.getDeviceSDK(response.getGateway());
            OsdDeviceTypeEnum typeEnum = OsdDeviceTypeEnum.find(gateway.getType(), response.getFrom().equals(response.getGateway()));
            Map<String, Object> data = (Map<String, Object>) response.getData();
            // 如果设备不是网关设备，则获取设备的payload信息或者填充空的payload信息
            if (!typeEnum.isGateway()) {
                List payloadData = (List) data.getOrDefault(PayloadModelEnum.PAYLOAD_KEY, new ArrayList<>());
                PayloadModelEnum.getAllIndexWithPosition().stream().filter(data::containsKey)
                    .map(data::get).forEach(payloadData::add);
                data.put(PayloadModelEnum.PAYLOAD_KEY, payloadData);
            }
            // 通过设备的类型，将上报的osd信息序列化为不同的实体
            return response.setData(Common.getObjectMapper().convertValue(data, typeEnum.getClassType()));
        })
        // 根据不同的设备类型，将上报的osd信息交给不同的通道处理
        .<TopicOsdRequest, OsdDeviceTypeEnum>route(response -> OsdDeviceTypeEnum.find(response.getData().getClass()),
            mapping -> Arrays.stream(OsdDeviceTypeEnum.values()).forEach(key -> mapping.channelMapping(key, key.getChannelName())))
        .get();
}

osd处理后的消息根据设备类型不同，会交给不同的通道进行处理，通道可能如下：ChannelName.INBOUND_OSD_RC、ChannelName.INBOUND_OSD_DOCK、ChannelName.INBOUND_OSD_RC_DRONE、ChannelName.INBOUND_OSD_DOCK_DRONE。此处以ChannelName.INBOUND_OSD_RC为例，消息会交给osd封装的接口处理：AbstractDeviceService#osdRemoteControl

@ServiceActivator(inputChannel = ChannelName.INBOUND_OSD_RC)
public void osdRemoteControl(TopicOsdRequest<OsdRemoteControl> request, MessageHeaders headers) {
    throw new UnsupportedOperationException("osdRemoteControl not implemented");
}

@Override
public void osdRemoteControl(TopicOsdRequest<OsdRemoteControl> request, MessageHeaders headers) {
    // 获取消息的上报者sn
    String from = request.getFrom();
    Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(from);
    // 如果设备不在线，直接返回
    if (deviceOpt.isEmpty()) {
        deviceOpt = deviceService.getDeviceBySn(from);
        if (deviceOpt.isEmpty()) {
            log.error("Please restart the drone.");
            return;
        }
    }
    DeviceDTO device = deviceOpt.get();
    // 收到了设备上报的osd信息，延长设备在线的时间
    deviceRedisService.setDeviceOnline(device);
    OsdRemoteControl data = request.getData();
    // 将设备上报的osd信息广播给同一工作空间下的Pilot设备。
    deviceService.pushOsdDataToPilot(device.getWorkspaceId(), from,
        new DeviceOsdHost()
            .setLatitude(data.getLatitude())
            .setLongitude(data.getLongitude())
            .setHeight(data.getHeight()));
    deviceService.pushOsdDataToWeb(device.getWorkspaceId(), BizCodeEnum.RC_OSD, from, data);
}

4.8.3 开发注意事项

1、在Pilot2上加载“态势感知模块”之前，需要先加载API模块和WS模块。