法律与合规声明

• 本指南仅用于防御性安全教育与合法合规的数字取证研究，不得用于任何违法行为。
• 进行无人机（UAV/Drone）取证前，应完成授权与告知，明确管辖权与保全范围，并遵循当地法律法规（如中国民航相关规定、FAA/EASA/U-Space要求、隐私与数据保护法）。
• 全程执行证据保全链（Chain of Custody）：记录“谁、何时、何地、做了什么”，并对所有原始介质与导出文件进行哈希校验（MD5/SHA-256）。

1. 取证准备与数据源清单

• 介质类型：
  • 机载存储（飞控/黑匣子日志、传感器原始数据）、SD卡（照片/视频/飞行缓存）、电池智能芯片信息。
  • 地面端设备（遥控器、移动端APP：DJI GO/DJI Fly/Autel Explorer/QGroundControl/Mission Planner、云端同步 AirData 等）。
• 常见日志/文件格式：
  • DJI：飞控 .DAT（二进制）、移动端 .TXT（需转换）、照片/视频 EXIF/XMP 元数据。
  • ArduPilot/PX4：.BIN/.LOG（数据闪存日志）、参数 .param、任务 .waypoints。
  • Parrot/Autel/Skydio：JSON/CSV 等厂商格式。
• 保全与导出建议：
  • 断电并防篡改，优先镜像抽取；手机端使用ADB/iTunes备份只读模式。
  • 对原始介质与导出文件计算哈希并存档，使用写阻器（如有条件）。

示例：Windows PowerShell 快速哈希与只读复制

# 计算目录内所有文件SHA256
Get-ChildItem -Recurse | Get-FileHash -Algorithm SHA256 | Format-Table Path, Hash

# 复制到证据盘（建议证据盘只读挂载）
robocopy C:\DroneMedia E:\Evidence\DroneMedia /E /COPY:DAT /DCOPY:T /R:1 /W:1

2. 日志格式与转换工具速查

• DJI Log Viewer / Flight Reader / DatCon：转换 DJI .DAT/.TXT 为 CSV/JSON。
• Mission Planner / APM Planner / PX4 Tools：解析 ArduPilot/PX4 .BIN/.LOG。
• AirData、PhantomHelp：云端日志聚合与可视化。
• ExifTool：批量提取图像/视频 EXIF/XMP/厂商私有标签。
• QGIS/ArcGIS：地理空间分析、叠加管控区（NFZ）与地理围栏。

3. 日志解析与事件重建（扩展代码）

下面在现有 FlightDataAnalyzer 基础上扩展事件检测、KML 导出、地理围栏违规检测与电池健康分析。代码对字段名做了“有则用、无则跳过”的鲁棒性处理。

import math
from pathlib import Path

class FlightEventReconstructor(FlightDataAnalyzer):
    def point_in_polygon(self, x, y, polygon):
        """射线法判断点是否在多边形内。polygon为[(x1,y1),(x2,y2),...]"""
        inside = False
        n = len(polygon)
        for i in range(n):
            x1, y1 = polygon[i]
            x2, y2 = polygon[(i+1) % n]
            # 检查与边的交点
            if ((y1 > y) != (y2 > y)):
                xinters = (x2 - x1) * (y - y1) / (y2 - y1 + 1e-12) + x1
                if xinters > x:
                    inside = not inside
        return inside

    def detect_events(self, alt_threshold=2.0, speed_threshold=0.5):
        """检测起飞/着陆/RTH/模式变化/电池告警等事件"""
        events = []
        df = self.df

        # 起飞：高度从近0到>阈值或总速度突增
        for i in range(1, len(df)):
            alt_prev = df.iloc[i-1]['altitude'] if 'altitude' in df.columns else 0
            alt_curr = df.iloc[i]['altitude'] if 'altitude' in df.columns else 0
            v_prev = df.iloc[i-1]['total_speed'] if 'total_speed' in df.columns else 0
            v_curr = df.iloc[i]['total_speed'] if 'total_speed' in df.columns else 0
            if alt_prev <= alt_threshold and alt_curr > alt_threshold and v_curr - v_prev > speed_threshold:
                events.append({
                    'type': 'takeoff',
                    'time': df.iloc[i]['timestamp'],
                    'lat': df.iloc[i]['latitude'] if 'latitude' in df.columns else None,
                    'lon': df.iloc[i]['longitude'] if 'longitude' in df.columns else None
                })
                break

        # 着陆：高度降至阈值以下且速度接近0
        for i in range(len(df)-2, -1, -1):
            alt_curr = df.iloc[i]['altitude'] if 'altitude' in df.columns else 0
            v_curr = df.iloc[i]['total_speed'] if 'total_speed' in df.columns else 0
            if alt_curr <= alt_threshold and v_curr <= speed_threshold:
                events.append({
                    'type': 'landing',
                    'time': df.iloc[i]['timestamp'],
                    'lat': df.iloc[i]['latitude'] if 'latitude' in df.columns else None,
                    'lon': df.iloc[i]['longitude'] if 'longitude' in df.columns else None
                })
                break

        # 模式变化 / RTH：依赖 flight_mode 字段或推断（水平速度急剧变化+返航方向）
        if 'flight_mode' in df.columns:
            prev_mode = df.iloc[0]['flight_mode']
            for i in range(1, len(df)):
                mode = df.iloc[i]['flight_mode']
                if mode != prev_mode:
                    events.append({
                        'type': 'mode_change', 'from': prev_mode, 'to': mode,
                        'time': df.iloc[i]['timestamp'],
                        'lat': df.iloc[i]['latitude'] if 'latitude' in df.columns else None,
                        'lon': df.iloc[i]['longitude'] if 'longitude' in df.columns else None
                    })
                    # 识别返航模式
                    if str(mode).upper() in ['RTH', 'GO_HOME', 'RETURN_TO_HOME']:
                        events.append({
                            'type': 'RTH',
                            'time': df.iloc[i]['timestamp'],
                            'lat': df.iloc[i]['latitude'] if 'latitude' in df.columns else None,
                            'lon': df.iloc[i]['longitude'] if 'longitude' in df.columns else None
                        })
                    prev_mode = mode

        # 电池告警：battery_level 低于阈值或 voltage 值突降
        if 'battery_level' in df.columns:
            low_batt = df[df['battery_level'] <= 20]
            for _, row in low_batt.iterrows():
                events.append({
                    'type': 'battery_low', 'level': row['battery_level'],
                    'time': row['timestamp'], 'lat': row.get('latitude'), 'lon': row.get('longitude')
                })
        if 'voltage' in df.columns:
            # 简单检测电压突降
            dv = df['voltage'].diff()
            drops = df[dv < -0.5]
            for _, row in drops.iterrows():
                events.append({
                    'type': 'voltage_drop', 'voltage': row['voltage'],
                    'time': row['timestamp'], 'lat': row.get('latitude'), 'lon': row.get('longitude')
                })
        return events

    def export_kml(self, output_file='flight.kml'):
        """导出KML以在Google Earth中回放轨迹"""
        coords = []
        if {'longitude','latitude','altitude'} <= set(self.df.columns):
            for _, r in self.df.iterrows():
                coords.append(f"{r['longitude']},{r['latitude']},{r['altitude']}")
        kml = f"""
<?xml version="1.0" encoding="UTF-8"?>
<kml xmlns="http://www.opengis.net/kml/2.2">
  <Document>
    <name>Drone Flight Path</name>
    <Placemark>
      <name>Path</name>
      <LineString>
        <coordinates>
          {' '.join(coords)}
        </coordinates>
      </LineString>
    </Placemark>
  </Document>
</kml>
"""
        Path(output_file).write_text(kml, encoding='utf-8')
        print(f"KML exported to {output_file}")

    def check_nfz_violations(self, nfz_polygons):
        """检查进入地理围栏/禁飞区事件。nfz_polygons: List[List[(lon,lat),...]]"""
        violations = []
        if {'longitude','latitude'} <= set(self.df.columns):
            for _, r in self.df.iterrows():
                x, y = r['longitude'], r['latitude']
                for idx, poly in enumerate(nfz_polygons):
                    # 注意: 多边形点使用(lon,lat)
                    if self.point_in_polygon(x, y, poly):
                        violations.append({
                            'type': 'nfz_violation', 'nfz_id': idx,
                            'time': r['timestamp'], 'lat': y, 'lon': x
                        })
                        break
        return violations

    def battery_health_analysis(self):
        """估算电池健康: 电压曲线/容量衰减/温度影响（字段存在时分析）"""
        result = {}
        if 'voltage' in self.df.columns:
            result['voltage_mean'] = float(self.df['voltage'].mean())
            result['voltage_min'] = float(self.df['voltage'].min())
        if 'battery_level' in self.df.columns:
            result['level_min'] = float(self.df['battery_level'].min())
        if 'temperature' in self.df.columns:
            result['temp_max'] = float(self.df['temperature'].max())
        return result

# 使用示例（扩展）
def advanced_reconstruction_demo():
    recon = FlightEventReconstructor('parsed_flight_data.csv')
    recon.preprocess_data()
    events = recon.detect_events()
    print(f"Detected events: {len(events)}")
    for e in events[:10]:
        print(e)

    # 导出KML
    recon.export_kml('flight_path.kml')

    # 示例NFZ多边形（经纬度点）
    nfz = [[(116.39,39.90),(116.41,39.90),(116.41,39.92),(116.39,39.92)]]
    violations = recon.check_nfz_violations(nfz)
    print(f"NFZ violations: {len(violations)}")

    batt = recon.battery_health_analysis()
    print("Battery health:", batt)

4. 图像/视频EXIF元数据取证

• 目标：定位拍摄时间线、拍摄设备（厂商/型号）、GPS坐标、相机朝向、云台角度、序列号等。
• 建议：保留原始文件结构，使用 ExifTool/Pillow 读取；注意 DJI/厂商私有标签可能存储在 XMP。

from PIL import Image, ExifTags
import json

def extract_exif(path_glob='Media/*.JPG', output_json='exif_evidence.json'):
    results = []
    for p in Path('.').glob(path_glob):
        try:
            img = Image.open(p)
            exif = img._getexif() or {}
            mapped = {}
            for k, v in exif.items():
                tag = ExifTags.TAGS.get(k, str(k))
                mapped[tag] = v
            # 常见字段提取
            gps = mapped.get('GPSInfo', {})
            dt = mapped.get('DateTimeOriginal') or mapped.get('DateTime')
            make = mapped.get('Make'); model = mapped.get('Model')
            results.append({
                'file': str(p), 'datetime': dt, 'make': make, 'model': model,
                'gps': str(gps)
            })
        except Exception as e:
            results.append({'file': str(p), 'error': str(e)})
    Path(output_json).write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding='utf-8')
    print(f"EXIF evidence exported to {output_json}")

5. 行为重建与时间线整合

• 将飞行日志事件、EXIF拍摄时间、遥控器/APP操作记录、云端同步数据整合为统一时间线。
• 标准化时间统一到UTC，记录时区偏移与设备时钟漂移校正。

示例时间线结构（JSON）：

{
  "timeline": [
    {"t": "2024-07-20T08:21:03Z", "type": "takeoff", "lat": 39.90, "lon": 116.40},
    {"t": "2024-07-20T08:21:35Z", "type": "photo", "file": "DJI_0001.JPG"},
    {"t": "2024-07-20T08:25:00Z", "type": "RTH"},
    {"t": "2024-07-20T08:26:12Z", "type": "landing"}
  ]
}

6. 风险与合规检查清单

• 高度/速度/距离限制是否违规；是否进入NFZ/临近机场/军区等敏感区域。
• 是否启用远程ID（如适用）；飞行目的是否合规；是否有超视距（BVLOS）迹象。
• 隐私与个人数据处理合规（影像采集、人员信息、车牌面部等）。

7. 取证报告模板（精简版）

• 1. 案件背景与授权范围
• 2. 取证方法与保全链
• 3. 数据源与转换工具
• 4. 飞行日志分析（轨迹、事件、告警、NFZ）
• 5. 影像EXIF/视频元数据分析
• 6. 行为重建时间线
• 7. 风险评估与合规结论
• 8. 证据清单与哈希校验
• 9. 建议与改进措施

8. 工具与资源速查

• 数据转换：DatCon / Flight Reader / DJI Log Viewer / ExifTool
• 飞行日志解析：Mission Planner / QGroundControl / PX4 Tools
• 可视化与GIS：Google Earth / QGIS / ArcGIS / Folium
• 云端聚合：AirData / PhantomHelp（注意隐私与授权）

9. 参考文献与资源

• NIST Digital Evidence Collection Guidelines（数字证据采集指南）
• ArduPilot/PX4 官方日志与参数文档
• DJI 开发者与日志社区资料（DatCon/Flight Reader 文档）
• FAA/EASA/民航局无人机合规与远程ID相关文件

10. 结论

• 本补充内容完善了无人机取证实战从“数据保全—格式转换—日志解析—事件重建—可视化—合规评估—报告输出”的全流程。
• 附带的扩展代码可在保持字段兼容性的前提下，快速检测关键飞行事件、导出KML、识别禁飞区违规并进行电池健康评估，便于形成可审计的调查结论。