Compressor/build_fault_package.py

import csv
import json
import os
import subprocess
import sys
from datetime import datetime, timedelta

import pymysql
from docx import Document
from docx.shared import Inches

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "backend"))
from app.db import get_connection  # noqa: E402
from app.services.data_service import DataService, PHASES  # noqa: E402

ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "故障分析包")
ECHARTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "frontend", "node_modules", "echarts", "dist", "echarts.min.js")
CHROME = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
SIDES = ("压力盖侧", "压力轴侧")

DATASETS = [
    {
        "key": "6号机_20260422_气阀故障_一级3缸",
        "unit": "6",
        "device_code": "6#",
        "device_part": "6号机组三缸",
        "cylinder": "三缸（一级3缸）",
        "event_date": "2026-04-22",
        "fault_desc": "6#机一级3缸排气温度高",
        "pks_item": "YSJ6_22",
        "pks_col": "YSJ_22",
        "pks_desc": "6#压缩机一级3#缸排气温度",
        "thr_high": 115.0,
        "thr_highhigh": 125.0,
        "raw_dir": r"E:\RAW\exp_6#_2026.1.1-2026.7.31_20260101000000_20260731235959",
        "segments": {
            "低": ("2026-04-12 13:44:00", "2026-04-12 15:31:00"),
            "高": ("2026-04-20 11:59:00", "2026-04-20 13:46:00"),
            "修复后": ("2026-04-25 09:59:00", "2026-04-25 11:46:00"),
        },
    },
    {
        "key": "4号机_20260610_气阀故障_一级2缸",
        "unit": "4",
        "device_code": "4#",
        "device_part": "4号机组二缸",
        "cylinder": "二缸（一级2缸）",
        "event_date": "2026-06-10",
        "fault_desc": "4#压缩机一级2#3#缸温度异常",
        "pks_item": "YSJ4_10",
        "pks_col": "YSJ_10",
        "pks_desc": "4#压缩机一级2#缸排气温度",
        "thr_high": 130.0,
        "thr_highhigh": 140.0,
        "raw_dir": r"E:\RAW\exp_4#_2026.1.1-2026.7.31_20260101000000_20260731235959",
        "segments": {
            "低": ("2026-06-08 01:59:00", "2026-06-08 03:46:00"),
            "高": ("2026-06-10 05:59:00", "2026-06-10 07:46:00"),
            "修复后": ("2026-06-11 18:59:00", "2026-06-11 20:46:00"),
        },
    },
    {
        "key": "4号机_20260610_气阀故障_一级3缸",
        "unit": "4",
        "device_code": "4#",
        "device_part": "4号机组三缸",
        "cylinder": "三缸（一级3缸）",
        "event_date": "2026-06-10",
        "fault_desc": "4#压缩机一级2#3#缸温度异常",
        "pks_item": "YSJ4_11",
        "pks_col": "YSJ_11",
        "pks_desc": "4#压缩机一级3#缸排气温度",
        "thr_high": 130.0,
        "thr_highhigh": 140.0,
        "raw_dir": r"E:\RAW\exp_4#_2026.1.1-2026.7.31_20260101000000_20260731235959",
        "segments": {
            "低": ("2026-06-08 01:59:00", "2026-06-08 03:46:00"),
            "高": ("2026-06-10 05:59:00", "2026-06-10 07:46:00"),
            "修复后": ("2026-06-11 18:59:00", "2026-06-11 20:46:00"),
        },
    },
]

SVC = DataService()
COPY_LIST = []


def ensure_dirs(ds):
    base = os.path.join(ROOT, ds["key"])
    subs = {name: os.path.join(base, name) for name in
            ("01_原始首周期数据", "02_PKS温度数据", "03_功图", "04_曲线图")}
    for path in subs.values():
        os.makedirs(path, exist_ok=True)
    return base, subs


def query_wave_files(device_part, side, start, end):
    with get_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT id, sample_time, file_name, rpm, cycle_start, cycle_end "
                "FROM wave_file WHERE device_part=%s AND device_point=%s "
                "AND sample_time>=%s AND sample_time<%s ORDER BY sample_time LIMIT 8",
                (device_part, side, start, end),
            )
            return cursor.fetchall()


def query_pks(device_code, column, start, end):
    with get_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                f"SELECT sample_time, `{column}` AS value FROM pks_long_sample "
                "WHERE device_code=%s AND sample_time>=%s AND sample_time<%s ORDER BY sample_time",
                (device_code, start, end),
            )
            return cursor.fetchall()


def write_first_cycle_csv(path, samples):
    with open(path, "w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.writer(handle)
        writer.writerow(["sample_index", "signal_value", "second_value"])
        for row in samples:
            writer.writerow([int(row[0]), float(row[1]), "" if row[2] != row[2] else float(row[2])])


def write_pks_csv(path, rows):
    with open(path, "w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.writer(handle)
        writer.writerow(["sample_time", "value"])
        for row in rows:
            writer.writerow([row["sample_time"].strftime("%Y-%m-%d %H:%M:%S"), row["value"]])


def screenshot(html_path, png_path):
    subprocess.run(
        [CHROME, "--headless=new", "--disable-gpu", "--hide-scrollbars",
         "--window-size=1180,660", "--force-device-scale-factor=2",
         "--virtual-time-budget=2000", f"--screenshot={png_path}",
         "file://" + html_path],
        check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )


def render_html(html_path, option, title):
    html = f"""<!doctype html><html><head><meta charset="utf-8"><title>{title}</title>
<style>html,body{{margin:0;background:#fff}}#c{{width:1160px;height:620px}}</style>
<script src="../../echarts.min.js"></script></head>
<body><div id="c"></div><script>var chart=echarts.init(document.getElementById('c'));
chart.setOption({json.dumps(option, ensure_ascii=False)});</script></body></html>"""
    with open(html_path, "w", encoding="utf-8") as handle:
        handle.write(html)


def pv_option(detail, side, title):
    volume = detail["volume"]
    series = []
    for name, color, start, end in PHASES:
        last = name == PHASES[-1][0]
        data = []
        for index, angle in enumerate(detail["angles360"]):
            x = volume[index]
            if x is None:
                continue
            in_phase = angle >= start if last else (start <= angle < end)
            if in_phase:
                data.append([x, detail["pressure"][index]])
        series.append({"name": name, "type": "line", "showSymbol": False,
                       "lineStyle": {"color": color, "width": 2}, "itemStyle": {"color": color}, "data": data})
    return {
        "title": {"text": title, "left": "center", "textStyle": {"fontSize": 14}},
        "animation": False,
        "grid": {"left": 70, "right": 40, "top": 50, "bottom": 60},
        "legend": {"top": 22, "textStyle": {"fontSize": 12}},
        "tooltip": {"trigger": "axis"},
        "xAxis": {"type": "value", "name": "气缸体积 (L)", "nameLocation": "middle", "nameGap": 30},
        "yAxis": {"type": "value", "name": "压力原始值", "nameLocation": "middle", "nameGap": 50},
        "series": series,
    }


def ptheta_option(detail, side, title):
    series = []
    for name, color, start, end in PHASES:
        last = name == PHASES[-1][0]
        data = []
        for index, angle360 in enumerate(detail["angles360"]):
            in_phase = angle360 >= start if last else (start <= angle360 < end)
            if in_phase:
                data.append([detail["angles"][index], detail["pressure"][index]])
        series.append({"name": name, "type": "line", "showSymbol": False,
                       "lineStyle": {"color": color, "width": 2}, "itemStyle": {"color": color}, "data": data})
    return {
        "title": {"text": title, "left": "center", "textStyle": {"fontSize": 14}},
        "animation": False,
        "grid": {"left": 70, "right": 40, "top": 50, "bottom": 60},
        "legend": {"top": 22, "textStyle": {"fontSize": 12}},
        "tooltip": {"trigger": "axis"},
        "xAxis": {"type": "value", "name": "曲轴角度 (°)", "nameLocation": "middle", "nameGap": 30, "min": 0, "max": 180},
        "yAxis": {"type": "value", "name": "压力原始值", "nameLocation": "middle", "nameGap": 50},
        "series": series,
    }


def pressure_overlay_option(series_map, title):
    series = []
    colors = {"压力盖侧": "#e05252", "压力轴侧": "#287f9e"}
    for side, values in series_map.items():
        series.append({"name": side, "type": "line", "showSymbol": False,
                       "lineStyle": {"color": colors[side], "width": 1.6},
                       "itemStyle": {"color": colors[side]}, "data": values})
    return {
        "title": {"text": title, "left": "center", "textStyle": {"fontSize": 14}},
        "animation": False,
        "grid": {"left": 70, "right": 40, "top": 50, "bottom": 60},
        "legend": {"top": 22, "textStyle": {"fontSize": 12}},
        "tooltip": {"trigger": "axis"},
        "xAxis": {"type": "value", "name": "首周期采样点序号", "nameLocation": "middle", "nameGap": 30},
        "yAxis": {"type": "value", "name": "压力原始值", "nameLocation": "middle", "nameGap": 50},
        "series": series,
    }


def temp_option(times, values, title, thr_high, thr_highhigh):
    data = [[time, value] for time, value in zip(times, values) if value is not None]
    return {
        "title": {"text": title, "left": "center", "textStyle": {"fontSize": 14}},
        "animation": False,
        "grid": {"left": 70, "right": 60, "top": 50, "bottom": 60},
        "tooltip": {"trigger": "axis"},
        "xAxis": {"type": "time", "axisLabel": {"fontSize": 10, "hideOverlap": True},
                  "name": "时间", "nameLocation": "middle", "nameGap": 30},
        "yAxis": {"type": "value", "name": "温度 (℃)", "nameLocation": "middle", "nameGap": 50},
        "series": [{
            "name": "温度", "type": "line", "showSymbol": False,
            "lineStyle": {"color": "#c62828", "width": 1.6}, "itemStyle": {"color": "#c62828"},
            "data": data,
            "markLine": {"silent": True, "symbol": "none", "data": [
                {"yAxis": thr_high, "label": {"formatter": f"高报 {thr_high:g}"}, "lineStyle": {"color": "#e6a23c"}},
                {"yAxis": thr_highhigh, "label": {"formatter": f"高高报 {thr_highhigh:g}"}, "lineStyle": {"color": "#f56c6c"}},
            ]},
        }],
    }


def temp_compare_option(series_list, title):
    series = []
    colors = {"低": "#409eff", "高": "#f56c6c", "修复后": "#67c23a"}
    for segment, values in series_list:
        series.append({"name": segment, "type": "line", "showSymbol": False,
                       "lineStyle": {"color": colors[segment], "width": 2},
                       "itemStyle": {"color": colors[segment]}, "data": values})
    return {
        "title": {"text": title, "left": "center", "textStyle": {"fontSize": 14}},
        "animation": False,
        "grid": {"left": 70, "right": 40, "top": 50, "bottom": 60},
        "legend": {"top": 22, "textStyle": {"fontSize": 12}},
        "tooltip": {"trigger": "axis"},
        "xAxis": {"type": "category", "data": [f"{m}" for m in range(0, 121, 5)], "name": "段内分钟"},
        "yAxis": {"type": "value", "name": "温度 (℃)", "nameLocation": "middle", "nameGap": 50},
        "series": series,
    }


def process_dataset(ds):
    base, subs = ensure_dirs(ds)
    temp_windows = {}
    for segment, (start, end) in ds["segments"].items():
        tag = f"{ds['unit']}号机_{ds['cylinder'].split('（')[0]}_{segment}"
        first_times = None
        last_times = None
        overlay_map = {}
        for side in SIDES:
            rows = query_wave_files(ds["device_part"], side, start, end)
            if not rows:
                print(f"WARN no wave files: {ds['key']} {segment} {side}")
                continue
            side_tag = "盖侧" if side == "压力盖侧" else "轴侧"
            if first_times is None:
                first_times = rows[0]["sample_time"]
            last_times = rows[-1]["sample_time"]
            overlay_values = []
            for row in rows:
                meta, samples = SVC._load_db_wave(int(row["id"]))
                time_label = row["sample_time"].strftime("%Y%m%d%H%M%S")
                csv_name = f"提取首周期_{tag}_{side_tag}_{time_label}.csv"
                write_first_cycle_csv(os.path.join(subs["01_原始首周期数据"], csv_name), samples)
                detail = DataService._build_period_detail(meta, samples, 1)
                if side == "压力盖侧":
                    overlay_values = [[i, float(v)] for i, v in enumerate(samples[:, 1])]
                pv = pv_option(detail, side, f"{ds['device_part']} {segment} {side} P-V 功图 ({row['sample_time']:%Y-%m-%d %H:%M})")
                pt = ptheta_option(detail, side, f"{ds['device_part']} {segment} {side} 压力-角度图 ({row['sample_time']:%Y-%m-%d %H:%M})")
                pv_html = os.path.join(subs["03_功图"], f"PV_{tag}_{side_tag}_{time_label}.html")
                pt_html = os.path.join(subs["03_功图"], f"Ptheta_{tag}_{side_tag}_{time_label}.html")
                render_html(pv_html, pv, "PV")
                render_html(pt_html, pt, "Ptheta")
                screenshot(pv_html, pv_html.replace(".html", ".png"))
                screenshot(pt_html, pt_html.replace(".html", ".png"))
                os.remove(pv_html)
                os.remove(pt_html)
                if side == "压力盖侧":
                    overlay_map["压力盖侧"] = overlay_values
                else:
                    overlay_map["压力轴侧"] = [[i, float(v)] for i, v in enumerate(samples[:, 1])]
                COPY_LIST.append({
                    "数据集": ds["key"], "段": segment, "机组": f"{ds['unit']}号机",
                    "缸": ds["cylinder"], "测点": side, "wave_file_id": int(row["id"]),
                    "device_part": ds["device_part"], "device_point": side,
                    "sample_time": row["sample_time"].strftime("%Y-%m-%d %H:%M:%S"),
                    "原始文件相对路径": row["file_name"],
                    "原始目录": ds["raw_dir"],
                    "完整路径": ds["raw_dir"] + "\\" + str(row["file_name"]),
                })
        if first_times and last_times:
            temp_start = first_times
            temp_end = last_times + timedelta(minutes=15)
            temp_rows = query_pks(ds["device_code"], ds["pks_col"], temp_start, temp_end)
            temp_csv = os.path.join(subs["02_PKS温度数据"], f"PKS温度_{tag}_{temp_start:%Y%m%d%H%M}-{temp_end:%Y%m%d%H%M}.csv")
            write_pks_csv(temp_csv, temp_rows)
            times = [r["sample_time"].strftime("%Y-%m-%d %H:%M:%S") for r in temp_rows]
            values = [None if r["value"] is None else round(float(r["value"]), 3) for r in temp_rows]
            temp_option_json = temp_option(times, values, f"{ds['device_part']} {segment} {ds['pks_desc']}", ds["thr_high"], ds["thr_highhigh"])
            html = os.path.join(subs["04_曲线图"], f"温度曲线_{tag}.html")
            render_html(html, temp_option_json, "temp")
            screenshot(html, html.replace(".html", ".png"))
            os.remove(html)
            step = max(1, len(values) // 24)
            temp_windows[segment] = values[::step][:25]
            if overlay_map:
                overlay_html = os.path.join(subs["04_曲线图"], f"首周期压力叠图_{tag}.html")
                render_html(overlay_html, pressure_overlay_option(overlay_map, f"{ds['device_part']} {segment} 首周期压力（盖侧/轴侧）"), "overlay")
                screenshot(overlay_html, overlay_html.replace(".html", ".png"))
                os.remove(overlay_html)
    if len(temp_windows) == 3:
        compare = temp_compare_option(
            [(seg, temp_windows[seg]) for seg in ("低", "高", "修复后")],
            f"{ds['device_part']} {ds['pks_desc']} 三段对比",
        )
        html = os.path.join(subs["04_曲线图"], f"温度三段对比_{ds['key']}.html")
        render_html(html, compare, "compare")
        screenshot(html, html.replace(".html", ".png"))
        os.remove(html)
    return base


def build_word(bases):
    document = Document()
    document.add_heading("压缩机气阀故障分析数据包说明", 0)
    document.add_paragraph(
        "本数据包围绕两起气阀故障（6号机 2026-04-22 一级3缸排气温度高；"
        "4号机 2026-06-10 一级2#3#缸温度异常）整理，覆盖 3 个气缸数据集。"
        "每个数据集按“温度低 / 温度高 / 停机修复后”三段各取约 2 小时（8 个首周期）的"
        "压力盖侧与压力轴侧数据，并配套 PKS 温度数据、功图与曲线图。"
    )
    document.add_heading("一、目录结构", level=1)
    document.add_paragraph("每个数据集文件夹包含 4 个子目录：")
    for name, desc in [
        ("01_原始首周期数据", "从数据库 wave_sample_one 导出的首周期采样（sample_index/signal_value/second_value）。原始 CSV 需按《原始文件拷贝清单.csv》从线下目录复制到此。"),
        ("02_PKS温度数据", "该段 2 小时窗口内的 PKS 温度原始数据（5 秒采样，DCS 系统）。"),
        ("03_功图", "每个首周期的压力-体积功图（P-V）与压力-角度图（P-θ），盖侧/轴侧分别成图，一幅图一个文件。"),
        ("04_曲线图", "PKS 温度曲线（含报警阈值线）、三段温度对比、首周期压力盖侧/轴侧叠图。"),
    ]:
        document.add_paragraph(f"{name}：{desc}", style="List Bullet")
    document.add_heading("二、数据集与时间窗", level=1)
    for ds in DATASETS:
        document.add_heading(f"{ds['key']}（{ds['device_part']}）", level=2)
        document.add_paragraph(f"故障：{ds['fault_desc']}（{ds['event_date']}）")
        document.add_paragraph(f"PKS 温度点：{ds['pks_item']} {ds['pks_desc']}；报警阈值：高报 {ds['thr_high']:g}℃ / 高高报 {ds['thr_highhigh']:g}℃")
        table = document.add_table(rows=1, cols=3)
        table.style = "Light Grid Accent 1"
        header = table.rows[0].cells
        header[0].text, header[1].text, header[2].text = "段", "起始", "结束"
        for segment, (start, end) in ds["segments"].items():
            cells = table.add_row().cells
            cells[0].text, cells[1].text, cells[2].text = segment, start, end
    document.add_heading("三、文件命名规则", level=1)
    for line in [
        "提取首周期_<机组>_<缸>_<段>_<盖侧|轴侧>_<时间YYYYMMDDHHMMSS>.csv",
        "PKS温度_<机组>_<缸>_<段>_<起>-<止>.csv",
        "PV_<机组>_<缸>_<段>_<盖侧|轴侧>_<时间>.png / Ptheta_同.png",
        "温度曲线_<机组>_<缸>_<段>.png / 温度三段对比_<数据集>.png / 首周期压力叠图_<机组>_<缸>_<段>.png",
    ]:
        document.add_paragraph(line, style="List Bullet")
    document.add_heading("四、CSV 字段说明", level=1)

    def add_field_table(title, fields):
        document.add_paragraph(title, style="List Bullet")
        table = document.add_table(rows=1, cols=2)
        table.style = "Light Grid Accent 1"
        table.rows[0].cells[0].text = "字段"
        table.rows[0].cells[1].text = "说明"
        for name, desc in fields:
            cells = table.add_row().cells
            cells[0].text, cells[1].text = name, desc

    add_field_table("01_原始首周期数据/提取首周期_*.csv（每个波形文件该周期内的原始采样）", [
        ("sample_index", "该波形文件内的原始采样点序号（从 0 开始，连续）。"),
        ("signal_value", "传感器信号原始值（压力盖侧/压力轴侧的压力值，未做标定换算）。"),
        ("second_value", "键相信号（二值与/或多值），用于识别每转脉冲与周期起点，>=30 视为脉冲触发。"),
    ])
    add_field_table("02_PKS温度数据/PKS温度_*.csv（该段 2 小时窗口）", [
        ("sample_time", "PKS 温度采样时刻，5 秒一条。"),
        ("value", "对应温度测点（见各数据集说明）的原始值，单位 ℃。"),
    ])
    add_field_table("原始文件拷贝清单.csv（供复制线下原始波形文件）", [
        ("数据集", "所属数据集（机组_日期_故障_缸）。"),
        ("段", "低 / 高 / 修复后。"),
        ("机组", "4号机 / 6号机。"),
        ("缸", "对应气缸（如 三缸（一级3缸））。"),
        ("测点", "压力盖侧 / 压力轴侧。"),
        ("wave_file_id", "数据库 wave_file 主键 id。"),
        ("device_part", "机组与部位，如 6号机组三缸。"),
        ("device_point", "测点，压力盖侧 / 压力轴侧。"),
        ("sample_time", "该波形文件的采样时刻。"),
        ("原始文件相对路径", "wave_file.file_name（相对路径）。"),
        ("原始目录", "该批次导入目录（compressor_data_import.directory_path）。"),
        ("完整路径", "原始目录 + 相对路径，可直接用于拷贝。"),
    ])
    for base in bases:
        document.add_heading(f"五、文件清单：{os.path.basename(base)}", level=1)
        for sub in ("01_原始首周期数据", "02_PKS温度数据", "03_功图", "04_曲线图"):
            folder = os.path.join(base, sub)
            files = sorted(os.listdir(folder)) if os.path.isdir(folder) else []
            document.add_paragraph(f"{sub}（{len(files)} 个文件）", style="List Bullet")
    doc_path = os.path.join(ROOT, "故障分析说明.docx")
    document.save(doc_path)
    return doc_path


def main():
    os.makedirs(ROOT, exist_ok=True)
    with open(ECHARTS, "rb") as src, open(os.path.join(ROOT, "echarts.min.js"), "wb") as dst:
        dst.write(src.read())
    bases = []
    for ds in DATASETS:
        print("== processing", ds["key"])
        bases.append(process_dataset(ds))
    if COPY_LIST:
        list_path = os.path.join(ROOT, "原始文件拷贝清单.csv")
        with open(list_path, "w", newline="", encoding="utf-8-sig") as handle:
            writer = csv.DictWriter(handle, fieldnames=list(COPY_LIST[0].keys()))
            writer.writeheader()
            writer.writerows(COPY_LIST)
        print("copy list rows:", len(COPY_LIST))
    build_word(bases)
    print("DONE ->", ROOT)


if __name__ == "__main__":
    main()