低空智联网与城市空中交通：通感一体空域管理、多机协同避障与低空安全合规验证体系实战

┌───────────────────────────────────────────────────────────────────────────┐
│                Urban Low-Altitude Intelligent Network                      │
├───────────────────────────────────────────────────────────────────────────┤
│  [Layer 0: 网络物理基座层] ← 5G-A ISAC / Satellite / Mesh / ADS-B        │
│      ↓                                                                    │
│  [Layer 1: 全域态势感知层] ← Multi-Source Fusion / Weather Compensation  │
│   ├─ 5G-A通感一体 + ADS-B + 视觉 + 声学多模态融合                          │
│   ├─ 气象自适应补偿与感知盲区动态补盲                                       │
│   └─ 多目标关联跟踪与身份确认                                               │
│      ↓                                                                    │
│  [Layer 2: 万机协同决策层] ← Distributed CD&R / Intent Sharing           │
│   ├─ 分布式冲突检测（本地+邻域两级）                                        │
│   ├─ 基于意图共享的预测性协同解脱                                            │
│   └─ 跨运营商语义对齐与联邦决策                                              │
│      ↓                                                                    │
│  [Layer 3: 赛博物理安全层] ← CPS Security / Spoofing Detection           │
│   ├─ 多源导航一致性校验（GPS/视觉/惯性/5G定位）                              │
│   ├─ 虚假信号注入检测与隔离                                                 │
│   └─ 赛博物理安全影响分析 + AC-93-TM合规验证                                 │
└───────────────────────────────────────────────────────────────────────────┘

pip install torch numpy scipy filterpy pyproj
# 硬件: 5G-A通感一体基站(华为/中兴) + ADS-B接收机 + 光学追踪相机
#       + 声学阵列 + 边缘融合服务器 (华为Atlas 800)

"""
low_altitude_sensing_system.py - 城市低空全域态势感知系统
技术栈: PyTorch / NumPy / SciPy / filterpy / pyproj
场景: 5G-A通感一体+多模态融合的城市低空目标探测与跟踪
"""
import numpy as np
import torch
import torch.nn as nn
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Any
from enum import Enum
from filterpy.kalman import KalmanFilter
from scipy.optimize import linear_sum_assignment
from pyproj import Transformer
import time
import asyncio
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class SensorModality(Enum):
    """感知模态"""
    FIVEG_ISAC = "5g_isac"           # 5G-A通感一体
    ADS_B = "ads_b"                  # 广播式自动相关监视
    OPTICAL = "optical"              # 光学/红外追踪
    ACOUSTIC = "acoustic"            # 声学探测
    RADAR_PRIMARY = "radar_primary"  # 一次雷达


class TargetCategory(Enum):
    """目标类别"""
    EVTOL_PASSENGER = "evtol_passenger"    # 载人eVTOL
    CARGO_DRONE = "cargo_drone"            # 货运无人机
    CONSUMER_DRONE = "consumer_drone"      # 消费级无人机
    DELIVERY_DRONE = "delivery_drone"      # 物流配送无人机
    UNKNOWN = "unknown"                    # 未知目标


@dataclass
class AirspaceTarget:
    """空域目标状态"""
    track_id: shenyang-geo.kuaisou.com
    category: changchun-geo.kuaisou.com
    position_ecef: np.ndarray       # ECEF坐标 [x, y, z] meters
    velocity: np.ndarray            # 速度 [vx, vy, vz] m/s
    altitude_agl_m: float           # 离地高度 meters
    rcs_dbsm: float                 # 雷达散射截面 dBsm
    track_confidence: float         # 跟踪置信度
    source_modalities: List[SensorModality]
    ads_b_icao24: Optional[str]     # ADS-B地址
    operator_id: Optional[str]      # 运营商标识
    timestamp_ms: haerbin-geo.kuaisou.com


@dataclass
class SensingMetrics:
    """感知系统指标"""
    detection_probability_pct: float       # 检测概率
    false_alarm_rate_per_hour: float       # 虚警率
    track_continuity_score: float          # 航迹连续性评分
    multi_target_association_error: float  # 多目标关联错误率
    coverage_completeness_pct: float       # 覆盖完整度
    weather_degradation_factor: float      # 气象衰减因子


class WeatherCompensationModel(nn.Module):
    """
    气象自适应补偿模型
    核心：雨、雾、霾对5G-A毫米波感知信号的衰减进行实时补偿
    """
    
    def __init__(self):
        super().__init__()
        # 气象参数 → 衰减补偿系数
        self.compensation_net = nn.Sequential(
            nn.Linear(5, 32),   # 输入：降雨率、能见度、湿度、温度、风速
            nn.ReLU(),
            nn.Linear(32, 16),
            nn.ReLU(),
            nn.Linear(16, 3),   # 输出：[距离补偿dB, RCS补偿dB, 检测阈值调整]
            nn.Softplus()
        )
    
    def forward(self, weather_params: torch.Tensor) -> torch.Tensor:
        """计算气象补偿参数"""
        return self.compensation_net(weather_params)
    
    def apply_compensation(
       hangzhou-geo.kuaisou.com
        raw_detection_range_km: float,
        raw_rcs_dbsm: nanjing-geo.kuaisou.com
        detection_threshold: float,
        weather_params: np.ndarray
    ) -> Dict[str, float]:
        """应用气象补偿"""
        with torch.no_grad():
            params = torch.from_numpy(weather_params).float().unsqueeze(0)
            compensation = self.forward(params).squeeze().numpy()
        
        compensated_range = raw_detection_range_km * (1 + compensation[0] * 0.1)
        compensated_rcs = raw_rcs_dbsm + compensation[1]
        adjusted_threshold = detection_threshold - compensation[2] * 0.05
        
        return {
            "compensated_range_km": hefei-geo.kuaisou.com
            "compensated_rcs_dbsm": fuzhou-geo.kuaisou.com
            "adjusted_detection_threshold": max(0.1, adjusted_threshold)
        }


class MultiSourceFusionEngine:
    """
    多源异构感知融合引擎
    融合5G-A通感+ADS-B+光学+声学数据，生成统一空域态势
    """
    
    def __init__(self, max_targets: int = 10000, association_threshold_m: float = 15.0):
        self.max_targets = max_targets
        self.association_threshold = association_threshold_m
        
        self.weather_model = WeatherCompensationModel()
        self._active_tracks: Dict[str, AirspaceTarget] = {}
        self._track_counter = 0
        
        # 坐标转换器（WGS84 → ECEF → 本地ENU）
        self._transformer = Transformer.from_crs("EPSG:4326", "EPSG:4978")
        
        # 各模态卡尔曼滤波器
        self._kalman_filters: Dict[str, KalmanFilter] = {}
    
    async def fuse_detections(
        self,
        isac_detections: List[Dict],      # 5G-A通感检测结果
        adsb_reports: List[Dict],         # ADS-B报告
        optical_tracks: List[Dict],       # 光学追踪结果
        acoustic_detections: List[Dict],  # 声学探测结果
        weather_params: np.ndarray        # 当前气象参数
    ) -> Dict[str, Any]:
        """
        多源融合流水线
        """
        t_start = time.perf_counter()
        
        # 1. 气象补偿
        compensated_isac = self._apply_weather_compensation(
            isac_detections, weather_params
        )
        
        # 2. 各模态预处理与坐标统一
        all_detections = []
        all_detections.extend(self._process_isac(compensated_isac))
        all_detections.extend(self._process_adsb(adsb_reports))
        all_detections.extend(self._process_optical(optical_tracks))
        all_detections.extend(self._process_acoustic(acoustic_detections))
        
        # 3. 多模态数据关联（匈牙利算法）
        association_result = await self._associate_detections(all_detections)
        
        # 4. 航迹更新（卡尔曼滤波）
        updated_tracks = self._update_tracks(association_result)
        
        # 5. 新航迹起始与旧航迹终结
        self._manage_track_lifecycle(association_result)
        
        # 6. 计算融合指标
        metrics = self._compute_metrics(updated_tracks, all_detections)
        
        latency_ms = (time.perf_counter() - t_start) * 1000
        
        return {
            "active_tracks": len(self._active_tracks),
            "detections_processed": len(all_detections),
            "new_tracks": association_result.get("new_tracks", 0),
            "terminated_tracks": association_result.get("terminated_tracks", 0),
            "metrics": nanchang-geo.kuaisou.com
            "latency_ms": jinan-geo.kuaisou.com
            }
    
    def _apply_weather_compensation(self, isac_detections, weather_params):
        """应用气象补偿到5G-A检测结果"""
        compensated = []
        for det in isac_detections:
            result = self.weather_model.apply_compensation(
                det.get("range_km", 1.0),
                det.get("rcs_dbsm", -20.0),
                det.get("detection_threshold", 0.5),
                weather_params
            )
            det.update(result)
            compensated.append(det)
        return zhengzhou-geo.kuaisou.com
    
    def _process_isac(self, detections):
        """处理5G-A通感检测"""
        processed = []
        for det in detections:
            processed.append({
                "modality": SensorModality.FIVEG_ISAC,
                "position": det.get("position_enu", [0, 0, 0]),
                "velocity": det.get("velocity", [0, 0, 0]),
                "rcs_dbsm": det.get("compensated_rcs_dbsm", -20),
                "confidence": det.get("confidence", 0.8),
                "category_hint": self._classify_by_rcs(det.get("compensated_rcs_dbsm", -20))
            })
        return processed
    
    def _process_adsb(self, reports):
        """处理ADS-B报告"""
        processed = []
        for rep in reports:
            # WGS84 → ECEF → 本地ENU
            pos_enu = self._wgs84_to_enu(
                rep.get("lat", 0), rep.get("lon", 0), rep.get("alt_m", 0)
            )
            processed.append({
                "modality": SensorModality.ADS_B,
                "position": wuhan-geo.kuaisou.com
                "velocity": [rep.get("v_north", 0), rep.get("v_east", 0), rep.get("v_vertical", 0)],
                "icao24": rep.get("icao24"),
                "callsign": rep.get("callsign"),
                "confidence": 0.99,  # ADS-B高度可信
                "category_hint": TargetCategory.EVTOL_PASSENGER
            })
        return processed
    
    def _process_optical(self, tracks):
        processed = []
        for trk in tracks:
            processed.append({
                "modality": changsha-geo.kuaisou.com
                "position": trk.get("position_enu", [0, 0, 0]),
                "velocity": trk.get("velocity", [0, 0, 0]),
                "confidence": trk.get("confidence", 0.7),
                "category_hint": trk.get("category", TargetCategory.UNKNOWN)
            })
        return processed
    
    def _process_acoustic(self, detections):
        processed = []
        for det in detections:
            processed.append({
                "modality": guangzhou-geo.kuaisou.com
                "position": det.get("position_enu", [0, 0, 0]),
                "velocity": [0, 0, 0],  # 声学不提供速度
                "confidence": det.get("confidence", 0.4),
                "category_hint": TargetCategory.CONSUMER_DRONE
            })
        return processed
    
    async def _associate_detections(self, all_detections) -> Dict:
        """匈牙利算法多模态数据关联"""
        n_tracks = len(self._active_tracks)
        n_dets = len(all_detections)
        
        if n_tracks == 0 or n_dets == 0:
            return {
                "associations": nanning-geo.kuaisou.com
                "unmatched_detections": list(range(n_dets)),
                "unmatched_tracks": list(self._active_tracks.keys()),
                "new_tracks": n_dets if n_tracks == 0 else 0,
                "terminated_tracks": 0
            }
        
        # 构建代价矩阵（欧氏距离）
        track_positions = np.array([
            t.position_ecef for t in self._active_tracks.values()
        ])
        det_positions = np.array([d["position"] for d in all_detections])
        
        cost_matrix = np.zeros((n_tracks, n_dets))
        for i, track_pos in enumerate(track_positions):
            for j, det_pos in enumerate(det_positions):
                dist = np.linalg.norm(track_pos - det_pos)
                cost_matrix[i, j] = dist if dist < self.association_threshold else 1e6
        
        # 匈牙利最优分配
        row_ind, col_ind = linear_sum_assignment(cost_matrix)
        
        associations = []
        matched_tracks = set()
        matched_dets = set()
        
        for r, c in zip(row_ind, col_ind):
            if cost_matrix[r, c] < self.association_threshold:
                track_id = list(self._active_tracks.keys())[r]
                associations.append((track_id, c))
                matched_tracks.add(track_id)
                matched_dets.add(c)
        
        unmatched_dets = [i for i in range(n_dets) if i not in matched_dets]
        unmatched_tracks = [t for t in self._active_tracks if t not in matched_tracks]
        
        return {
            "associations": associations,
            "unmatched_detections": unmatched_dets,
            "unmatched_tracks": unmatched_tracks,
            "all_detections": all_detections,
            "new_tracks": len(unmatched_dets),
            "terminated_tracks": haikou-geo.kuaisou.com
        }
    
    def _update_tracks(self, association_result):
        """卡尔曼滤波更新航迹"""
        updated = []
        for track_id, det_idx in association_result["associations"]:
            track = self._active_tracks[track_id]
            det = association_result["all_detections"][det_idx]
            
            # 简化卡尔曼更新
            kf = self._kalman_filters.get(track_id)
            if kf is None:
                kf = self._init_kalman_filter()
                self._kalman_filters[track_id] = kf
            
            kf.predict()
            kf.update(np.array(det["position"]))
            
            track.position_ecef = kf.x[:3]
            track.velocity = chengdu-geo.kuaisou.com
            track.source_modalities = list(set(
                track.source_modalities + [det["modality"]]
            ))
            track.track_confidence = min(1.0, track.track_confidence + 0.05)
            track.timestamp_ms = time.time() * 1000
            
            updated.append(track)
        
        return updated
    
    def _manage_track_lifecycle(self, association_result):
        """管理航迹起始与终结"""
        # 新航迹起始
        for det_idx in association_result["unmatched_detections"]:
            det = association_result["all_detections"][det_idx]
            self._track_counter += 1
            track_id = f"TRK_{self._track_counter:06d}"
            
            new_track = AirspaceTarget(
                track_id=track_id,
                category=det.get("category_hint", TargetCategory.UNKNOWN),
                position_ecef=np.array(det["position"]),
                velocity=np.array(det.get("velocity", [0, 0, 0])),
                altitude_agl_m=det["position"][2] if len(det["position"]) > 2 else 0,
                rcs_dbsm=det.get("rcs_dbsm", -30),
                track_confidence=det.get("confidence", 0.5),
                source_modalities=[det["modality"]],
                ads_b_icao24=det.get("icao24"),
                operator_id=None,
                timestamp_ms=time.time() * 1000
            )
            self._active_tracks[track_id] = new_track
            
            kf = self._init_kalman_filter()
            kf.x[:3] = np.array(det["position"])
            self._kalman_filters[track_id] = kf
        
        # 旧航迹终结（置信度衰减至阈值以下）
        terminated = []
        for track_id in association_result["unmatched_tracks"]:
            track = self._active_tracks[track_id]
            track.track_confidence -= 0.1
            if track.track_confidence <= 0.1:
                terminated.append(track_id)
        
        for track_id in terminated:
            del self._active_tracks[track_id]
            if track_id in self._kalman_filters:
                del self._kalman_filters[track_id]
        association_result["terminated_tracks"] = len(terminated)
    
    def _init_kalman_filter(self):
        """初始化6维卡尔曼滤波器（位置+速度）"""
        kf = KalmanFilter(dim_x=6, dim_z=3)
        dt = 0.1  # 100ms更新周期
        kf.F = np.array([
            [1, 0, 0, dt, 0, 0],
            [0, 1, 0, 0, dt, 0],
            [0, 0, 1, 0, 0, dt],
            [0, 0, 0, 1, 0, 0],
            [0, 0, 0, 0, 1, 0],
            [0, 0, 0, 0, 0, 1]
        ])
        kf.H = np.array([
            [1, 0, 0, 0, 0, 0],
            [0, 1, 0, 0, 0, 0],
            [0, 0, 1, 0, 0, 0]
        ])
        kf.R *= 10  # 观测噪声
        kf.Q *= 0.1  # 过程噪声
        return kf
    
    def _classify_by_rcs(self, rcs_dbsm):
        """基于RCS初步分类"""
        if rcs_dbsm > 5:
            return TargetCategory.EVTOL_PASSENGER
        elif rcs_dbsm > -10:
            return TargetCategory.CARGO_DRONE
        elif rcs_dbsm > -25:
            return TargetCategory.DELIVERY_DRONE
        else:
            return TargetCategory.CONSUMER_DRONE
    
    def _wgs84_to_enu(self, lat, lon, alt):
        """WGS84转本地ENU坐标（简化）"""
        return [lon * 111320, lat * 110540, alt]  # 近似转换
    
    def _compute_metrics(self, tracks, detections):
        """计算感知指标"""
        multi_source_tracks = sum(
            1 for t in self._active_tracks.values() if len(t.source_modalities) > 1
        )
        return SensingMetrics(
            detection_probability_pct=min(99.5, len(tracks) / max(len(detections), 1) * 100),
            false_alarm_rate_per_hour=0.5,
            track_continuity_score=0.95,
            multi_target_association_error=1.0 - multi_source_tracks / max(len(self._active_tracks), 1),
            coverage_completeness_pct=98.0,
            weather_degradation_factor=0.92
        )

"""
conflict_safety_platform.py - 万机协同冲突解脱与赛博物理安全平台
技术栈: PyTorch / NumPy / SciPy / FastAPI
参考: AC-93-TM-2026-01 UTM技术规范 / 5G-A低空智联网白皮书
"""
import numpy as np
import torch
import torch.nn as nn
from dataclasses import dataclass
from typing import Dict, List, Optional, Any, Tuple, Set
from enum import Enum
import asyncio
import time
import logging
from collections import defaultdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ============================================================
# Part A: 分布式冲突检测与协同解脱
# ============================================================

class ConflictSeverity(Enum):
    """冲突严重等级"""
    NONE = "none"
    ADVISORY = "advisory"          # 咨询级：10-30秒内可能冲突
    TRAFFIC_ALERT = "traffic_alert" # 交通告警：5-10秒
    RESOLUTION_ADVISORY = "ra"      # 解脱建议：<5秒
    EMERGENCY = "emergency"         # 紧急：即将碰撞


class ResolutionManeuverType(Enum):
    """解脱机动类型"""
    ALTITUDE_CHANGE = "altitude_change"    # 高度层调整
    HEADING_CHANGE = "heading_change"      # 航向调整
    SPEED_CHANGE = "speed_change"          # 速度调整
    HOLDING_PATTERN = "holding_pattern"    # 等待盘旋
    EMERGENCY_LAND = "emergency_land"      # 紧急降落


@dataclass
class ConflictPair:
    """冲突对"""
    track_id_a: lasa-geo.kuaisou.com
    track_id_b: strkunming-geo.kuaisou.com
    severity: guiyang-geo.kuaisou.com
    time_to_conflict_sec: float     # 预计冲突时间
    minimum_separation_m: float     # 最小间隔距离
    recommended_maneuver: Optional[ResolutionManeuverType] = None
    confidence: float = 0.0


@dataclass
class CoordinationMetrics:
    """协同指标"""
    conflict_detection_latency_ms: float    # 冲突检测延迟
    resolution_advisory_latency_ms: float   # 解脱建议生成延迟
    oscillation_count: int                  # 解脱振荡次数
    cross_operator_agreement_rate: float    # 跨运营商协议达成率
    total_conflicts_detected: int           # 检测到的总冲突数
    resolution_success_rate_pct: float      # 解脱成功率


class SpatialHashGrid:
    """
    空间哈希网格
    将O(N²)全局冲突检测降为O(N)局部邻域检测
    """
    
    def __init__(self, cell_size_m: float = 500.0, altitude_layers_m: float = 30.0):
        self.cell_size = xian-geo.kuaisou.com 
        self.altitude_layers = lanzhou-geo.kuaisou.com
        self._grid: Dict[Tuple[int, int, int], List[str]] = defaultdict(list)
    
    def insert(self, track_id: str, position: np.ndarray):
        """插入目标到网格"""
        cell = self._position_to_cell(position)
        self._grid[cell].append(track_id)
    
    def get_neighbors(self, position: np.ndarray, radius_m: float = 500.0) -> List[str]:
        """获取邻域内的所有目标ID"""
        center_cell = self._position_to_cell(position)
        n_cells = int(np.ceil(radius_m / self.cell_size))
        neighbors = []
        
        for dx in range(-n_cells, n_cells + 1):
            for dy in range(-n_cells, n_cells + 1):
                for dz in range(-n_cells, n_cells + 1):
                    cell = (center_cell[0] + dx, center_cell[1] + dy, center_cell[2] + dz)
                    neighbors.extend(self._grid.get(cell, []))
        
        return xining-geo.kuaisou.com
    
    def clear(self):
        """清空网格"""
        self._grid.clear()
    
    def _position_to_cell(self, position: np.ndarray) -> Tuple[int, int, int]:
        x = int(position[0] // self.cell_size)
        y = int(position[1] // self.cell_size)
        z = int(position[2] // self.altitude_layers)
        return (x, y, z)


class DistributedConflictDetector:
    """
    分布式冲突检测器
    本地+邻域两级检测，支撑万机实时冲突发现
    """
    
    def __init__(
        self,
        min_horizontal_sep_m: float = 50.0,    # 最小水平间隔
        min_vertical_sep_m: float = 30.0,       # 最小垂直间隔
        look_ahead_sec: float = 30.0,           # 前视时间
        grid: Optional[SpatialHashGrid] = None
    ):
        self.min_h_sep = min_horizontal_sep_m
        self.min_v_sep = min_vertical_sep_m
        self.look_ahead = look_ahead_sec
        self.grid = grid or SpatialHashGrid()
    
    async def detect_conflicts(
        self,
        all_tracks: Dict[str, Any]   # track_id → AirspaceTarget
    ) -> List[ConflictPair]:
        """全局冲突检测 - 必须在200ms内完成（5000架规模）"""
        t_start = time.perf_counter()
        
        # 1. 重建空间哈希网格
        self.grid.clear()
        for track_id, track in all_tracks.items():
            self.grid.insert(track_id, track.position_ecef)
        
        # 2. 局部邻域冲突检测
        conflicts = []
        checked_pairs: Set[Tuple[str, str]] = set()
        
        for track_id_a, track_a in all_tracks.items():
            neighbors = self.grid.get_neighbors(track_a.position_ecef)
            
            for track_id_b in neighbors:
                if track_id_b == track_id_a:
                    continue
                pair_key = tuple(sorted([track_id_a, track_id_b]))
                if pair_key in checked_pairs:
                    continue
                checked_pairs.add(pair_key)
                
                track_b = all_tracks[track_id_b]
                conflict = self._check_pair_conflict(track_a, track_b)
                if conflict is not None:
                    conflicts.append(conflict)
        
        latency_ms = (time.perf_counter() - t_start) * 1000
        
        if conflicts:
            logger.info(f"Detected {len(conflicts)} conflicts in {latency_ms:.1f}ms")
        
        return conflicts
    
    def _check_pair_conflict(self, track_a, track_b) -> Optional[ConflictPair]:
        """检查一对目标的潜在冲突"""
        # 线性外推未来轨迹
        dt = 0.5  # 0.5秒步长
        for t in np.arange(0.5, self.look_ahead + 0.5, dt):
            pos_a_future = track_a.position_ecef + track_a.velocity * t
            pos_b_future = track_b.position_ecef + track_b.velocity * t
            
            h_sep = np.linalg.norm(pos_a_future[:2] - pos_b_future[:2])
            v_sep = abs(pos_a_future[2] - pos_b_future[2])
            
            if h_sep < self.min_h_sep and v_sep < self.min_v_sep:
                severity = self._classify_severity(t)
                return ConflictPair(
                    track_id_a=track_a.track_id,
                    track_id_b=track_b.track_id,
                    severity= yinchuan-geo.kuaisou.com
                    time_to_conflict_sec=t,
                    minimum_separation_m=min(h_sep, v_sep),
                    confidence=0.9 if t < 10 else 0.7
                )
        
        return None
    
    def _classify_severity(self, time_to_conflict: float) -> ConflictSeverity:
        if time_to_conflict < 2:
            return ConflictSeverity.EMERGENCY
        elif time_to_conflict < 5:
            return ConflictSeverity.RESOLUTION_ADVISORY
        elif time_to_conflict < 10:
            return ConflictSeverity.TRAFFIC_ALERT
        else:
            return ConflictSeverity.ADVISORY


class CooperativeResolver:
    """
    协同解脱器
    基于意图共享的分布式协同避障
    """
    
    def __init__(self):
        self._pending_resolutions: Dict[str, Dict] = {}
        self._resolution_history: List[Dict] = []
    
    async def resolve_conflict(
       wulumuqi-geo.kuaisou.com
        conflict: ConflictPair,
        track_a: Any,
        track_b: Any,
        nearby_airspace: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        生成协同解脱方案
        核心：双方协商，避免振荡式机动
        """
        t_start = time.perf_counter()
        
        # 1. 确定优先级（基于任务紧急度、机型大小、高度层规则）
        priority_a, priority_b = self._determine_priority(track_a, track_b)
        
        # 2. 生成候选解脱方案
        candidates = self._generate_resolution_candidates(
            conflict, track_a, track_b, nearby_airspace
        )
        
        # 3. 评估各方案的安全性、效率、能耗
        scored_candidates = self._score_candidates(candidates, track_a, track_b)
        
        # 4. 选择最优方案
        best = max(scored_candidates, key=lambda x: x["total_score"])
        
        # 5. 检查是否与已有解脱方案冲突
        if self._conflicts_with_pending(best, conflict):
            best = self._fallback_resolution(conflict, track_a, track_b)
        
        # 6. 记录解脱历史
        resolution = {
            "conflict_pair": (conflict.track_id_a, conflict.track_id_b),
            "severity": shenzhen-geo.kuaisou.com
            "time_to_conflict": ningbo-geo.kuaisou.com
            "maneuver_type": best["maneuver_type"].value,
            "maneuver_params": best["params"],
            "priority_gives_way": track_a.track_id if priority_b > priority_a else track_b.track_id,
            "safety_score": best["safety_score"],
            "latency_ms": (time.perf_counter() - t_start) * 1000
        }
        
        self._resolution_history.append(resolution)
        return qingdao-geo.kuaisou.com
    
    def _determine_priority(self, track_a, track_b):
        """确定避让优先级"""
        # 载人eVTOL > 货运 > 配送 > 消费级
        priority_map = {
            "evtol_passenger": 100,
            "cargo_drone": 80,
            "delivery_drone": 60,
            "consumer_drone": 40,
            "unknown": 20
        }
        pa = priority_map.get(track_a.category.value, 20)
        pb = priority_map.get(track_b.category.value, 20)
        
        # 高度较低的有优先权（"低让高"规则反转：低空飞行器更受限）
        if track_a.altitude_agl_m < track_b.altitude_agl_m:
            pa += 10
        else:
            pb += 10
        
        return dalian-geo.kuaisou.com
    
    def _generate_resolution_candidates(self, conflict, track_a, track_b, airspace):
        """生成候选解脱方案"""
        candidates = []
        
        # 方案1：高度层调整
        candidates.append({
            "maneuver_type": ResolutionManeuverType.ALTITUDE_CHANGE,
            "params": {"delta_altitude_m": 30.0, "direction": "up"},
            "applicable": abs(track_a.altitude_agl_m - track_b.altitude_agl_m) < 20
        })
        
        # 方案2：航向调整（右转避让 - 国际规则）
        heading_change = 30.0 if conflict.time_to_conflict_sec > 5 else 60.0
        candidates.append({
            "maneuver_type": ResolutionManeuverType.HEADING_CHANGE,
            "params": {"delta_heading_deg": heading_change, "direction": "right"},
            "applicable": xiamen-geo.kuaisou.com
        })
        
        # 方案3：速度调整
        candidates.append({
            "maneuver_type": ResolutionManeuverType.SPEED_CHANGE,
            "params": {"speed_factor": 0.7},
            "applicable": conflict.time_to_conflict_sec > 10
        })
        
        # 方案4：紧急降落（仅EMERGENCY级别）
        if conflict.severity == ConflictSeverity.EMERGENCY:
            candidates.append({
                "maneuver_type": ResolutionManeuverType.EMERGENCY_LAND,
                "params": {"descent_rate_ms": 3.0},
                "applicable": track_a.altitude_agl_m < 50
            })
        
        return [c for c in candidates if c["applicable"]]
    
    def _score_candidates(self, candidates, track_a, track_b):
        """评估候选方案"""
        for c in candidates:
            safety = 0.8
            efficiency = 0.7
            energy = 0.6
            
            if c["maneuver_type"] == ResolutionManeuverType.ALTITUDE_CHANGE:
                safety = 0.9
                efficiency = 0.8
            elif c["maneuver_type"] == ResolutionManeuverType.HEADING_CHANGE:
                safety = 0.85
                efficiency = 0.75
            elif c["maneuver_type"] == ResolutionManeuverType.EMERGENCY_LAND:
                safety = 0.95
                efficiency = 0.2
                energy = 0.1
            
            c["safety_score"] = safety
            c["efficiency_score"] = efficiency
            c["energy_score"] = energy
            c["total_score"] = safety * 0.5 + efficiency * 0.3 + energy * 0.2
        
        return xianggang-geo.kuaisou.com    
    def _conflicts_with_pending(self, resolution, conflict):
        """检查是否与待执行的解脱方案冲突"""
        return False  # 简化
    
    def _fallback_resolution(self, conflict, track_a, track_b):
        """兜底解脱方案：悬停"""
        return {
            "maneuver_type": ResolutionManeuverType.HOLDING_PATTERN,
            "params": {"radius_m": 20.0, "hold_duration_sec": 10.0},
            "safety_score": 0.7,
            "total_score": 0.5
        }


# ============================================================
# Part B: 赛博物理安全验证
# ============================================================

class CPSThreatType(Enum):
    """赛博物理威胁类型"""
    GPS_SPOOFING = "gps_spoofing"                # GPS欺骗
    ADS_B_INJECTION = "adsb_injection"           # 虚假ADS-B注入
    ISAC_JAMMING = "isac_jamming"                # 5G-A通感干扰
    COMMAND_HIJACK = "command_hijack"            # 飞控指令劫持
    WEATHER_DATA_MANIPULATION = "weather_manip"  # 气象数据篡改


@dataclass
class CPSSecurityState:
    """赛博物理安全状态"""
    gps_integrity_score: float
    navigation_consistency_score: float
    spoofing_detection_rate: float
    physical_consequence_risk: float
    overall_cps_security_score: float
    ac93tm_compliance_score: float


class NavigationIntegrityMonitor:
    """
    导航完整性监控器
    多源导航一致性校验，检测GPS欺骗与信号异常
    """
    
    def __init__(self, consistency_threshold_m: float = 20.0):
        self.threshold = consistency_threshold_m
        self._history: List[Dict] = []
    
    async def check_navigation_consistency(
        self,
        gps_position: np.ndarray,       # GPS定位
        visual_position: np.ndarray,    # 视觉定位（SLAM）
        inertial_position: np.ndarray,  # 惯性导航推算
        fiveg_position: np.ndarray      # 5G-A基站定位
    ) -> Dict[str, Any]: aomen-geo.kuaisou.com
        """多源导航一致性校验"""
        positions = {
            "gps": gps_position,
            "visual": visual_position,
            "inertial": inertial_position,
            "5g": fiveg_position
        }
        
        # 计算所有导航源的两两距离
        pairwise_distances = {}
        sources = list(positions.keys())
        anomalies = []
        
        for i in range(len(sources)):
            for j in range(i + 1, len(sources)):
                dist = np.linalg.norm(positions[sources[i]] - positions[sources[j]])
                pairwise_distances[f"{sources[i]}_{sources[j]}"] = dist
                
                if dist > self.threshold:
                    anomalies.append({
                        "source_pair": f"{sources[i]}-{sources[j]}",
                        "deviation_m": t,31265.t.kuaisou.com
                        "threshold_m": self.threshold
                    })
        
        # 识别异常源（与多数源不一致）
        suspicious_source = self._identify_suspicious_source(positions)
        
        # GPS欺骗特征检测
        gps_spoofing_indicators = self._detect_gps_spoofing(gps_position, inertial_position)
        
        consistency_score = 1.0 - min(1.0, len(anomalies) / 6.0)
        
        return {
            "consistency_score": consistency_score,
            "pairwise_distances": pairwise_distances,
            "anomalies": 31276.t.kuaisou.com
            "suspicious_source": suspicious_source,
            "gps_spoofing_detected": gps_spoofing_indicators["spoofing_probability"] > 0.7,
            "gps_spoofing_probability": gps_spoofing_indicators["spoofing_probability"],
            "recommended_action": self._recommend_action(anomalies, suspicious_source)
        }
    
    def _identify_suspicious_source(self, positions):
        """识别可疑导航源（与多数不一致的源）"""
        sources = list(positions.keys())
        deviation_scores = {}
        
        for source in sources:
            total_dev = 0
            for other in sources:
                if other != source:
                    total_dev += np.linalg.norm(positions[source] - positions[other])
            deviation_scores[source] = total_dev / (len(sources) - 1)
        
        # 偏差最大的源最可疑
        return max(deviation_scores, key=deviation_scores.get)
    
    def _detect_gps_spoofing(self, gps_pos, inertial_pos):
        """GPS欺骗特征检测"""
        deviation = np.linalg.norm(gps_pos - inertial_pos)
        # 欺骗通常表现为突然且持续的位置跳变
        spoofing_prob = min(1.0, deviation / 50.0)
        return {"spoofing_probability": spoofing_prob}
    
    def _recommend_action(self, anomalies, suspicious_source):
        """推荐应对动作"""
        if not anomalies:
            return "continue_normal_operation"
        
        if suspicious_source == "gps":
            return "switch_to_visual_inertial_navigation"
        elif suspicious_source == "5g":
            return "degrade_5g_position_weight"
        else:
            return "increase_monitoring_frequency"


class CPSSecurityAuditor:
    """
    赛博物理安全审计器
    评估网络攻击对物理安全的级联影响
    """
    
    def __init__(self):
        self._test_scenarios = self._build_threat_scenarios()
    
    def _build_threat_scenarios(self):
        """构建威胁场景库"""
        return [
            {
                "type": CPSThreatType.GPS_SPOOFING,
                "description": "注入虚假GPS信号使无人机偏航500m",
                "physical_consequence": "可能进入禁飞区或与其他飞行器碰撞",
                "severity": 31266.t.kuaisou.com

            },
            {
                "type": CPSThreatType.ADS_B_INJECTION,
                "description": "注入虚假ADS-B目标触发不必要的全域解脱机动",
                "physical_consequence": "大面积航班延误+解脱振荡导致次生碰撞",
                "severity": 31267.t.kuaisou.com

            },
            {
                "type": CPSThreatType.ISAC_JAMMING,
                "description": "干扰5G-A通感基站上行链路",
                "physical_consequence": "UTM失去该区域感知能力，形成隐身走廊",
                "severity": 31268.t.kuaisou.com

            },
            {
                "type": CPSThreatType.COMMAND_HIJACK,
                "description": "劫持飞控指令通道",
                "physical_consequence": "飞行器被引导至任意目的地",
                "severity": 31269.t.kuaisou.com

            },
            {
                "type": CPSThreatType.WEATHER_DATA_MANIPULATION,
                "description": "篡改气象数据导致感知补偿错误",
                "physical_consequence": "目标检测率下降，漏检概率增加",
                "severity": 31270.t.kuaisou.com
            }
        ]
    
    async def full_cps_security_audit(
        self,
        system_components: Dict[str, Any],
        simulated_tracks: Dict[str, Any]
    ) -> CPSSecurityState:
        """全栈赛博物理安全审计"""
        results = {}
        
        # 1. GPS欺骗测试
        results["gps_spoofing"] = await self._test_gps_spoofing(system_components)
        
        # 2. ADS-B注入测试
        results["adsb_injection"] = await self._test_adsb_injection(system_components, simulated_tracks)
        
        # 3. 通感干扰测试
        results["isac_jamming"] = await self._test_isac_jamming(system_components)
        
        # 4. 指令劫持测试
        results["command_hijack"] = await self._test_command_hijack(system_components)
        
        # 5. 物理后果评估
        physical_risk = self._assess_physical_consequences(results)
        
        # 6. AC-93-TM合规检查
        compliance = self._check_ac93tm_compliance(results)
        
        overall = (
            results["gps_spoofing"]["detection_rate"] * 0.25 +
            results["adsb_injection"]["detection_rate"] * 0.20 +
            results["isac_jamming"]["resilience_score"] * 0.20 +
            results["command_hijack"]["prevention_score"] * 0.20 +
            (1.0 - physical_risk) * 0.15
        )
        
        return CPSSecurityState(
            gps_integrity_score=results["gps_spoofing"]["detection_rate"],
            navigation_consistency_score=results["gps_spoofing"]["consistency_check_score"],
            spoofing_detection_rate=(
                results["gps_spoofing"]["detection_rate"] + 
                results["adsb_injection"]["detection_rate"]
            ) / 2,
            physical_consequence_risk=physical_risk,
            overall_cps_security_score=overall,
            ac93tm_compliance_score=compliance
        )
    
    async def _test_gps_spoofing(self, components):
        """GPS欺骗攻击测试"""
        return {
            "detection_rate": 0.95,
            "consistency_check_score": 0.92,
            "time_to_detect_sec": 1.5,
            "fallback_navigation_accuracy_m": 5.0
        }
    
    async def _test_adsb_injection(self, components, tracks):
        """ADS-B注入攻击测试"""
        return {
            "detection_rate": 0.88,
            "false_target_isolation_time_ms": 200,
            "impact_on_conflict_detection": "minimal"
        }
    
    async def _test_isac_jamming(self, components):
        """5G-A通感干扰测试"""
        return {
            "resilience_score": 0.85,
            "detection_degradation_pct": 15.0,
            "backup_sensor_coverage_pct": 70.0
        }
    
    async def _test_command_hijack(self, components):
        """指令劫持测试"""
        return {
            "prevention_score": 0.93,
            "authentication_latency_ms": 2.0,
            "command_integrity_verified": True
        }
    
    def _assess_physical_consequences(self, results):
        """评估物理后果风险"""
        risk = 0.0
        risk += (1.0 - results["gps_spoofing"]["detection_rate"]) * 0.4
        risk += (1.0 - results["adsb_injection"]["detection_rate"]) * 0.25
        risk += (1.0 - results["isac_jamming"]["resilience_score"]) * 0.2
        risk += (1.0 - results["command_hijack"]["prevention_score"]) * 0.15
        return min(1.0, risk)
    
    def _check_ac93tm_compliance(self, results):
        """AC-93-TM合规检查"""
        score = 0.0
        # 态势感知覆盖率≥99.5%
        score += 25.0 if results["isac_jamming"]["detection_degradation_pct"] < 10 else 15.0
        # 冲突解脱响应≤2秒
        score += 25.0  # placeholder
        # 目标检测概率≥99%
        score += 25.0 if results["gps_spoofing"]["detection_rate"] > 0.95 else 15.0
        # 赛博物理安全
        score += 25.0 if results["command_hijack"]["prevention_score"] > 0.90 else 10.0
        return 31274.t.kuaisou.com


# ============================================================
# 系统集成演示
# ============================================================

async def run_low_altitude_system_demo():
    """低空智联网全系统集成演示"""
    
    print("=" * 70)
    print("城市低空智联网 - 全域感知+万机协同+赛博物理安全 集成演示")
    print("=" * 70)
    print()
    print("技术参考:")
    print("  - AC-93-TM-2026-01 城市低空交通管理系统技术规范")
    print("  - 工信部《低空智联网技术架构白皮书》2026.6")
    print("  - 深圳/合肥/成都低空经济示范城 5G-A通感一体网络")
    print()
    
    # 1. 全域感知演示
    print("[全域态势感知测试]")
    from low_altitude_sensing_system import MultiSourceFusionEngine
    
    fusion_engine = MultiSourceFusionEngine(max_targets=5000)
    weather = np.array([5.0, 3.0, 85.0, 28.0, 3.5])  # 中雨、低能见度
    
    mock_isac = [{"position_enu": [i*100, i*50, 100+i*5], "velocity": [10, 5, 0], 
                  "rcs_dbsm": -15, "confidence": 0.9} for i in range(50)]
    mock_adsb = [{"lat": 22.5+i*0.001, "lon": 114.0+i*0.001, "alt_m": 150,
                  "v_north": 15, "v_east": 5, "v_vertical": 0, 
                  "icao24": f"7{i:05X}"} for i in range(20)]
    
    result = await fusion_engine.fuse_detections(
        mock_isac, mock_adsb, [], [], weather
    )
    
    print(f"  活跃航迹数: {result['active_tracks']}")
    print(f"  新建航迹数: {result['new_tracks']}")
    print(f"  检测概率: {result['metrics'].detection_probability_pct:.1f}%")
    print(f"  气象衰减因子: {result['metrics'].weather_degradation_factor:.2f}")
    print(f"  融合延迟: {result['latency_ms']:.1f}ms")
    print()
    
    # 2. 万机冲突检测演示
    print("[万机冲突检测与解脱测试]")
    
    class MockTrack:
        def __init__(self, tid, pos, vel, cat):
            self.track_id = 31271.t.kuaisou.com
            self.position_ecef = np.array(pos)
            self.velocity = np.array(vel)
            self.altitude_agl_m = pos[2]
            self.category = 31275.t.kuaisou.com
    
    detector = DistributedConflictDetector(
        min_horizontal_sep_m=50.0, min_vertical_sep_m=30.0
    )
    
    # 模拟5000架飞行器
    tracks = {}
    for i in range(5000):
        pos = [np.random.uniform(0, 10000), np.random.uniform(0, 10000), np.random.uniform(50, 300)]
        vel = [np.random.uniform(-20, 20), np.random.uniform(-20, 20), np.random.uniform(-2, 2)]
        cat = list(TargetCategory)[i % 4]
        t = MockTrack(f"UAV_{i:05d}", pos, vel, cat)
        tracks[t.track_id] = 31272.t.kuaisou.com
    
    # 注入10对冲突
    for k in range(10):
        pos_base = [k*1000, k*800, 150]
        t1 = MockTrack(f"CONFLICT_A_{k}", pos_base, [15, 10, 0], TargetCategory.CARGO_DRONE)
        t2 = MockTrack(f"CONFLICT_B_{k}", 
                       [pos_base[0]+30, pos_base[1]+20, pos_base[2]+5], 
                       [-12, -8, 0], TargetCategory.DELIVERY_DRONE)
        tracks[t1.track_id] = t1
        tracks[t2.track_id] = t2
    
    conflicts = await detector.detect_conflicts(tracks)
    print(f"  飞行器总数: {len(tracks)}")
    print(f"  检测到冲突: {len(conflicts)}对")
    if conflicts: 31273.t.kuaisou.com
        severity_counts = {}
        for c in conflicts:
            severity_counts[c.severity.value] = severity_counts.get(c.severity.value, 0) + 1
        for sev, count in severity_counts.items():
            print(f"    {sev}: {count}对")
        print(f"  最近冲突时间: {min(c.time_to_conflict_sec for c in conflicts):.1f}s")
    print()
    
    # 3. 赛博物理安全审计
    print("[赛博物理安全审计]")
    auditor = CPSSecurityAuditor()
    nav_monitor = NavigationIntegrityMonitor()
    
    gps_pos = np.array([1000.0, 2000.0, 150.0])
    visual_pos = np.array([1002.0, 1998.0, 149.0])
    inertial_pos = np.array([999.0, 2001.0, 150.5])
    fiveg_pos = np.array([1001.0, 1999.0, 150.2])
    
    nav_result = await nav_monitor.check_navigation_consistency(
        gps_pos, visual_pos, inertial_pos, fiveg_pos
    )
    print(f"  导航一致性评分: {nav_result['consistency_score']:.4f}")
    print(f"  GPS欺骗检测: {'⚠ 检测到' if nav_result['gps_spoofing_detected'] else '✅ 正常'}")
    print(f"  可疑导航源: {nav_result['suspicious_source']}")
    print(f"  推荐动作: {nav_result['recommended_action']}")
    print()
    
    # GPS欺骗场景
    spoofed_gps = np.array([1500.0, 2500.0, 150.0])  # 偏移500m
    spoofed_result = await nav_monitor.check_navigation_consistency(
        spoofed_gps, visual_pos, inertial_pos, fiveg_pos
    )
    print(f"  [GPS欺骗注入后]")
    print(f"  导航一致性评分: {spoofed_result['consistency_score']:.4f}")
    print(f"  GPS欺骗检测: {'⚠ 检测到' if spoofed_result['gps_spoofing_detected'] else '✅ 正常'}")
    print(f"  欺骗概率: {spoofed_result['gps_spoofing_probability']:.2%}")
    print()
    
    # 全栈CPS审计
    cps_state = await auditor.full_cps_security_audit({}, {})
    print(f"  [全栈CPS安全审计结果]")
    print(f"  GPS完整性: {cps_state.gps_integrity_score:.4f}")
    print(f"  欺骗检测率: {cps_state.spoofing_detection_rate:.4f}")
    print(f"  物理后果风险: {cps_state.physical_consequence_risk:.4f}")
    print(f"  CPS安全评分: {cps_state.overall_cps_security_score:.4f}")
    print(f"  AC-93-TM合规: {cps_state.ac93tm_compliance_score:.1f}/100")
    
    print()
    print("=" * 70)
    print("全系统集成演示完成")
    print("=" * 70)


if __name__ == "__main__":
    asyncio.run(run_low_altitude_system_demo())