Compressor/backend/indicator_features/spring_compare.py

"""Daily-weighted spring state comparison, not a test of seasonal causality."""
import csv
from collections import defaultdict
from pathlib import Path

import numpy as np


def main():
    root = Path("cache/indicator")
    output = []
    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") as handle:
            rows = list(csv.DictReader(handle))
        daily = defaultdict(list)
        for row in rows:
            day = row["sample_time"][:10]
            if not "2026-03-01" <= day < "2026-04-20":
                continue
            if (unit, row["cylinder"], row["side"]) in {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}:
                continue
            if float(row["p_span"]) < .3:
                continue
            keys = ["p_suction", "p_discharge", "p_span"] + [f"prof_{i:02d}" for i in range(12)]
            vector = np.array([float(row[k]) for k in keys])
            if np.all(np.isfinite(vector)):
                daily[(row["cylinder"], row["side"], day)].append(vector)
        for point in sorted({(c, s) for c, s, _ in daily}):
            periods = []
            for start, end in [("2026-03-01", "2026-03-21"), ("2026-03-21", "2026-04-01"), ("2026-04-01", "2026-04-20")]:
                selected = [(day, values) for (c, s, day), values in daily.items() if (c, s) == point and start <= day < end]
                median = np.median([np.median(values, axis=0) for _, values in selected], axis=0) if selected else None
                periods.append((selected, median))
            for label, index in [("3月上中旬至3月下旬", 1), ("3月上中旬至4月", 2)]:
                before, a = periods[0]
                after, b = periods[index]
                if a is None or b is None:
                    continue
                delta = b - a
                result = {"机组": unit, "测点": f"{point[0]}缸{point[1]}", "对照": label,
                          "前段日期": "/".join(sorted(d for d, _ in before)), "后段日期": "/".join(sorted(d for d, _ in after)),
                          "前段周期": sum(len(v) for _, v in before), "后段周期": sum(len(v) for _, v in after),
                          "吸气变化": round(delta[0], 5), "排气变化": round(delta[1], 5),
                          "幅度变化": round(delta[2], 5), "形状距离12点": round(float(np.linalg.norm(delta[3:])), 5)}
                output.append(result)
        print(f"{unit}号")
        for label in ("3月上中旬至3月下旬", "3月上中旬至4月"):
            selected = [r for r in output if r["机组"] == unit and r["对照"] == label]
            print(label, "测点", len(selected))
            for key in ("吸气变化", "排气变化", "幅度变化"):
                print(key, "中位", round(float(np.median([r[key] for r in selected])), 5) if selected else None,
                      "上升", sum(r[key] > 0 for r in selected))
            print("形状变化", sorted([(r["测点"], r["形状距离12点"]) for r in selected], key=lambda v: -v[1]))
    with (root / "春季同期跨机组对照.csv").open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(output[0]))
        writer.writeheader()
        writer.writerows(output)


if __name__ == "__main__":
    main()