回放飞行时的数据记录

能办。核心思路：DJI 原始飞控 bin 没有公开字段规范，不建议直接裸解；先把 bin 用 DJI Assistant 2 for Matrice 导出为 CSV，或 DatCon 转 CSV，然后用下面的 Python 脚本按告警时间自动截取，看告警窗口里的电压、电量、链路丢包率，直接生成异常聚焦简报。

使用步骤

# 1. 安装依赖
pip install pandas matplotlib

# 2. 运行分析
python m300_alert_report.py your_log.csv

# 3. 可选：自定义窗口和阈值
python m300_alert_report.py your_log.csv \
  --window 15 \
  --low-voltage 46.0 \
  --low-battery 20 \
  --high-loss 5

脚本会生成：

• your_log_alert_report.md：异常聚焦简报
• your_log_alert_timeline.png：告警时间线图

Python 脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
M300 RTK 电力巡检飞行日志异常聚焦分析

用法:
    python m300_alert_report.py your_log.csv
    python m300_alert_report.py your_log.csv --window 15 --low-voltage 46.0 --low-battery 20 --high-loss 5

说明:
    - 原始飞控 bin 请先用 DJI Assistant 2 for Matrice 导出为 CSV，或 DatCon 转 CSV 后再喂给本脚本。
    - 脚本自动识别时间、电压、电量、链路丢包率/质量、告警码列。
    - 生成 <输入文件名>_alert_report.md，并默认生成 <输入文件名>_alert_timeline.png。
"""

import argparse
import os
import re
import sys
from datetime import timedelta

import pandas as pd

ALERT_COL_PATTERN = r'warn|error|alert|fault|code|notice'
NORMAL_STRINGS = {'', '0', '0.0', 'none', 'normal', 'nan', 'ok', 'n/a', '-'}

def find_col_priority(df, keywords, exclude=None):
    """按关键词优先级从左到右找列，排除包含 exclude 关键词的列。"""
    exclude = exclude or []
    for keyword in keywords:
        for col in df.columns:
            col_l = str(col).lower()
            if keyword in col_l and not any(e in col_l for e in exclude):
                return col
    return None

def get_alert_columns(df):
    return [c for c in df.columns if re.search(ALERT_COL_PATTERN, str(c), re.I)]

def to_numeric(s):
    return pd.to_numeric(s, errors='coerce')

def extract_row_alert_codes(df, idx, alert_cols):
    """从单行提取告警字段，格式为 列名=值。"""
    codes = []
    for c in alert_cols:
        try:
            val = df.at[idx, c]
        except Exception:
            continue
        if pd.isna(val):
            continue
        if str(val).strip().lower() in NORMAL_STRINGS:
            continue
        codes.append(f"{c}={val}")
    return codes

def detect_alert_rows(df, alert_cols):
    """返回告警行索引列表。"""
    mask = pd.Series(False, index=df.index)
    for c in alert_cols:
        s = df[c]
        s_num = to_numeric(s)
        if s_num.notna().sum() > 0.7 * len(s):
            # 数字列：非 0 认为是告警/异常码
            mask |= s_num.fillna(0) != 0
        else:
            ss = s.astype(str).str.lower()
            col_mask = ss.str.contains(r'warn|error|fault|alert|notice', regex=True, na=False)
            col_mask &= ~ss.str.strip().isin(NORMAL_STRINGS)
            mask |= col_mask
    return df.index[mask]

def merge_intervals(intervals):
    intervals = sorted(intervals, key=lambda x: x[0])
    merged = []
    for a, b in intervals:
        if not merged or a > merged[-1][1]:
            merged.append([a, b])
        else:
            merged[-1][1] = max(merged[-1][1], b)
    return merged

def parse_args():
    p = argparse.ArgumentParser(description='M300 RTK 飞行告警聚焦分析')
    p.add_argument('input', help='CSV/TXT 文件，建议从 DJI Assistant 2 或 DatCon 导出')
    p.add_argument('--window', type=int, default=10, help='告警前后截取窗口秒数，默认 10')
    p.add_argument('--low-voltage', type=float, default=46.0,
                   help='电压低阈值(V)，默认 46.0，按实际电池规格调整')
    p.add_argument('--low-battery', type=float, default=20.0,
                   help='电量低阈值(%%)，默认 20')
    p.add_argument('--high-loss', type=float, default=5.0,
                   help='链路丢包率高