示例成品 · 平台演示，按左边这组点选真跑出来的
能办。核心思路：DJI 原始飞控 bin 没有公开字段规范，不建议直接裸解；先把 bin 用 **DJI Assistant 2 for Matrice** 导出为 CSV，或 **DatCon** 转 CSV，然后用下面的 Python 脚本按告警时间自动截取，看告警窗口里的电压、电量、链路丢包率，直接生成异常聚焦简报。
## 使用步骤
```bash
# 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 脚本
```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='链路丢包率高
点左边「开工 · 直接出成品」，出一份你自己的版本（文字免费）