Compressor/_tmp_cylinder_vibration/make_report.py

"""生成振动统一标签表、截图和 Word 报告。"""
from __future__ import annotations

import json
import os
import subprocess
import tempfile
from pathlib import Path

import pandas as pd
from docx import Document
from docx.enum.section import WD_ORIENT
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml.ns import qn
from docx.shared import Cm, Pt, RGBColor

from . import config, unified_labels

SHOT_SCRIPT = r'''
const { chromium } = require('playwright');
(async () => {
  const html = process.argv[2];
  const outDir = process.argv[3];
  const count = Number(process.argv[4]);
  const browser = await chromium.launch({ channel: 'chrome', headless: true });
  const page = await browser.newPage({ viewport: { width: 1700, height: 1200 } });
  await page.goto('file://' + html, { waitUntil: 'networkidle' });
  await page.waitForTimeout(1500);
  const cards = page.locator('.card');
  for (let i = 0; i < count; i++) {
    if (i > 0) await cards.nth(i).evaluate(el => el.click());
    await page.waitForTimeout(180);
    const name = 'shot_' + String(i + 1).padStart(4, '0') + '.png';
    await page.locator('#chart').screenshot({ path: outDir + '/' + name });
  }
  await browser.close();
})();
'''


def capture_screenshots(count: int) -> Path:
    out_dir = config.OUTPUT_DIR / "report_assets"
    out_dir.mkdir(parents=True, exist_ok=True)
    script = Path(tempfile.gettempdir()) / "compressor_cylinder_capture.js"
    script.write_text(SHOT_SCRIPT, encoding="utf-8")
    env = dict(os.environ)
    env["NODE_PATH"] = "/Users/zhouhao/Code/Python/Compressor/frontend/node_modules"
    subprocess.run(
        ["node", str(script), str(config.OUTPUT_DIR / "html" / "waveforms.html"), str(out_dir), str(count)],
        check=True,
        env=env,
        timeout=900,
    )
    return out_dir


def _set_black(style) -> None:
    style.font.color.rgb = RGBColor(0, 0, 0)
    rPr = style.element.get_or_add_rPr()
    color = rPr.find(qn("w:color"))
    if color is not None:
        for attr in ("themeColor", "themeTint", "themeShade"):
            color.attrib.pop(qn("w:" + attr), None)


def add_table(doc: Document, rows: list[list], headers: list[str]) -> None:
    table = doc.add_table(rows=1, cols=len(headers))
    table.style = "Table Grid"
    for index, header in enumerate(headers):
        cell = table.rows[0].cells[index]
        cell.text = str(header)
        for paragraph in cell.paragraphs:
            for run in paragraph.runs:
                run.font.name = "宋体"
                run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
                run.font.size = Pt(9.5)
                run.bold = True
                run.font.color.rgb = RGBColor(0, 0, 0)
    for row in rows:
        cells = table.add_row().cells
        for index, value in enumerate(row):
            cells[index].text = "" if value is None else str(value)
            for paragraph in cells[index].paragraphs:
                for run in paragraph.runs:
                    run.font.name = "宋体"
                    run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
                    run.font.size = Pt(9.5)
                    run.font.color.rgb = RGBColor(0, 0, 0)


def _style_attr_cell(cell, bold_label: bool = True) -> None:
    for paragraph in cell.paragraphs:
        for run in paragraph.runs:
            run.font.name = "宋体"
            run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
            run.font.size = Pt(9.5)
            run.font.color.rgb = RGBColor(0, 0, 0)


def add_attribute_table(doc: Document, attributes: list[tuple[str, object]], trigger: str) -> None:
    """标签属性：一行 3 个属性；阶段触发依据单独占一整行。"""
    table = doc.add_table(rows=0, cols=3)
    table.style = "Table Grid"
    for index in range(0, len(attributes), 3):
        row = table.add_row()
        for offset in range(3):
            cell = row.cells[offset]
            if index + offset < len(attributes):
                field, value = attributes[index + offset]
                paragraph = cell.paragraphs[0]
                label = paragraph.add_run(f"{field}：")
                label.bold = True
                label.font.name = "宋体"
                label._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
                label.font.size = Pt(9.5)
                label.font.color.rgb = RGBColor(0, 0, 0)
                content = paragraph.add_run("" if value is None else str(value))
                content.font.name = "宋体"
                content._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
                content.font.size = Pt(9.5)
                content.font.color.rgb = RGBColor(0, 0, 0)
    row = table.add_row()
    merged = row.cells[0].merge(row.cells[1]).merge(row.cells[2])
    paragraph = merged.paragraphs[0]
    label = paragraph.add_run("阶段触发依据：")
    label.bold = True
    for run in paragraph.runs:
        run.font.name = "宋体"
        run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
        run.font.size = Pt(9.5)
        run.font.color.rgb = RGBColor(0, 0, 0)
    content = paragraph.add_run("" if trigger is None else str(trigger))
    content.font.name = "宋体"
    content._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
    content.font.size = Pt(9.5)
    content.font.color.rgb = RGBColor(0, 0, 0)
    _style_attr_cell(merged)


def build_docx(unified: pd.DataFrame, validation: pd.DataFrame, assets: Path) -> Path:
    doc = Document()
    for section in doc.sections:
        section.orientation = WD_ORIENT.LANDSCAPE
        section.page_width = Cm(29.7)
        section.page_height = Cm(21.0)
        section.left_margin = Cm(2.0)
        section.right_margin = Cm(2.0)
        section.top_margin = Cm(2.0)
        section.bottom_margin = Cm(2.0)

    normal = doc.styles["Normal"]
    normal.font.name = "宋体"
    normal._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
    normal.font.size = Pt(12)
    _set_black(normal)
    normal.paragraph_format.line_spacing = 1.5
    for name, size, before, after in (("Heading 1", 16, 18, 10), ("Heading 2", 14, 12, 6)):
        style = doc.styles[name]
        style.font.name = "黑体"
        style._element.rPr.rFonts.set(qn("w:eastAsia"), "黑体")
        style.font.size = Pt(size)
        style.font.bold = True
        _set_black(style)
        style.paragraph_format.line_spacing = 1.0
        style.paragraph_format.space_before = Pt(before)
        style.paragraph_format.space_after = Pt(after)

    for text in ("组态振动数据演化过程", "故障标签手册"):
        title = doc.add_paragraph()
        title.alignment = WD_ALIGN_PARAGRAPH.CENTER
        run = title.add_run(text)
        run.bold = True
        run.font.name = "黑体"
        run._element.rPr.rFonts.set(qn("w:eastAsia"), "黑体")
        run.font.size = Pt(26)
        run.font.color.rgb = RGBColor(0, 0, 0)
    subtitle = doc.add_paragraph()
    subtitle.alignment = WD_ALIGN_PARAGRAPH.CENTER
    run = subtitle.add_run(f"振动事件 · 小时标签（共 {len(unified)} 条）")
    run.font.name = "宋体"
    run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
    run.font.size = Pt(16)
    run.font.color.rgb = RGBColor(0, 0, 0)
    date_line = doc.add_paragraph("2026 年 11 月")
    date_line.alignment = WD_ALIGN_PARAGRAPH.CENTER
    for run in date_line.runs:
        run.font.color.rgb = RGBColor(0, 0, 0)

    doc.add_heading("文档阅读说明", level=1)
    doc.add_paragraph(
        "本手册汇总 5 号机缸体振动的规则扫描标签，逐条给出标签属性、演化等级、局部基线、"
        "阶段触发依据和标签前后曲线。标签是规则扫描结果，不等同于物理故障结论；温度只作参考，不参与振动标签生成。"
    )
    doc.add_paragraph(
        "曲线截图来自 waveforms.html：只截取右侧 1×6 子图，不包含左侧标签列表。"
        "图中每个 15 分钟槽位拼接该文件 0~60ms 首周期波形；橙色为异常分数，紫色虚线为实际峭度；"
        "红色/黄色背景分别为当前缸/其他缸的标签段，淡蓝色为上下文内其他标签。"
    )

    doc.add_heading("一、标签数量总结", level=1)
    counts = unified.groupby(["阶段"]).size().reset_index(name="数量")
    add_table(doc, counts.values.tolist(), ["阶段", "数量"])
    doc.add_paragraph(f"标签总数：{len(unified)}；多特征确认：{int((unified['是否形成确认等级'] == '是').sum())}。")
    doc.add_paragraph(
        "严重异常、异常、轻微分别对应不同分数阈值；阈值只由局部基线和稳健尺度决定，"
        "不针对具体维修事件单独调整。"
    )

    doc.add_heading("二、事件验证", level=1)
    rows = []
    for row in validation.itertuples(index=False):
        rows.append([
            row.event_id, row.cylinder, row.hit, row.confirmed, row.start_time,
            row.end_time, row.duration_hours, row.direction, row.level, row.lead_days,
        ])
    add_table(
        doc,
        rows,
        ["事件", "缸", "命中", "确认", "开始", "结束", "小时", "方向", "等级", "提前天"],
    )
    doc.add_paragraph(
        "事件只用于验证和抽样观察，不参与标签生成；同一次事件可能对应多个标签，"
        "也可能只有温度证据而没有振动标签。"
    )

    doc.add_heading("三、字段说明", level=1)
    doc.add_paragraph(
        "统一核心字段包括：标签ID、标签类型、机组、批次、周期编号、阶段、阶段部位、阶段点位、"
        "开始小时、结束小时、持续自然小时数、有效运行小时数、基线值、正常波动带、阶段开始/结束/最大值、"
        "相对基线变化、阶段最大连续异常有效小时数、高/高高报警小时数、方向、特征类型、置信度、"
        "是否形成确认等级、阶段触发依据、事件关联、启动过渡比例、局部基线可用比例、数据来源、算法版本。"
    )
    doc.add_paragraph(
        "振动专有字段：近期基线RMS、RMS正常波动带、阶段开始/结束/最大RMS、阶段最大峭度、"
        "阶段最大峭度/基线、阶段最大残差峭度z、触发特征、多特征确认。"
    )

    doc.add_heading("四、标签详情", level=1)
    ordered = unified.to_dict("records")
    for index, row in enumerate(ordered, start=1):
        doc.add_heading(
            f"{row['阶段部位']}-{row['阶段']}-{row['方向']}-{str(row['开始小时'])[:16]}", level=2
        )
        attributes = [
            ("标签ID", row.get("标签ID")),
            ("阶段", row.get("阶段")),
            ("阶段部位", row.get("阶段部位")),
            ("开始小时", row.get("开始小时")),
            ("结束小时", row.get("结束小时")),
            ("持续自然小时数", row.get("持续自然小时数")),
            ("方向", row.get("方向")),
            ("近期基线RMS", row.get("近期基线RMS")),
            ("正常波动带", row.get("RMS正常波动带")),
            ("阶段最大RMS", row.get("阶段最大RMS")),
            ("阶段最大峭度", row.get("阶段最大峭度")),
            ("阶段最大峭度/基线", row.get("阶段最大峭度/基线")),
            ("是否形成确认等级", row.get("是否形成确认等级")),
        ]
        add_attribute_table(doc, attributes, row.get("阶段触发依据"))
        shot = assets / f"shot_{index:04d}.png"
        if shot.exists():
            doc.add_picture(str(shot), width=Cm(19.5))
            caption = doc.add_paragraph("图：标签前后运行曲线（仅右侧 1×6 子图）")
            caption.alignment = WD_ALIGN_PARAGRAPH.CENTER
        if index % 8 == 0:
            doc.add_page_break()

    path = config.BASE_DIR.parent / "总结文档" / "缸体振动演化过程故障标签手册.docx"
    path.parent.mkdir(parents=True, exist_ok=True)
    doc.save(path)
    return path


def main() -> None:
    hourly = pd.read_csv(config.OUTPUT_DIR / "vibration_hourly_labels.csv", parse_dates=["start_hour", "end_hour"])
    review = pd.read_csv(config.OUTPUT_DIR / "label_review.csv", parse_dates=["start_hour", "end_hour"])
    frame = pd.read_csv(config.OUTPUT_DIR / "vibration_features_full_local.csv", parse_dates=["sample_time"])
    validation = pd.read_csv(config.OUTPUT_DIR / "event_validation_full.csv")

    unified = unified_labels.build_vibration_unified(hourly, frame, review)
    unified_path = config.OUTPUT_DIR / "cylinder_vibration_label_unified.csv"
    unified.to_csv(unified_path, index=False, date_format="%Y-%m-%d %H:%M:%S")

    combined = unified_labels.build_combined_preview(config.OUTPUT_DIR)
    combined.to_csv(config.OUTPUT_DIR / "fault_label_unified_preview.csv", index=False, date_format="%Y-%m-%d %H:%M:%S")

    assets = capture_screenshots(len(unified))
    path = build_docx(unified, validation, assets)
    print(f"written -> {unified_path}")
    print(f"written -> {config.OUTPUT_DIR / 'fault_label_unified_preview.csv'}")
    print(f"written -> {path}")


if __name__ == "__main__":
    main()