a-tuoheng-device/src/main/java/com/ruoyi/device/service/impl/TuohengService.java

package com.ruoyi.device.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.device.domain.api.IDockAircraftDomain;
import com.ruoyi.device.domain.api.IDockDomain;
import com.ruoyi.device.domain.api.IAircraftDomain;
import com.ruoyi.device.domain.api.IDeviceDomain;
import com.ruoyi.device.domain.impl.machine.mqtt.MqttCallbackRegistry;
import com.ruoyi.device.domain.impl.machine.state.CoverState;
import com.ruoyi.device.domain.impl.machine.state.DroneState;
import com.ruoyi.device.domain.impl.machine.state.AirportState;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
import com.ruoyi.device.domain.impl.machine.statemachine.MachineStateManager;
import com.ruoyi.device.domain.impl.tuohengmqtt.callback.IRealTimeBasicCallback;
import com.ruoyi.device.domain.impl.tuohengmqtt.callback.ITuohengEventsCallback;
import com.ruoyi.device.domain.impl.tuohengmqtt.callback.ITuohengOsdCallback;
import com.ruoyi.device.domain.impl.tuohengmqtt.callback.ITuohengRealTimeDataCallback;
import com.ruoyi.device.domain.impl.tuohengmqtt.config.TuohengMqttClientConfig;
import com.ruoyi.device.domain.impl.tuohengmqtt.handler.TuohengMqttMessageHandler;
import com.ruoyi.device.domain.impl.tuohengmqtt.manager.TuohengMqttClientManager;
import com.ruoyi.device.domain.impl.tuohengmqtt.model.AirportOsdData;
import com.ruoyi.device.domain.impl.tuohengmqtt.model.EventsData;
import com.ruoyi.device.domain.impl.tuohengmqtt.model.RealTimeBasicData;
import com.ruoyi.device.domain.impl.tuohengmqtt.model.TuohengRealTimeData;
import com.ruoyi.device.domain.model.Aircraft;
import com.ruoyi.device.domain.model.Device;
import com.ruoyi.device.domain.model.Dock;
import com.ruoyi.device.domain.model.DockAircraft;
import com.ruoyi.device.service.config.TuohengMqttProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
@Slf4j
public class TuohengService {

    @Autowired
    private TuohengMqttClientManager clientManager;

    @Autowired
    private TuohengMqttProperties mqttProperties;

    @Autowired
    private IDockAircraftDomain dockAircraftDomain;

    @Autowired
    private IDockDomain dockDomain;

    @Autowired
    private IAircraftDomain aircraftDomain;

    @Autowired
    private IDeviceDomain deviceDomain;

    @Autowired
    private MachineStateManager stateManager;

    @Autowired
    private MqttCallbackRegistry mqttCallbackRegistry;

    private final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 机场心跳时间戳记录 (deviceSn -> lastHeartbeatTime)
     */
    private final Map<String, Long> airportHeartbeatMap = new java.util.concurrent.ConcurrentHashMap<>();

    /**
     * 心跳超时时间：5分钟
     */
    private static final long HEARTBEAT_TIMEOUT = 5 * 60 * 1000;

    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationReady() {
        TuohengMqttClientConfig config = TuohengMqttClientConfig.builder()
                .host(mqttProperties.getHost())
                .port(mqttProperties.getPort())
                .clientId(mqttProperties.getClientId())
                .username(mqttProperties.getUsername())
                .password(mqttProperties.getPassword())
                .connectionTimeout(mqttProperties.getConnectionTimeout())
                .keepAliveInterval(mqttProperties.getKeepAliveInterval())
                .autoReconnect(mqttProperties.getAutoReconnect())
                .cleanSession(mqttProperties.getCleanSession())
                .useSharedSubscription(true)
                .sharedGroupName("tuoheng-group")
                .build();

        clientManager.initClient(config);

        TuohengMqttMessageHandler handler = clientManager.getHandler();

        // 设置 MqttCallbackRegistry 到 handler（用于指令回调）
        handler.setMqttCallbackRegistry(mqttCallbackRegistry);

        Map<String, String> mapping = loadAirportDroneMapping();

        handler.registerRealTimeDataCallback(new ITuohengRealTimeDataCallback() {
            @Override
            public void onRealTimeData(String deviceSn, TuohengRealTimeData data) {
//                log.info("========== 收到拓恒实时数据 ==========");
//                log.info("收到拓恒实时数据 设备SN: {}", deviceSn);
                try {
//                    log.info("数据内容: {}", objectMapper.writeValueAsString(data));

                    // 更新机场心跳时间戳
                    updateAirportHeartbeat(deviceSn);

                    // 同步无人机开关机状态
                    syncDronePowerState(deviceSn, data);

                    // 同步舱门状态
                    syncCoverStateFromRealTimeData(deviceSn, data);

                } catch (Exception e) {
                    log.error("处理实时数据失败", e);
                }
//                log.info("=====================================");
            }
        });

        handler.registerOsdCallback(new ITuohengOsdCallback() {
            @Override
            public void onOsdData(String deviceSn, AirportOsdData data) {
//                log.info("========== 收到拓恒OSD数据 ==========");
//                log.info("收到拓恒OSD数据 设备SN: {}", deviceSn);
                try {
//                    log.info("数据内容: {}", objectMapper.writeValueAsString(data));

                    // 同步飞行状态到 MachineStateManager
                    syncFlightState(deviceSn, data);

                } catch (Exception e) {
                    log.error("序列化数据失败", e);
                }
//                log.info("=====================================");
            }
        });

        handler.registerEventsCallback(new ITuohengEventsCallback() {
            @Override
            public void onEventsData(String deviceSn, EventsData data) {
//                log.info("========== 收到拓恒Events数据 ==========");
//                log.info("收到拓恒Events数据 设备SN: {}", deviceSn);
                try {
//                    log.info("数据内容: {}", objectMapper.writeValueAsString(data));
                } catch (Exception e) {
                    log.error("序列化数据失败", e);
                }
//                log.info("=====================================");
            }
        });

        handler.registerRealTimeBasicCallback(new IRealTimeBasicCallback() {
            @Override
            public void onRealTimeBasicData(String deviceSn, RealTimeBasicData data) {
                try {
                    // 同步舱门状态到 MachineStateManager
                    syncCoverState(deviceSn, data);
                } catch (Exception e) {
                    log.error("处理实时基础数据失败", e);
                }
            }
        });

        handler.registerHeartbeatMessageCallback(new com.ruoyi.device.domain.impl.tuohengmqtt.callback.IHeartbeatMessageCallback() {
            @Override
            public void onHeartbeatMessage(String deviceSn, com.ruoyi.device.domain.impl.tuohengmqtt.model.HeartbeatMessageData data) {
                try {
                    // 同步舱门状态到 MachineStateManager
                    syncCoverStateFromHeartbeat(deviceSn, data);
                } catch (Exception e) {
                    log.error("处理心跳消息失败", e);
                }
            }
        });

        log.info("TuohengService 初始化完成，已注册所有回调");
    }

    private Map<String, String> loadAirportDroneMapping() {
        Map<String, String> mapping = new HashMap<>();

        try {
            List<DockAircraft> dockAircraftList = dockAircraftDomain.selectDockAircraftList(new DockAircraft());

            for (DockAircraft dockAircraft : dockAircraftList) {
                Dock dock = dockDomain.selectDockByDockId(dockAircraft.getDockId());
                Aircraft aircraft = aircraftDomain.selectAircraftByAircraftId(dockAircraft.getAircraftId());

                if (dock != null && aircraft != null) {
                    Device dockDevice = deviceDomain.selectDeviceByDeviceId(dock.getDeviceId());
                    Device aircraftDevice = deviceDomain.selectDeviceByDeviceId(aircraft.getDeviceId());

                    if (dockDevice != null && aircraftDevice != null) {
                        String airportSn = dockDevice.getDeviceSn();
                        String droneSn = aircraftDevice.getDeviceSn();

                        if (airportSn != null && droneSn != null) {
                            if (airportSn.startsWith("THJS") || droneSn.startsWith("THJS")) {
                                log.debug("跳过大疆设备 - 机场SN: {}, 无人机SN: {}", airportSn, droneSn);
                                continue;
                            }
                            mapping.put(airportSn, droneSn);
                            log.info("加载机场-无人机映射: {} -> {}", airportSn, droneSn);
                        }
                    }
                }
            }

            log.info("从数据库加载机场-无人机映射完成，共 {} 条记录", mapping.size());
        } catch (Exception e) {
            log.error("从数据库加载机场-无人机映射失败", e);
        }

        return mapping;
    }


    /**
     * 同步飞行状态到 MachineStateManager
     * 根据 OSD 数据中的 flighttask_step_code 和 mode_code 判断飞行状态
     */
    private void syncFlightState(String deviceSn, AirportOsdData data) {
        try {
            if (data == null) {
                return;
            }

            String flighttaskStepCode = data.getFlighttaskStepCode();
            String modeCode = data.getModeCode();

            // 同步无人机状态
            DroneState droneState = determineDroneState(flighttaskStepCode, modeCode);
            if (droneState != null) {
                stateManager.setDroneState(deviceSn, droneState);
                log.debug("同步飞行状态: sn={}, flighttaskStepCode={}, modeCode={}, state={}",
                    deviceSn, flighttaskStepCode, modeCode, droneState);
            }

            // 注意：机场在线状态由 IOT 平台的心跳机制判断（5分钟超时）
            // 不在这里简单地根据收到数据就判断为在线

        } catch (Exception e) {
            log.error("同步飞行状态失败: sn={}", deviceSn, e);
        }
    }

    /**
     * 根据任务状态码和模式码判断无人机状态
     */
    private DroneState determineDroneState(String flighttaskStepCode, String modeCode) {
        // 优先根据 flighttask_step_code 判断
        if (flighttaskStepCode != null) {
            switch (flighttaskStepCode) {
                case "1":
                    // 飞行作业中
                    return DroneState.FLYING;
                case "2":
                    // 作业后状态恢复，可能是返航或已到达
                    return DroneState.RETURNING;
                case "5":
                    // 任务空闲
                    return DroneState.ONLINE;
                case "255":
                    // 飞行器异常
                    return DroneState.UNKNOWN;
            }
        }

        // 根据 mode_code 辅助判断
        if (modeCode != null) {
            if (modeCode.equals("3") || modeCode.equals("4") || modeCode.equals("5")) {
                // 飞行中状态
                return DroneState.FLYING;
            }
        }

        return null;
    }

    /**
     * 更新机场心跳时间戳并设置在线状态
     */
    private void updateAirportHeartbeat(String deviceSn) {
        long currentTime = System.currentTimeMillis();
        airportHeartbeatMap.put(deviceSn, currentTime);

        // 收到心跳，设置机场为在线
        stateManager.setAirportState(deviceSn, AirportState.ONLINE);
        log.debug("更新机场心跳: sn={}, time={}", deviceSn, currentTime);
    }

    /**
     * 同步无人机开关机状态
     */
    private void syncDronePowerState(String deviceSn, TuohengRealTimeData data) {
        try {
            if (data == null || data.getDroneBattery() == null || data.getDroneBattery().getData() == null) {
                return;
            }

            Integer powerOn = data.getDroneBattery().getData().getBPowerON();
            if (powerOn == null) {
                return;
            }

            log.info("【状态同步】收到无人机电源状态: sn={}, bPowerON={}", deviceSn, powerOn);

            // 根据 bPowerON 值更新无人机状态
            // 1 = 开机, 2 = 关机
            if (powerOn == 1) {
                // 开机时，只有在特定状态下才更新为 ONLINE
                // 只有从 POWER_OFF、UNKNOWN、ARRIVED、RETURNING 这些状态才能转为 ONLINE
                MachineStates currentStates = stateManager.getStates(deviceSn);
                if (currentStates != null && currentStates.getDroneState() != null) {
                    DroneState currentState = currentStates.getDroneState();
                    if (currentState == DroneState.POWER_OFF ||
                        currentState == DroneState.UNKNOWN ||
                        currentState == DroneState.ARRIVED ||
                        currentState == DroneState.RETURNING) {
                        stateManager.setDroneState(deviceSn, DroneState.ONLINE);
                        log.info("【状态同步】同步无人机开机状态: sn={}, powerOn={}, 从 {} 更新为 ONLINE",
                            deviceSn, powerOn, currentState);
                    } else {
                        log.debug("无人机已开机，但当前状态为 {}，不更新为 ONLINE", currentState);
                    }
                } else {
                    // 如果没有当前状态，直接设置为 ONLINE
                    stateManager.setDroneState(deviceSn, DroneState.ONLINE);
                    log.debug("同步无人机开机状态: sn={}, powerOn={}, 初始化为 ONLINE", deviceSn, powerOn);
                }
            } else if (powerOn == 2) {
                stateManager.setDroneState(deviceSn, DroneState.POWER_OFF);
                log.info("【状态同步】同步无人机关机状态: sn={}, powerOn={}", deviceSn, powerOn);
            }

        } catch (Exception e) {
            log.error("同步无人机开关机状态失败: sn={}", deviceSn, e);
        }
    }

    /**
     * 从 realTime/data 主题同步舱门状态
     */
    private void syncCoverStateFromRealTimeData(String deviceSn, TuohengRealTimeData data) {
        try {
            if (data == null || data.getNestDoor() == null || data.getNestDoor().getData() == null) {
                return;
            }

            Integer status = data.getNestDoor().getData().getStatus();
            if (status == null) {
                return;
            }

            log.info("【状态同步】收到舱门状态: sn={}, status={}", deviceSn, status);

            // 根据 status 值更新舱门状态
            // 1 = 开仓, 2 = 关仓
            CoverState coverState;
            if (status == 1) {
                coverState = CoverState.OPENED;
                log.info("【状态同步】同步舱门开启状态: sn={}, status={}", deviceSn, status);
            } else if (status == 2) {
                coverState = CoverState.CLOSED;
                log.info("【状态同步】同步舱门关闭状态: sn={}, status={}", deviceSn, status);
            } else {
                log.warn("未知的舱门状态值: sn={}, status={}", deviceSn, status);
                return;
            }

            stateManager.setCoverState(deviceSn, coverState);

        } catch (Exception e) {
            log.error("同步舱门状态失败: sn={}", deviceSn, e);
        }
    }

    /**
     * 同步舱门状态（从 realTime/basic 主题，已废弃）
     */
    private void syncCoverState(String deviceSn, com.ruoyi.device.domain.impl.tuohengmqtt.model.RealTimeBasicData data) {
        try {
            if (data == null || data.getData() == null || data.getData().getNestDoor() == null) {
                return;
            }

             RealTimeBasicData.NestDoorInfo nestDoor = data.getData().getNestDoor();
            if (nestDoor.getData() == null) {
                return;
            }

            Integer status = nestDoor.getData().getStatus();
            if (status == null) {
                return;
            }

            log.info("【状态同步】收到舱门状态: sn={}, status={}", deviceSn, status);

            // 根据 status 值更新舱门状态
            // 1 = 开仓, 2 = 关仓
            CoverState coverState;
            if (status == 1) {
                coverState = CoverState.OPENED;
                log.info("【状态同步】同步舱门开启状态: sn={}, status={}", deviceSn, status);
            } else if (status == 2) {
                coverState = CoverState.CLOSED;
                log.info("【状态同步】同步舱门关闭状态: sn={}, status={}", deviceSn, status);
            } else {
                log.warn("未知的舱门状态值: sn={}, status={}", deviceSn, status);
                return;
            }

            stateManager.setCoverState(deviceSn, coverState);

        } catch (Exception e) {
            log.error("同步舱门状态失败: sn={}", deviceSn, e);
        }
    }

    /**
     * 从 heartbeat/message 主题同步舱门状态
     */
    private void syncCoverStateFromHeartbeat(String deviceSn, com.ruoyi.device.domain.impl.tuohengmqtt.model.HeartbeatMessageData data) {
        try {
            if (data == null || data.getNestDoor() == null || data.getNestDoor().getData() == null) {
                return;
            }

            Integer status = data.getNestDoor().getData().getStatus();
            if (status == null) {
                return;
            }

            log.info("【状态同步】收到舱门状态(heartbeat): sn={}, status={}", deviceSn, status);

            // 根据 status 值更新舱门状态
            // 1 = 开仓, 2 = 关仓
            CoverState coverState;
            if (status == 1) {
                coverState = CoverState.OPENED;
                log.info("【状态同步】同步舱门开启状态(heartbeat): sn={}, status={}", deviceSn, status);
            } else if (status == 2) {
                coverState = CoverState.CLOSED;
                log.info("【状态同步】同步舱门关闭状态(heartbeat): sn={}, status={}", deviceSn, status);
            } else {
                log.warn("未知的舱门状态值(heartbeat): sn={}, status={}", deviceSn, status);
                return;
            }

            stateManager.setCoverState(deviceSn, coverState);

        } catch (Exception e) {
            log.error("同步舱门状态失败(heartbeat): sn={}", deviceSn, e);
        }
    }

    /**
     * 定时检查机场心跳超时
     * 每分钟执行一次
     */
    @Scheduled(fixedRate = 60000)
    public void checkAirportHeartbeatTimeout() {
        long currentTime = System.currentTimeMillis();

        for (Map.Entry<String, Long> entry : airportHeartbeatMap.entrySet()) {
            String deviceSn = entry.getKey();
            Long lastHeartbeatTime = entry.getValue();

            long timeDiff = currentTime - lastHeartbeatTime;

            if (timeDiff > HEARTBEAT_TIMEOUT) {
                // 超时，设置为离线
                stateManager.setAirportState(deviceSn, AirportState.OFFLINE);
                log.warn("机场心跳超时，设置为离线: sn={}, 超时时长={}秒",
                    deviceSn, timeDiff / 1000);
            }
        }
    }
}