无人机的端侧AI推理实战：从目标检测到自主避障的实时工程方案

左手厨刀右手茼蒿

295人浏览 · 2026-07-23 07:35:59

左手厨刀右手茼蒿 · 2026-07-23 07:35:59 发布

无人机的端侧AI推理实战：从目标检测到自主避障的实时工程方案

一、无人机自主飞行的基础约束：计算资源与实时性的双重瓶颈

无人机的自主避障需要在100ms内完成"感知-决策-执行"闭环——摄像头或激光雷达采集环境数据，AI模型识别障碍物位置和类型，规划算法生成避障路径，飞控系统调整姿态和速度。100ms的上限来自物理约束：无人机以10m/s速度飞行时，100ms内移动1米——超过这个延迟，障碍物可能在决策完成前就已经碰撞。

端侧部署的第二个约束是计算资源。无人机常用的计算平台有三种：STM32系列MCU（算力约100DMIPS，只够跑规则算法）、Jetson Nano/Xavier（GPU算力0.5-20TOPS，能跑轻量级CNN）、RK3588等国产SoC（NPU算力6TOPS，支持INT8量化推理）。重量和功耗是硬约束——计算板卡增加50克重量就意味着续航减少3-5分钟，功耗超过5W就需要更大的电池和散热面积。

端侧AI的工程挑战不是"模型能不能跑"，而是"模型推理延迟+数据预处理+后处理+通信开销的总和是否在100ms以内"。一个YOLOv5s模型在Jetson Nano上的单帧推理时间是30-40ms，看起来足够。但加上图像解码（10ms）、NMS后处理（5ms）、结果传输到飞控（2ms）、飞控执行延迟（10ms），总计57ms——留给规划算法的时间只剩43ms。如果摄像头分辨率从640x480升级到1280x720，推理时间翻倍到60-80ms，闭环延迟立刻超标。

二、端侧AI推理的完整数据流与优化架构

推理层的核心优化是模型量化。YOLOv5s从FP32量化到INT8后，推理时间从40ms降到12-20ms（TensorRT加速后更低）。量化的精度损失在无人机场景中可以接受——避障只需要检测障碍物的大致位置和类别（树/建筑/人），不需要精确的边界框坐标。FP32模型的检测mAP约为0.35，INT8量化后约为0.32，差异仅影响小目标检测——远距离的小障碍物本身就不在避障的有效距离范围内。

感知层的深度数据对齐是关键工程环节。摄像头给出2D像素坐标的障碍物位置，ToF雷达给出深度矩阵的3D距离信息。两者的视场角和分辨率不同——摄像头FOV 60度，ToF FOV 45度。对齐策略：将2D检测框的像素坐标映射到ToF深度矩阵的对应区域，取该区域的最小深度值作为障碍物距离。映射精度取决于两个传感器的安装间距和角度偏差——间距过大时需要做立体校正。

决策层的威胁等级判定直接决定避障策略的选择。三级威胁模型：Safe（距离>5m，继续巡航）、Warning（3-5m，减速并准备避障）、Critical（<3m，立即执行避障动作）。Critical级别的决策跳过路径规划直接输出急停或简单规避指令——因为路径规划的43ms预算在<3m距离内不够用（10m/s速度下，3m距离只有300ms的碰撞时间，扣除感知和推理的50ms后只剩250ms）。

三、无人机端侧AI推理的生产级代码实现

# drone_edge_ai_pipeline.py
# 无人机端侧AI推理与自主避障的完整Pipeline

import numpy as np
from dataclasses import dataclass, field
from typing import Optional, List
from enum import Enum
import time

class ThreatLevel(Enum):
    SAFE = "safe"       # >5m, 继续巡航
    WARNING = "warning"  # 3-5m, 减速准备
    CRITICAL = "critical"  # <3m, 立即避障

class AvoidAction(Enum):
    CONTINUE = "continue"
    DECELERATE = "decelerate"
    AVOID_LEFT = "avoid_left"
    AVOID_RIGHT = "avoid_right"
    AVOID_UP = "avoid_up"
    EMERGENCY_STOP = "emergency_stop"

@dataclass
class Detection:
    class_name: str        # tree | building | person | vehicle
    bbox: tuple            # (x1, y1, x2, y2) 像素坐标
    confidence: float
    distance_m: float      # 障碍物距离(米)
    direction: str         # left | center | right

@dataclass
class ObstacleMap:
    detections: List[Detection]
    nearest_distance: float
    threat_level: ThreatLevel
    safe_directions: List[str]

@dataclass
class FlightCommand:
    action: AvoidAction
    vx: float              # 前向速度 m/s
    vy: float              # 横向速度 m/s (正=右)
    vz: float              # 垂直速度 m/s (正=上)
    yaw_rate: float        # 偏航角速度 deg/s

class ImagePreprocessor:
    """图像预处理Pipeline：解码+缩放+归一化"""

    def __init__(self, model_input_size: tuple = (640, 480)):
        self.model_size = model_input_size

    def preprocess(self, raw_frame: np.ndarray,
                   depth_matrix: Optional[np.ndarray] = None
                   ) -> tuple:
        """图像预处理：色彩转换+缩放+归一化"""
        start = time.time()

        # YUV422→RGB转换（如果输入是YUV格式）
        if raw_frame.shape[2] == 2:  # YUV422: 2通道
            rgb = self._yuv422_to_rgb(raw_frame)
        else:
            rgb = raw_frame

        # 缩放到模型输入尺寸
        resized = self._bilinear_resize(rgb, self.model_size)

        # 归一化: [0,255]→[0,1]
        normalized = resized.astype(np.float32) / 255.0

        preprocess_time = (time.time() - start) * 1000

        return normalized, depth_matrix, preprocess_time

    def _yuv422_to_rgb(self, yuv: np.ndarray) -> np.ndarray:
        """YUV422格式转RGB"""
        h, w = yuv.shape[:2]
        y = yuv[:, :, 0].astype(np.float32)
        u = yuv[:, :, 1].astype(np.float32) if yuv.shape[2] > 1 \
            else np.zeros_like(y)

        # YUV→RGB简化转换
        r = np.clip(y + 1.4 * (u - 128), 0, 255)
        g = np.clip(y - 0.34 * (u - 128), 0, 255)
        b = np.clip(y + 1.77 * (u - 128), 0, 255)

        rgb = np.stack([r, g, b], axis=-1).astype(np.uint8)
        return rgb

    def _bilinear_resize(self, img: np.ndarray,
                          target: tuple) -> np.ndarray:
        """双线性插值缩放"""
        th, tw = target
        ih, iw = img.shape[:2]

        # 计算缩放坐标映射
        y_ratio = ih / th
        x_ratio = iw / tw

        y_coords = np.arange(th) * y_ratio
        x_coords = np.arange(tw) * x_ratio

        # 简化实现：最近邻插值（更快）
        y_idx = np.clip(np.round(y_coords).astype(int), 0, ih - 1)
        x_idx = np.clip(np.round(x_coords).astype(int), 0, iw - 1)

        return img[y_idx][:, x_idx]

class DepthAligner:
    """深度数据与2D检测框的对齐"""

    def __init__(self, cam_fov_deg: float = 60,
                 tof_fov_deg: float = 45,
                 cam_res: tuple = (640, 480),
                 tof_res: tuple = (100, 100)):
        self.cam_fov = cam_fov_deg
        self.tof_fov = tof_fov_deg
        self.cam_res = cam_res
        self.tof_res = tof_res

    def align_detection_depth(self, detection: Detection,
                               depth_matrix: np.ndarray
                               ) -> Detection:
        """将2D检测框映射到ToF深度矩阵获取距离"""
        if depth_matrix is None:
            return detection

        # 计算检测框在ToF矩阵中的对应区域
        x1, y1, x2, y2 = detection.bbox

        # 像素坐标→角度坐标
        cam_hfov = self.cam_fov
        cam_vfov = cam_hfov * self.cam_res[1] / self.cam_res[0]

        angle_x1 = (x1 / self.cam_res[0] - 0.5) * cam_hfov
        angle_x2 = (x2 / self.cam_res[0] - 0.5) * cam_hfov
        angle_y1 = (y1 / self.cam_res[1] - 0.5) * cam_vfov
        angle_y2 = (y2 / self.cam_res[1] - 0.5) * cam_vfov

        # 角度坐标→ToF矩阵索引
        tof_hfov = self.tof_fov
        tof_vfov = tof_hfov * self.tof_res[1] / self.tof_res[0]

        # 角度偏移校正（ToF和cam的FOV中心对齐）
        offset = (cam_hfov - tof_hfov) / 2

        tof_x1 = int((angle_x1 + offset + tof_hfov / 2)
                      / tof_hfov * self.tof_res[0])
        tof_x2 = int((angle_x2 + offset + tof_hfov / 2)
                      / tof_hfov * self.tof_res[0])
        tof_y1 = int((angle_y1 + tof_vfov / 2)
                      / tof_vfov * self.tof_res[1])
        tof_y2 = int((angle_y2 + tof_vfov / 2)
                      / tof_vfov * self.tof_res[1])

        # 取深度矩阵对应区域的最小值（最近距离）
        tof_x1 = max(0, min(tof_x1, self.tof_res[0] - 1))
        tof_x2 = max(0, min(tof_x2, self.tof_res[0] - 1))
        tof_y1 = max(0, min(tof_y1, self.tof_res[1] - 1))
        tof_y2 = max(0, min(tof_y2, self.tof_res[1] - 1))

        region = depth_matrix[tof_y1:tof_y2+1, tof_x1:tof_x2+1]
        if region.size > 0:
            nearest_dist = np.min(region[region > 0])
            detection.distance_m = nearest_dist

        # 判定方向
        center_x = (x1 + x2) / 2 / self.cam_res[0]
        if center_x < 0.35:
            detection.direction = "left"
        elif center_x > 0.65:
            detection.direction = "right"
        else:
            detection.direction = "center"

        return detection

class AvoidancePlanner:
    """避障路径规划：基于威胁等级的决策"""

    def __init__(self, cruise_speed: float = 5.0,
                 safe_distance: float = 5.0,
                 critical_distance: float = 3.0,
                 avoid_offset: float = 2.0):
        self.cruise_speed = cruise_speed
        self.safe_dist = safe_distance
        self.critical_dist = critical_distance
        self.avoid_offset = avoid_offset

    def plan(self, obstacle_map: ObstacleMap
             ) -> FlightCommand:
        """基于障碍物地图生成飞行指令"""
        if obstacle_map.threat_level == ThreatLevel.SAFE:
            return FlightCommand(
                action=AvoidAction.CONTINUE,
                vx=self.cruise_speed, vy=0, vz=0,
                yaw_rate=0,
            )

        if obstacle_map.threat_level == ThreatLevel.CRITICAL:
            # Critical级别：跳过路径规划，直接执行简单避障
            nearest = min(obstacle_map.detections,
                          key=lambda d: d.distance_m)

            if nearest.direction == "center":
                # 前方障碍：优先选择空间更大的方向
                if "left" in obstacle_map.safe_directions:
                    return FlightCommand(
                        action=AvoidAction.AVOID_LEFT,
                        vx=1.0,
                        vy=-self.avoid_offset,
                        vz=0, yaw_rate=-20,
                    )
                elif "right" in obstacle_map.safe_directions:
                    return FlightCommand(
                        action=AvoidAction.AVOID_RIGHT,
                        vx=1.0,
                        vy=self.avoid_offset,
                        vz=0, yaw_rate=20,
                    )
                elif "up" in obstacle_map.safe_directions:
                    return FlightCommand(
                        action=AvoidAction.AVOID_UP,
                        vx=0.5, vy=0,
                        vz=self.avoid_offset,
                        yaw_rate=0,
                    )
                else:
                    return FlightCommand(
                        action=AvoidAction.EMERGENCY_STOP,
                        vx=0, vy=0, vz=0,
                        yaw_rate=0,
                    )

            elif nearest.direction == "left":
                return FlightCommand(
                    action=AvoidAction.AVOID_RIGHT,
                    vx=2.0, vy=self.avoid_offset,
                    vz=0, yaw_rate=15,
                )
            else:
                return FlightCommand(
                    action=AvoidAction.AVOID_LEFT,
                    vx=2.0, vy=-self.avoid_offset,
                    vz=0, yaw_rate=-15,
                )

        # Warning级别：减速并准备避障
        return FlightCommand(
            action=AvoidAction.DECELERATE,
            vx=2.0, vy=0, vz=0, yaw_rate=0,
        )

    def build_obstacle_map(self, detections: List[Detection]
                            ) -> ObstacleMap:
        """构建障碍物地图和威胁等级"""
        if not detections:
            return ObstacleMap(
                detections=[], nearest_distance=999,
                threat_level=ThreatLevel.SAFE,
                safe_directions=["left", "center", "right", "up"],
            )

        nearest = min(detections, key=lambda d: d.distance_m)
        nearest_dist = nearest.distance_m

        # 判定威胁等级
        if nearest_dist < self.critical_dist:
            threat = ThreatLevel.CRITICAL
        elif nearest_dist < self.safe_dist:
            threat = ThreatLevel.WARNING
        else:
            threat = ThreatLevel.SAFE

        # 计算安全方向
        occupied = set(d.direction for d in detections
                       if d.distance_m < self.safe_dist)
        safe_dirs = []
        if "left" not in occupied:
            safe_dirs.append("left")
        if "right" not in occupied:
            safe_dirs.append("right")
        if "center" not in occupied:
            safe_dirs.append("center")
        # 上方默认安全（除非检测到上方障碍）
        if "up" not in occupied:
            safe_dirs.append("up")

        return ObstacleMap(
            detections=detections,
            nearest_distance=nearest_dist,
            threat_level=threat,
            safe_directions=safe_dirs,
        )

class DroneAIPipeline:
    """无人机端侧AI完整Pipeline"""

    def __init__(self):
        self.preprocessor = ImagePreprocessor()
        self.depth_aligner = DepthAligner()
        self.planner = AvoidancePlanner()
        self.last_latency = {}

    def process_frame(self, raw_frame: np.ndarray,
                      depth_matrix: Optional[np.ndarray] = None,
                      detections_raw: Optional[List] = None
                      ) -> FlightCommand:
        """处理单帧图像，输出飞行指令"""
        cycle_start = time.time()

        # 1. 图像预处理
        tensor, depth, pre_time = self.preprocessor.preprocess(
            raw_frame, depth_matrix
        )

        # 2. 目标检测（模拟推理结果）
        infer_start = time.time()
        detections = detections_raw or []
        infer_time = (time.time() - infer_start) * 1000

        # 3. 深度对齐
        for det in detections:
            det = self.depth_aligner.align_detection_depth(
                det, depth
            )

        # 4. 避障决策
        obstacle_map = self.planner.build_obstacle_map(detections)
        command = self.planner.plan(obstacle_map)

        # 5. 延迟统计
        total_ms = (time.time() - cycle_start) * 1000
        self.last_latency = {
            "preprocess_ms": pre_time,
            "inference_ms": infer_time,
            "total_cycle_ms": total_ms,
            "threat_level": obstacle_map.threat_level.value,
            "nearest_distance": obstacle_map.nearest_distance,
        }

        return command

四、端侧AI推理落地的关键决策与工程误区

第一个误区是"追求检测模型的最高精度"。避障场景对检测精度有两个放宽：远距离小目标（>10m）的检测精度不重要，因为无人机有足够的反应时间；边界框的精确坐标不重要，只需要大致方位判断左/中/右方向。INT8量化导致mAP下降3个百分点，但这些下降集中在小目标和精确边界框上——对避障决策几乎没有影响。YOLOv5s-INT8+TensorRT在Jetson Nano上的12ms推理时间是关键，精度损失是可接受的代价。

第二个误区是"只用摄像头做深度估计"。单目深度估计模型在无人机场景中不可靠——其精度依赖于训练数据的场景覆盖度，户外环境的复杂度远超室内训练集。双目立体视觉的深度精度受限于基线距离——无人机上两个摄像头的间距通常<10cm，超过5m距离后深度误差急剧增大。ToF激光雷达是可靠的深度来源——精度不依赖场景纹理和基线距离，5m范围内误差<5%。摄像头+ToF的组合是最务实的方案。

第三个误区是"路径规划用复杂算法"。A和RRT等全局路径规划算法的计算时间在100ms预算内难以完成——特别是当障碍物数量增加时。Critical级别（<3m距离）直接跳过路径规划，用简单规则（方向偏移+减速）替代。只有Warning级别（3-5m）才有时间做局部路径规划。工程折中：简单规则处理紧急避障，局部规划处理中长期路径调整。

关键决策是闭环延迟预算的分配。100ms预算的典型分配：预处理10ms、推理12-20ms、深度对齐3ms、NMS后处理5ms、路径规划30-40ms、飞控执行10ms、通信2-5ms。推理时间的压缩直接决定了留给路径规划的时间量——INT8量化+TensorRT从40ms压缩到12ms，路径规划预算从10ms增加到40ms，这是避障系统从"仅急停"升级到"可规划规避"的关键转折点。

五、总结

无人机端侧AI避障系统的闭环延迟预算为100ms（10m/s飞行速度下1米移动距离），预算分配为预处理10ms、INT8量化推理12-20ms、深度对齐3ms、NMS后处理5ms、路径规划30-40ms、飞控执行10ms。YOLOv5s从FP32量化到INT8，推理从40ms压缩到12ms（TensorRT加速），精度损失仅影响小目标检测（对避障决策无实质影响），但路径规划预算从10ms增加到40ms——这是从"仅急停"到"可规划规避"的转折点。深度估计使用摄像头+ToF激光雷达组合而非单目深度模型，2D检测框通过FOV角度映射对齐到ToF深度矩阵获取距离。威胁等级分三级（Safe>5m/Warning3-5m/Critical<3m），Critical级别跳过路径规划直接执行方向偏移规则，Warning级别有40ms预算做局部路径规划。量化精度和推理速度的权衡是端侧AI的核心决策——速度提升释放的规划预算比精度损失更有决策价值。