#!/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'{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'''
__UNIT__号机全时段功图形状偏离
__UNIT__号机全时段功图形状偏离
每个测点每天从实际存在首周期的文件中按时间均匀抽取最多__PER_DAY__个周期。每周期采用原始角度顺序，经21点平滑、90点重采样和吸排气平台归一化；不自动旋转相位。先求每日中位功图，再对所有日期等权取中位，形成该测点固定的全历史中心模板。Y轴为每个抽样周期与该模板的90点欧氏距离。
__EVENTS__
'''.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()