1. 航次故障日志原始记录格式解析与字段对齐

为什么原始日志不能直接分类

一次水下机器人航次中，故障信息往往分散在三类来源里：控制台输出（文本行，带本地时间）、传感器记录（CSV，带采样序号与设备时间）、任务日志（JSON 行，带任务阶段与 UTC 时间）。三者的时间基准、字段命名和粒度都不同，直接合并会导致同一故障被当成多条记录，或时间顺序错乱。本节的任务是把它们解析成一张字段统一、时间可对齐的结构化日志表。

三类来源的字段差异

统一字段模型

对齐后的目标表固定为 8 列，任何来源都必须映射到这套字段，映射不上的字段留空但保留行：

• ts_utc：统一 UTC 毫秒时间戳（对齐基准）
• source：console / sensor / mission
• level：INFO / WARN / ERROR / FATAL，传感器来源按阈值推断
• module：产生日志的模块或通道
• event_code：故障事件码，缺失时置 UNKNOWN
• message：原始描述文本
• value / unit：仅传感器来源有值

解析实现要点

下面给出一个可直接运行的最小实现，输入为三个示例文件，输出统一表并打印对齐校验结果。示例数据为教学构造，用于说明解析逻辑，不代表真实航次。

import csv, json, re
from datetime import datetime, timezone, timedelta

VOYAGE_DATE = "2024-05-12"          # 航次基准日期
LOCAL_TZ = timezone(timedelta(hours=8))  # 控制台本地时区

def parse_console(path):
    rows = []
    pat = re.compile(r"^(\d{2}:\d{2}:\d{2}\.\d{3})\s+\[(\w+)\]\s+(\w+)\s+(.*)$")
    with open(path, encoding="utf-8") as f:
        for line in f:
            m = pat.match(line.strip())
            if not m:
                continue
            t, lvl, mod, msg = m.groups()
            local = datetime.strptime(f"{VOYAGE_DATE} {t}", "%Y-%m-%d %H:%M:%S.%f")
            utc = local.replace(tzinfo=LOCAL_TZ).astimezone(timezone.utc)
            rows.append({
                "ts_utc": int(utc.timestamp() * 1000),
                "source": "console", "level": lvl.upper(),
                "module": mod, "event_code": "UNKNOWN",
                "message": msg, "value": None, "unit": None,
            })
    return rows

def parse_sensor(path, warn=40.0, err=45.0):
    rows = []
    with open(path, encoding="utf-8") as f:
        for r in csv.DictReader(f):
            v = float(r["val"])
            lvl = "ERROR" if v >= err else "WARN" if v >= warn else "INFO"
            rows.append({
                "ts_utc": int(r["epoch_ms"]),
                "source": "sensor", "level": lvl,
                "module": r["ch"], "event_code": "UNKNOWN",
                "message": f'{r["ch"]}={v}{r["unit"]}',
                "value": v, "unit": r["unit"],
            })
    return rows

def parse_mission(path):
    rows = []
    with open(path, encoding="utf-8") as f:
        for line in f:
            if not line.strip():
                continue
            o = json.loads(line)
            ts = datetime.fromisoformat(o["ts"].replace("Z", "+00:00"))
            rows.append({
                "ts_utc": int(ts.timestamp() * 1000),
                "source": "mission", "level": o.get("level", "INFO").upper(),
                "module": o.get("phase", "unknown"),
                "event_code": o.get("event_code", "UNKNOWN"),
                "message": o.get("desc", ""), "value": None, "unit": None,
            })
    return rows

def align(rows, tol_ms=50):
    base = sorted([r["ts_utc"] for r in rows if r["source"] == "mission"])
    for r in rows:
        if not base:
            r["align_flag"] = 1
            continue
        nearest = min(base, key=lambda b: abs(b - r["ts_utc"]))
        r["align_flag"] = 0 if abs(nearest - r["ts_utc"]) <= tol_ms else 1
    return sorted(rows, key=lambda r: r["ts_utc"])

if __name__ == "__main__":
    all_rows = (parse_console("console.log")
                + parse_sensor("sensor.csv")
                + parse_mission("mission.jsonl"))
    table = align(all_rows)
    cols = ["ts_utc","source","level","module","event_code","message","value","unit","align_flag"]
    with open("aligned_log.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=cols)
        w.writeheader()
        w.writerows(table)
    print(f"总记录 {len(table)}，超差 {sum(r['align_flag'] for r in table)} 条")

字段映射完整性检查

解析完成后必须逐来源核对映射，避免“看起来有表、实际丢字段”。检查项如下：

• 控制台：每行是否都提取到时间、级别、模块、消息四项；未匹配行数应单独计数并输出。
• 传感器：epoch_ms 是否全部可转为整数；ch 是否全部非空；val 是否可转浮点。
• 任务日志：ts 是否全部可解析；event_code 缺失行是否统一置为 UNKNOWN 而非丢弃。
• 统一表：ts_utc 无空值；source 取值仅限三种；行数应等于三来源有效行数之和。

时间戳对齐精度验证

对齐以任务日志的 UTC 毫秒为基准，容差取 50 ms。验证时关注两点：一是控制台本地时间换算后与任务日志的偏差是否在容差内，二是传感器 epoch_ms 与任务日志是否落在同一毫秒量级。若大量记录 align_flag=1，优先检查时区偏移是否设错、航次基准日期是否跨日、以及 epoch_ms 是秒还是毫秒。