2. 无人艇航行数据异常值检测与清洗

任务对象与判定口径

无人艇航行日志通常由飞控/导航计算机按固定频率记录，核心字段包括 timestamp、lat、lon、sog（对地速度，m/s）、cog（对地航向，°）、heading（艏向，°）、roll/pitch/yaw（IMU 姿态，°）以及 depth。本案例要处理三类质量问题：跳变（单点或短时突变，超出物理可达范围）、缺失（时间戳断档或字段为空）、重复（同一时间戳多条记录或整行重复）。

清洗前必须先定口径，否则“异常”无法判定。本案例采用以下可复现规则（示例阈值，可按艇型调整）：

• 重复：timestamp 完全相同视为重复，保留首条；若同时间戳但字段不同，保留字段完整度更高的一条。
• 缺失：相邻记录时间差 > 2.5 倍标称采样周期（标称 1 Hz，即 > 2.5 s）判为断档；关键字段为空也计入缺失。
• 跳变：位置用相邻点推算速度 v = Δs/Δt，超过 12 m/s 判为位置跳变；航向类字段用环形差值 Δθ = ((θ₂-θ₁+180) mod 360) - 180，|Δθ| > 45°/s 判为航向跳变；sog 超过 15 m/s 或为负判为速度异常。

清洗流程与数据流

下图给出从原始日志到干净数据集的处理链，以及每一步的判定分支和输出统计口径。跳变点不直接删除，而是先标记为缺失，再与真实断档一起进入插值环节，这样能保持时间轴连续、便于后续重采样。

可运行的清洗实现

下面用 Python + pandas 完成全流程。示例数据由脚本内构造，包含人为注入的重复、断档和跳变，便于验证每一步是否生效；换成真实日志时只需替换 load_log() 的读取逻辑。

import numpy as np
import pandas as pd

R_EARTH = 6371000.0
NOMINAL_DT = 1.0          # 标称采样周期 1 s
GAP_FACTOR = 2.5          # 断档判定倍数
V_MAX = 12.0              # 位置跳变速度阈值 m/s
YAW_RATE_MAX = 45.0       # 航向跳变角速度阈值 °/s
SOG_MAX = 15.0            # 速度上限 m/s
SHORT_GAP = 5             # 短断档最大插值点数

def haversine(lat1, lon1, lat2, lon2):
    p1, p2 = np.radians(lat1), np.radians(lat2)
    dp = p2 - p1
    dl = np.radians(lon2 - lon1)
    a = np.sin(dp/2)**2 + np.cos(p1)*np.cos(p2)*np.sin(dl/2)**2
    return 2 * R_EARTH * np.arcsin(np.sqrt(a))

def wrap_diff(a, b):
    """环形差值 b-a，结果落在 (-180, 180]"""
    return (b - a + 180.0) % 360.0 - 180.0

def load_log():
    # 构造 60 s 示例日志：1 Hz，注入重复、断档、跳变
    t = pd.date_range("2024-06-01 08:00:00", periods=60, freq="1s")
    lat = 30.0 + np.arange(60) * 1e-5
    lon = 122.0 + np.arange(60) * 1e-5
    sog = np.full(60, 3.2)
    cog = np.full(60, 45.0)
    heading = np.full(60, 45.0)
    df = pd.DataFrame({"timestamp": t, "lat": lat, "lon": lon,
                       "sog": sog, "cog": cog, "heading": heading})
    # 注入跳变：第 20 点纬度突变
    df.loc[20, "lat"] += 0.01
    # 注入航向跳变：第 35 点艏向突变
    df.loc[35, "heading"] = 200.0
    # 注入重复：复制第 10 行
    df = pd.concat([df, df.iloc[[10]]], ignore_index=True)
    # 注入断档：删除第 45~49 行
    df = df.drop(index=range(45, 50)).reset_index(drop=True)
    return df

def clean_log(df):
    stats = {}
    df = df.sort_values("timestamp").reset_index(drop=True)
    stats["raw_rows"] = len(df)

    # ① 去重
    before = len(df)
    df = df.drop_duplicates(subset="timestamp", keep="first").reset_index(drop=True)
    stats["dup_removed"] = before - len(df)

    # ② 断档识别
    dt = df["timestamp"].diff().dt.total_seconds()
    gap_mask = dt > NOMINAL_DT * GAP_FACTOR
    stats["gap_segments"] = int(gap_mask.sum())

    # ③ 跳变检测（位置）
    d = haversine(df["lat"].shift(), df["lon"].shift(), df["lat"], df["lon"])
    v = d / dt
    pos_jump = v > V_MAX
    stats["pos_jumps"] = int(pos_jump.sum())

    # ③ 跳变检测（航向，环形）
    dyaw = wrap_diff(df["heading"].shift(), df["heading"])
    yaw_rate = dyaw.abs() / dt
    yaw_jump = yaw_rate > YAW_RATE_MAX
    stats["yaw_jumps"] = int(yaw_jump.sum())

    # ③ 速度异常
    sog_bad = (df["sog"] > SOG_MAX) | (df["sog"] < 0)
    stats["sog_bad"] = int(sog_bad.sum())

    # 跳变点置为 NaN，与断档合并
    bad = pos_jump | yaw_jump | sog_bad
    for col in ["lat", "lon", "sog", "cog", "heading"]:
        df.loc[bad, col] = np.nan

    # ④ 插值：短断档线性，长断档保留 NaN
    df = df.set_index("timestamp").asfreq("1s")
    for col in ["lat", "lon", "sog", "cog"]:
        df[col] = df[col].interpolate(method="linear", limit=SHORT_GAP)
    # 航向用环形插值：先展开再插值再折回
    h = df["heading"].copy()
    unwrapped = np.degrees(np.unwrap(np.radians(h.fillna(method="ffill"))))
    df["heading"] = pd.Series(unwrapped, index=df.index).interpolate(
        method="linear", limit=SHORT_GAP) % 360.0

    df = df.reset_index()
    stats["clean_rows"] = len(df)
    stats["nan_remaining"] = int(df[["lat","lon","sog","cog","heading"]].isna().any(axis=1).sum())
    return df, stats

def quality_report(raw, clean):
    key = ["lat", "lon", "sog", "cog", "heading"]
    def completeness(d):
        return d[key].notna().sum().sum() / (len(d) * len(key))
    span = (clean["timestamp"].max() - clean["timestamp"].min()).total_seconds()
    covered = clean["timestamp"].notna().sum() * NOMINAL_DT
    return {
        "完整率_前": round(completeness(raw), 4),
        "完整率_后": round(completeness(clean), 4),
        "时间连续率": round(covered / span, 4) if span else 0.0,
    }

if __name__ == "__main__":
    raw = load_log()
    clean, stats = clean_log(raw)
    print(stats)
    print(quality_report(raw, clean))

结果判读与验证

运行后应看到：dup_removed = 1、gap_segments = 1、pos_jumps = 1、yaw_jumps = 1，说明三类注入问题都被识别。清洗后 clean_rows 为 60（1 Hz 等间隔），nan_remaining 为 0，因为断档长度 5 点正好等于 SHORT_GAP 上限，被线性插值补全；若把断档拉长到 8 点，则 nan_remaining 应大于 0，这是“长断档不硬插”的预期行为。

验证要点：

• 抽查第 20 点纬度，清洗后应回到相邻点连线上，而不是保留突变值。
• 抽查第 35 点艏向，环形插值后应落在 45° 附近，不能出现 200° 或 45°→200° 的线性穿越。
• 对比 完整率_前 与 完整率_后，后者应更高；时间连续率 应接近 1.0。
• 用 clean.to_csv("usv_clean.csv", index=False) 导出，供后续重采样与剖面分析使用。