package com.ard.sample.manage.service.impl; import com.ard.common.core.redis.RedisConst; import com.ard.common.core.redis.RedisOpsUtils; import com.ard.sample.component.mqtt.model.CommonTopicReceiver; import com.ard.sample.component.websocket.config.ConcurrentWebSocketSession; import com.ard.sample.component.websocket.enums.MessageType; import com.ard.sample.component.websocket.handler.DockWebSocketHandler; import com.ard.sample.component.websocket.model.BizCodeEnum; import com.ard.sample.component.websocket.model.CustomWebSocketMessage; import com.ard.sample.manage.model.dto.DeviceDTO; import com.ard.sample.manage.model.dto.DevicePayloadDTO; import com.ard.sample.manage.model.dto.TelemetryDTO; import com.ard.sample.manage.model.dto.TelemetryDeviceDTO; import com.ard.sample.manage.model.enums.DeviceDomainEnum; import com.ard.sample.manage.model.receiver.OsdPayloadReceiver; import com.ard.sample.manage.model.receiver.OsdSubDeviceReceiver; import com.ard.sample.manage.service.IDeviceService; import com.ard.sample.manage.service.IWaylineTrajectoryService; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Service; import java.util.Collection; import java.util.List; import java.util.Map; import java.util.stream.Collectors; /** * @author sean * @version 0.3 * @date 2022/2/21 */ @Service @Slf4j public class DeviceOSDServiceImpl extends AbstractTSAService { @Autowired private IDeviceService _deviceService; @Autowired private IWaylineTrajectoryService _waylineTrajectoryService; @Autowired private DockWebSocketHandler _dockWebSocketHandler; protected DeviceOSDServiceImpl(@Autowired @Qualifier("dockOSDServiceImpl") AbstractTSAService tsaService) { super(tsaService); } @Override public void pushTelemetryData(Collection sessions, CustomWebSocketMessage message, Object osdData) { if (osdData instanceof OsdSubDeviceReceiver) { OsdSubDeviceReceiver data = (OsdSubDeviceReceiver) osdData; TelemetryDTO telemetry = message.getData(); telemetry.setHost(TelemetryDeviceDTO.builder() .latitude(data.getLatitude()) .longitude(data.getLongitude()) .altitude(data.getElevation()) .attitudeHead(data.getAttitudeHead()) .elevation(data.getElevation()) .horizontalSpeed(data.getHorizontalSpeed()) .verticalSpeed(data.getVerticalSpeed()) .build()); this.sendMessageService.sendBatch(sessions, message); } } @Override public void handleOSD(CommonTopicReceiver receiver, DeviceDTO device, Collection webSessions, CustomWebSocketMessage wsMessage) { if (DeviceDomainEnum.SUB_DEVICE.getVal() == device.getDomain()) { wsMessage.setBizCode(BizCodeEnum.DEVICE_OSD.getCode()); OsdSubDeviceReceiver data = mapper.convertValue(receiver.getData(), OsdSubDeviceReceiver.class); List payloadsList = device.getPayloadsList(); //data.setChildDeviceSn(device.getDeviceSn()); String dockOsd = ""; receiver.setType(MessageType.STATUS.getType()+"-aircraft"); // receiver.setData(data); try { dockOsd=mapper.writeValueAsString(receiver); JsonNode jsonNode = mapper.readTree(dockOsd); //在JSON的data下插入childDeviceSn,飞行器的sn if(jsonNode.get("data") instanceof ObjectNode){ ObjectNode objectNode = (ObjectNode) jsonNode.get("data"); objectNode.put("childDeviceSn", device.getDeviceSn()); } if(jsonNode.has("data")){ if(jsonNode.get("data").has("mode_code")) { Long code = jsonNode.get("data").get("mode_code").asLong(); if ("0,1,2,3,4,5,8,9,10,11,13,14".contains(code.toString())) { _deviceService.updateWorkType(device.getDeviceSn(), code); RedisOpsUtils.set(RedisConst.OSD_STATUS_PREFIX + device.getDeviceSn(), code); } } } dockOsd=mapper.writeValueAsString(jsonNode); Map receiverData = (Map) receiver.getData(); data.setPayloads(payloadsList.stream() .map(payload -> mapper.convertValue( receiverData.getOrDefault(payload.getPayloadName(), Map.of()), OsdPayloadReceiver.class)) .collect(Collectors.toList())); } catch (NullPointerException | JsonProcessingException e) { log.warn("Please remount the payload, or restart the drone. Otherwise the data of the payload will not be received."); } RedisOpsUtils.setWithExpire(RedisConst.OSD_PREFIX + device.getDeviceSn(), data, RedisConst.DEVICE_ALIVE_SECOND); wsMessage.getData().setHost(data); // 发送给websocket客户端 _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd, ""); // 将飞行器的OSD数据dockOsd入库; _waylineTrajectoryService.setList(device.getDeviceSn(),"0", dockOsd); sendMessageService.sendBatch(webSessions, wsMessage); this.pushTelemetryData(data, device.getDeviceSn()); } tsaService.handleOSD(receiver, device, webSessions, wsMessage); } }