Compressor/backend/indicator_features/make_unit_shape_timeline.py

#!/usr/bin/env python3
"""Build a daily-sampled, full-history indicator shape-distance dashboard."""
from __future__ import annotations

import argparse
import csv
import json
import sys
from collections import defaultdict
from pathlib import Path

import numpy as np

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))

from indicator_features.extract_from_db import (  # noqa: E402
    CYLINDERS,
    SIDES,
    connect,
    fetch_cycles,
    point_name,
    sample_per_day,
)
from indicator_features.indicator_features import _plateau_levels, smooth  # noqa: E402

EVENTS_BY_UNIT = {
    "4": [
        {"date": "2026-06-10", "label": "气阀故障"},
    ],
    "5": [
        {"date": "2026-03-26", "label": "气阀故障"},
        {"date": "2026-04-01", "label": "气阀故障"},
        {"date": "2026-04-08", "label": "气阀故障"},
        {"date": "2026-06-12", "label": "气阀故障"},
        {"date": "2026-06-21", "label": "气阀故障"},
        {"date": "2026-07-11", "label": "气阀故障"},
    ],
    "6": [
        {"date": "2026-01-19", "label": "气阀故障"},
        {"date": "2026-01-30", "label": "气阀故障"},
        {"date": "2026-02-18", "label": "气阀故障"},
        {"date": "2026-02-27", "label": "气阀故障"},
        {"date": "2026-04-22", "label": "气阀故障"},
        {"date": "2026-04-26", "label": "气阀故障"},
    ],
    "7": [
        {"date": "2025-08-08", "label": "气阀污染"},
        {"date": "2025-10-07", "label": "气阀故障"},
        {"date": "2026-05-21", "label": "气阀事件"},
    ],
    "8": [
        {"date": "2025-02-17", "label": "气阀故障"},
        {"date": "2026-03-29", "label": "气阀故障"},
        {"date": "2026-04-01", "label": "气阀故障"},
    ],
    "9": [],
}
BAD_POINTS_BY_UNIT = {
    "4": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
    "5": set(),
    "6": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
    "7": {("四", "盖侧"), ("六", "盖侧")},
    "8": {("六", "盖侧")},
    "9": set(),
}
MIN_SPAN = 0.3


def fetch_files_with_cycles(connection, point: str) -> list[dict]:
    """Fetch metadata only for files that have a stored first cycle."""
    with connection.cursor() as cursor:
        cursor.execute(
            """
            SELECT wf.id, wf.sample_time, wf.rpm
            FROM wave_file wf
            WHERE wf.point_name = %s AND wf.measurement_type = '压力'
              AND EXISTS (
                SELECT 1 FROM wave_sample_one ws WHERE ws.wave_file_id = wf.id
              )
            ORDER BY wf.sample_time
            """,
            (point,),
        )
        return list(cursor.fetchall())


def profile(signal: np.ndarray, bins: int) -> np.ndarray | None:
    signal = np.asarray(signal, dtype=float)
    signal = signal[np.isfinite(signal)]
    if len(signal) < 60:
        return None
    pressure = smooth(signal, 21)
    pressure = np.where(np.isfinite(pressure), pressure, signal)
    suction, discharge, _ = _plateau_levels(pressure)
    span = discharge - suction
    if not np.isfinite(span) or span < MIN_SPAN:
        return None
    indices = np.arange(bins, dtype=float) * len(pressure) / bins
    resampled = np.interp(indices, np.arange(len(pressure)), pressure)
    result = (resampled - suction) / span
    return result if np.all(np.isfinite(result)) else None


def daily_template(samples: list[dict]) -> tuple[np.ndarray, dict[str, np.ndarray]]:
    by_day: dict[str, list[np.ndarray]] = defaultdict(list)
    for sample in samples:
        by_day[sample["time"][:10]].append(sample["profile"])
    daily_profiles = {
        day: np.median(np.vstack(values), axis=0)
        for day, values in by_day.items()
    }
    template = np.median(np.vstack(list(daily_profiles.values())), axis=0)
    return template, daily_profiles


def collect(unit: str, per_day: int, bins: int) -> dict:
    result = {
        "unit": unit,
        "bins": bins,
        "perDay": per_day,
        "events": EVENTS_BY_UNIT[unit],
        "points": [],
    }
    connection = connect()
    try:
        for cylinder in CYLINDERS:
            for side in SIDES:
                if (cylinder, side) in BAD_POINTS_BY_UNIT[unit]:
                    continue
                point = point_name(unit, cylinder, side)
                files = fetch_files_with_cycles(connection, point)
                picked = sample_per_day(files, per_day)
                cycles = fetch_cycles(connection, [int(row["id"]) for row in picked], 100)
                samples = []
                for row in picked:
                    file_id = int(row["id"])
                    values = profile(cycles.get(file_id, np.array([])), bins)
                    if values is None:
                        continue
                    samples.append({
                        "fileId": file_id,
                        "time": str(row["sample_time"]),
                        "profile": values,
                    })
                samples.sort(key=lambda item: item["time"])
                if not samples:
                    print(f"{point}: 无有效周期", flush=True)
                    continue
                template, daily_profiles = daily_template(samples)
                by_day_distance: dict[str, list[float]] = defaultdict(list)
                for sample in samples:
                    distance = float(np.linalg.norm(sample.pop("profile") - template))
                    sample["distance"] = round(distance, 6)
                    by_day_distance[sample["time"][:10]].append(distance)
                daily = [
                    {"date": day, "distance": round(float(np.median(values)), 6), "count": len(values)}
                    for day, values in sorted(by_day_distance.items())
                ]
                result["points"].append({
                    "name": f"{cylinder}缸{side}",
                    "availableFiles": len(files),
                    "sampleCount": len(samples),
                    "dayCount": len(daily_profiles),
                    "start": samples[0]["time"][:10],
                    "end": samples[-1]["time"][:10],
                    "template": np.round(template, 6).tolist(),
                    "samples": samples,
                    "daily": daily,
                })
                print(
                    f"{point}: 有周期文件 {len(files):,}，抽取有效周期 {len(samples):,}，"
                    f"有效日期 {len(daily_profiles)}，{samples[0]['time'][:10]}~{samples[-1]['time'][:10]}",
                    flush=True,
                )
    finally:
        connection.close()
    return result


def write_csv(data: dict, output: Path) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    with output.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=["机组", "测点", "文件ID", "采样时间", "形状距离", "当日样本数", "当日距离中位数"],
        )
        writer.writeheader()
        for point in data["points"]:
            daily = {row["date"]: row for row in point["daily"]}
            for sample in point["samples"]:
                summary = daily[sample["time"][:10]]
                writer.writerow({
                    "机组": data["unit"], "测点": point["name"], "文件ID": sample["fileId"],
                    "采样时间": sample["time"], "形状距离": sample["distance"],
                    "当日样本数": summary["count"], "当日距离中位数": summary["distance"],
                })


def make_html(data: dict) -> str:
    payload = json.dumps(data, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
    event_summary = "".join(
        f'<span class="event-line"></span>{event["date"]} {event["label"]}'
        for event in data["events"]
    ) or "该机组现有记录中没有气阀故障日期"
    excluded = "、".join(
        f"{cylinder}缸{side}" for cylinder, side in sorted(BAD_POINTS_BY_UNIT[data["unit"]])
    ) or "无已知坏道"
    return r'''<!doctype html>
<html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>__UNIT__号机全时段功图形状偏离</title><script src="https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"></script>
<style>
body{margin:0;background:#fff;color:#263238;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI","Microsoft YaHei",sans-serif}.wrap{width:100%;box-sizing:border-box;padding:20px 18px 42px}h1{text-align:center;margin:0 0 8px;font-size:26px}.intro{max-width:1500px;margin:0 auto 14px;color:#657279;font-size:13px;line-height:1.65}.events{text-align:center;font-size:13px;margin:10px 0 18px;color:#434f55}.event-line{display:inline-block;width:22px;border-top:1.5px dashed #a33;margin:0 5px 3px 14px}.grid{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:20px 16px}.card{min-width:0}.meta{font-size:12px;color:#68757b;line-height:1.5;min-height:36px;text-align:center}.chart{height:420px;border:1px solid #eee9e2;border-radius:7px}.legend{display:flex;gap:18px;justify-content:center;padding:7px 0 0;font-size:12px;color:#4a565c}.legend label{display:inline-flex;align-items:center;gap:5px;cursor:pointer}.legend input{margin:0;width:13px;height:13px}.dot{width:9px;height:9px;border-radius:50%;background:#6085a6;display:inline-block}.line{width:22px;border-top:3px solid #cf593d;display:inline-block}.foot{font-size:12px;color:#68757b;line-height:1.65;margin-top:18px}@media(max-width:1100px){.grid{grid-template-columns:repeat(2,minmax(0,1fr))}}@media(max-width:700px){.wrap{padding:12px 8px}.grid{grid-template-columns:1fr}.chart{height:390px}}
</style></head><body><main class="wrap"><h1>__UNIT__号机全时段功图形状偏离</h1>
<div class="intro">每个测点每天从实际存在首周期的文件中按时间均匀抽取最多__PER_DAY__个周期。每周期采用原始角度顺序，经21点平滑、90点重采样和吸排气平台归一化；不自动旋转相位。先求每日中位功图，再对所有日期等权取中位，形成该测点固定的全历史中心模板。Y轴为每个抽样周期与该模板的90点欧氏距离。</div>
<div class="events">__EVENTS__</div>
<div id="grid" class="grid"></div><div class="foot">排除情况：__EXCLUDED__。蓝点是抽样周期，红线是每日距离中位数；超过2天的缺测区间断线。虚线只在该测点实际数据范围内显示，且只标气阀故障日期。全历史模板是中心状态，不是经现场确认的健康模板；距离升高表示形状不像中心模板，不自动等于故障恶化。</div></main>
<script>
const DATA=__PAYLOAD__,charts=[];
function create(tag,cls,text){const node=document.createElement(tag);if(cls)node.className=cls;if(text)node.textContent=text;return node}
function brokenDaily(rows){const result=[];let previous=null;for(const row of rows){const time=Date.parse(row.date+'T12:00:00');if(previous!==null&&time-previous>2*86400000)result.push([new Date(previous+86400000).toISOString(),null]);result.push([row.date+'T12:00:00',row.distance,row.count]);previous=time}return result}
function eventLines(point){return DATA.events.filter(event=>event.date>=point.start&&event.date<=point.end).map(event=>({name:event.date+' '+event.label,xAxis:event.date+'T00:00:00',lineStyle:{color:'#a33',type:'dashed',width:1.3},label:{show:true,formatter:event.date+'\n'+event.label,position:'insideEndTop',rotate:90,color:'#8a302b',fontSize:11,distance:5}}))}
for(const point of DATA.points){
 const card=create('section','card'),meta=create('div','meta',`${point.start} ～ ${point.end} · ${point.dayCount}个有效日期 · ${point.sampleCount}个周期 · 全库有周期文件${point.availableFiles}个`),dom=create('div','chart'),labels=create('div','legend');card.append(meta,dom,labels);document.getElementById('grid').append(card);
 const scatterName='抽样周期',dailyName='每日中位';
 const scatter={name:scatterName,type:'scatter',symbolSize:5,large:true,largeThreshold:1000,data:point.samples.map(s=>({value:[s.time.replace(' ','T'),s.distance],fileId:s.fileId,time:s.time})),itemStyle:{color:'#6085a6',opacity:.58},emphasis:{itemStyle:{opacity:1}}};
 const daily={name:dailyName,type:'line',connectNulls:false,showSymbol:true,symbolSize:5,data:brokenDaily(point.daily),lineStyle:{color:'#cf593d',width:2.5},itemStyle:{color:'#cf593d'},markLine:{silent:false,symbol:['none','none'],animation:false,data:eventLines(point)}};
 const chart=echarts.init(dom);chart.setOption({animation:false,title:{text:point.name,left:'center',top:10,textStyle:{fontSize:17}},legend:{show:false,data:[scatterName,dailyName]},tooltip:{trigger:'item',confine:true,formatter:param=>{if(param.seriesName===scatterName)return `${point.name}<br>采样时间：${param.data.time}<br>形状距离：${Number(param.value[1]).toFixed(6)}<br>文件ID：${param.data.fileId}`;if(param.value&&param.value[1]!=null)return `${point.name}<br>日期：${String(param.value[0]).slice(0,10)}<br>每日中位：${Number(param.value[1]).toFixed(6)}<br>当日样本：${param.value[2]||''}`;return param.name||''}},grid:{left:68,right:24,top:72,bottom:74},xAxis:{type:'time',name:'采样日期',nameLocation:'middle',nameGap:32,axisLabel:{hideOverlap:true}},yAxis:{type:'value',name:'形状偏离距离',scale:true,splitLine:{lineStyle:{color:'#edf1f2'}}},dataZoom:[{type:'inside',filterMode:'filter'},{type:'slider',filterMode:'filter',bottom:15,height:18}],series:[scatter,daily]});
 [{name:scatterName,kind:'dot'},{name:dailyName,kind:'line'}].forEach(item=>{const label=create('label'),input=document.createElement('input');input.type='checkbox';input.checked=true;label.append(input,create('i',item.kind),document.createTextNode(item.name));input.onchange=()=>chart.dispatchAction({type:input.checked?'legendSelect':'legendUnSelect',name:item.name});labels.append(label)});charts.push(chart);
}
if(typeof echarts==='undefined')document.getElementById('grid').textContent='ECharts加载失败，请联网后刷新。';
window.addEventListener('resize',()=>charts.forEach(chart=>chart.resize()));
</script></body></html>'''.replace("__PAYLOAD__", payload).replace("__UNIT__", data["unit"]).replace("__PER_DAY__", str(data["perDay"])).replace("__EVENTS__", event_summary).replace("__EXCLUDED__", excluded)


def main() -> None:
    parser = argparse.ArgumentParser(description="生成单机全时段功图形状偏离网页")
    parser.add_argument("--unit", default="7", choices=("4", "5", "6", "7", "8", "9"))
    parser.add_argument("--per-day", type=int, default=6)
    parser.add_argument("--bins", type=int, default=90)
    parser.add_argument("--html", required=True)
    parser.add_argument("--csv", required=True)
    args = parser.parse_args()
    data = collect(args.unit, args.per_day, args.bins)
    html_path, csv_path = Path(args.html), Path(args.csv)
    html_path.parent.mkdir(parents=True, exist_ok=True)
    html_path.write_text(make_html(data), encoding="utf-8")
    write_csv(data, csv_path)
    print(f"HTML -> {html_path}")
    print(f"CSV  -> {csv_path}")


if __name__ == "__main__":
    main()