#!/usr/bin/env python3
"""Standalone actual-waveform spring comparison for units 7, 8 and 9."""
from __future__ import annotations
import argparse
import csv
import json
from collections import defaultdict
from pathlib import Path
import numpy as np
from extract_from_db import connect, fetch_cycles
from indicator_features.indicator_features import CYLINDER_BORE_MM, _plateau_levels, smooth, volume_from_angle
BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}
def daily_median(samples, key):
if not samples:
return []
days = defaultdict(list)
for sample in samples:
days[sample["date"][:10]].append(sample[key])
return np.median([np.median(values, axis=0) for values in days.values()], axis=0).round(6).tolist()
def build(root):
bins = 90
machines = []
connection = connect()
try:
for unit in ("7", "8", "9"):
filename = f"高密度特征_{unit}号_2026.csv" if unit != "9" else "高密度特征_9号_2026春季.csv"
with (root / filename).open(encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))
rows = [r for r in rows if "2026-03-01" <= r["sample_time"][:10] < "2026-04-20"
and float(r.get("p_span") or 0) >= .3
and (unit, r["cylinder"], r["side"]) not in BAD]
grouped = defaultdict(list)
# Fetch bounded batches; retain only the compact plotted profiles.
for offset in range(0, len(rows), 100):
chunk = rows[offset:offset + 100]
cycles = fetch_cycles(connection, [int(r["wave_file_id"]) for r in chunk], 100)
for row in chunk:
signal = cycles.get(int(row["wave_file_id"]))
if signal is None or len(signal) < 60 or not np.all(np.isfinite(signal)):
continue
pressure = smooth(signal, 21)
suction, discharge, _ = _plateau_levels(pressure)
if discharge - suction < .3:
continue
indices = np.arange(bins) * len(pressure) / bins
absolute = np.interp(indices, np.arange(len(pressure)), pressure)
profile = (absolute - suction) / (discharge - suction)
grouped[(row["cylinder"], row["side"])].append({
"date": row["sample_time"], "profile": profile.round(6).tolist(),
"absolute": absolute.round(6).tolist(),
})
points = []
for (cylinder, side), samples in sorted(grouped.items()):
samples.sort(key=lambda s: s["date"])
march = [s for s in samples if s["date"][:10] < "2026-03-21"]
april = [s for s in samples if s["date"][:10] >= "2026-04-01"]
if not march or not april:
continue
template = np.asarray(daily_median(march, "profile"))
for sample in samples:
sample["distance"] = round(float(np.linalg.norm(np.asarray(sample["profile"]) - template)), 5)
sample["period"] = "3月上中旬" if sample["date"][:10] < "2026-03-21" else "4月" if sample["date"][:10] >= "2026-04-01" else "3月下旬"
points.append({
"point": f"{cylinder}缸{side}", "samples": samples,
"volume": volume_from_angle(bins, CYLINDER_BORE_MM[cylinder]).round(6).tolist(),
"preMedian": template.tolist(), "postMedian": daily_median(april, "profile"),
"preAbsolute": daily_median(march, "absolute"), "postAbsolute": daily_median(april, "absolute"),
"preDates": sorted({s["date"][:10] for s in march}),
"postDates": sorted({s["date"][:10] for s in april}),
"preCount": len(march), "postCount": len(april),
})
machines.append({"unit": unit, "points": points})
print(f"{unit}号：{len(points)}个可比较测点，{sum(len(p['samples']) for p in points)}条实际周期")
finally:
connection.close()
return {"angle": (np.arange(bins) * 360 / bins).tolist(), "machines": machines}
def make_html(data):
template = r'''
7、8、9号春季同期工况对照
7、8、9号春季同期工况对照
蓝色为3月上中旬，红色为4月。中位功图按每天等权计算；形状距离以同一测点3月上中旬模板为参考。每台机可切换测点，每行三图。
同期变化不等于季节因果，也不等于故障。不同机器的可用日期并不一致；缺测处不连线、不推断变化发生时间。P-V采用现有前端体积映射，没有自动相位旋转；平台归一化不能消除全部工况影响。单周期曲线经过21点平滑和90点重采样，适合看整体形状，不用于判断高频毛刺。
'''
return template.replace("__PAYLOAD__", json.dumps(data, ensure_ascii=False, separators=(",", ":"), allow_nan=False))
def main():
parser = argparse.ArgumentParser(description="生成实际功图的7/8/9号春季同期对照")
parser.add_argument("--root", default="cache/indicator")
parser.add_argument("--out", required=True)
args = parser.parse_args()
data = build(Path(args.root))
out = Path(args.out)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(make_html(data), encoding="utf-8")
print(f"春季同期网页 -> {out}")
if __name__ == "__main__":
main()