3. 任务阶段自动划分与状态机重建

从连续日志到阶段边界：规则状态机的设计思路

本案例的输入是一份已经完成时间对齐的 AUV 任务日志，每行包含 t（秒）、depth（米）、heading（度）、mode（模式字，如 IDLE/DIVE/CRUISE/ASCEND/RECOVER）、rpm（推进器转速）。任务是把这些逐行记录自动切分为下潜、定深航行、上浮、回收四个阶段，并输出每个阶段的起止时间与持续时间。

核心难点在于：模式字并不总是可靠——下潜过程中可能短暂出现 CRUISE，定深航行时深度会有小幅波动。因此不能只靠单一字段，而要用“模式字 + 深度趋势 + 航向指令”三者共同判定，并引入最小驻留时间抑制抖动。下面给出一个可运行的规则状态机实现。

状态机模型与判定规则

定义四个状态，转移条件基于滑动窗口内的深度变化率 dz/dt 与模式字：

• DIVE（下潜）：mode=DIVE 或 dz/dt > 0.15 m/s 持续 3 秒以上。
• CRUISE（定深航行）：|dz/dt| ≤ 0.15 m/s 且 mode=CRUISE，深度在目标深度 ±2 m 内。
• ASCEND（上浮）：dz/dt < -0.15 m/s 持续 3 秒以上。
• RECOVER（回收）：mode=RECOVER 且深度 < 1 m。

为避免单点噪声导致状态反复跳变，状态切换需满足最小驻留时间 5 秒；若候选状态未驻留满 5 秒，则保持原状态。

可运行的阶段划分实现

下面代码读取日志列表（每行一个字典），计算滑动窗口深度变化率，按状态机规则输出阶段边界。示例数据为教学构造，用于演示判定逻辑。

import csv

MIN_DWELL = 5.0      # 最小驻留时间（秒）
DZ_THRESH = 0.15     # 深度变化率阈值（m/s）
WIN = 3              # 滑动窗口点数

def load_log(path):
    rows = []
    with open(path, newline="") as f:
        for r in csv.DictReader(f):
            rows.append({
                "t": float(r["t"]),
                "depth": float(r["depth"]),
                "heading": float(r["heading"]),
                "mode": r["mode"].strip().upper(),
                "rpm": float(r["rpm"]),
            })
    return rows

def depth_rate(rows, i):
    """用窗口首尾点估算 dz/dt"""
    j = max(0, i - WIN + 1)
    dt = rows[i]["t"] - rows[j]["t"]
    if dt <= 0:
        return 0.0
    return (rows[i]["depth"] - rows[j]["depth"]) / dt

def classify(rows, i):
    dz = depth_rate(rows, i)
    m = rows[i]["mode"]
    d = rows[i]["depth"]
    if m == "RECOVER" and d < 1.0:
        return "RECOVER"
    if dz > DZ_THRESH or m == "DIVE":
        return "DIVE"
    if dz < -DZ_THRESH:
        return "ASCEND"
    if m == "CRUISE" and abs(dz) <= DZ_THRESH:
        return "CRUISE"
    return None  # 无法判定，沿用当前状态

def segment(rows):
    phases = []
    cur = None
    cand = None
    cand_start = 0.0
    for i, r in enumerate(rows):
        label = classify(rows, i)
        if label is None:
            continue
        if cur is None:
            cur, cand, cand_start = label, label, r["t"]
            phases.append({"phase": cur, "start": r["t"]})
            continue
        if label == cur:
            cand = cur
            continue
        # 候选新状态
        if label != cand:
            cand, cand_start = label, r["t"]
        if r["t"] - cand_start >= MIN_DWELL:
            phases[-1]["end"] = cand_start
            cur = cand
            phases.append({"phase": cur, "start": cand_start})
    if phases:
        phases[-1]["end"] = rows[-1]["t"]
    for p in phases:
        p["duration"] = round(p["end"] - p["start"], 1)
    return phases

if __name__ == "__main__":
    rows = load_log("auv_task.csv")
    for p in segment(rows):
        print(f'{p["phase"]:8s} {p["start"]:7.1f} -> {p["end"]:7.1f}  '
              f'持续 {p["duration"]:5.1f} s')

示例日志与预期输出

构造一段 60 秒的教学日志片段（节选关键点），用于验证状态机：

按上述规则运行，预期输出（教学示例，非实测）：

DIVE       5.0 ->  22.0  持续  17.0 s
CRUISE    22.0 ->  48.0  持续  26.0 s
ASCEND    48.0 ->  56.0  持续   8.0 s
RECOVER   56.0 ->  60.0  持续   4.0 s