Compressor/backend/indicator_features/repair_compare.py

#!/usr/bin/env python3
"""Compare indicator features between a pre-repair and a post-repair segment.

For each cylinder/side it reports the segment medians and, importantly, the
*peer-relative* value (deviation from the same-stage/side median), so a repair
that only changes one cylinder can be separated from a process-wide shift.
Read-only.
"""
from __future__ import annotations

import argparse
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,
    fetch_file_rows,
    point_name,
    sample_per_day,
)
from indicator_features.indicator_features import CYLINDER_STAGE, cycle_features  # noqa: E402

FEATURES = ("p_discharge", "pressure_ratio", "discharge_frac", "indicated_work", "efficiency", "n_comp")


def extract(connection, unit, cylinders, sides, start, end, per_day):
    rows = []
    for cylinder in cylinders:
        for side in sides:
            point = point_name(unit, cylinder, side)
            files = fetch_file_rows(connection, point, start, end)
            picked = sample_per_day(files, per_day)
            if not picked:
                continue
            id_to_time = {int(r["id"]): r for r in picked}
            for file_id, signal in fetch_cycles(connection, list(id_to_time), 200).items():
                feats = cycle_features(signal, cylinder)
                if feats:
                    rows.append({"cylinder": cylinder, "side": side, "stage": CYLINDER_STAGE[cylinder], **feats})
    return rows


def summary(rows):
    table = {}
    groups = defaultdict(list)
    for row in rows:
        groups[(row["cylinder"], row["side"])].append(row)
    for key, items in groups.items():
        table[key] = {feature: float(np.median([r[feature] for r in items if r[feature] is not None]))
                      for feature in FEATURES}
    return table


def peer_relative(table, feature):
    """Deviation from the same-stage/side median, per cylinder."""
    by_stage = defaultdict(dict)
    for (cylinder, side), values in table.items():
        stage = CYLINDER_STAGE[cylinder]
        by_stage[(stage, side)][cylinder] = values[feature]
    result = {}
    for (stage, side), members in by_stage.items():
        clean = {c: v for c, v in members.items() if v is not None}
        if len(clean) < 2:
            continue
        med = float(np.median(list(clean.values())))
        if abs(med) < 1e-9:
            continue
        for cylinder, value in clean.items():
            result[(cylinder, side)] = (value - med) / med
    return result


def main():
    parser = argparse.ArgumentParser(description="修前/修后功图特征同级对照")
    parser.add_argument("--unit", required=True, choices=("7", "8", "9"))
    parser.add_argument("--cylinders", nargs="+", default=list(CYLINDERS), choices=CYLINDERS)
    parser.add_argument("--sides", nargs="+", default=list(SIDES), choices=SIDES)
    parser.add_argument("--pre", nargs=2, required=True, metavar=("START", "END"))
    parser.add_argument("--post", nargs=2, required=True, metavar=("START", "END"))
    parser.add_argument("--per-day", type=int, default=5)
    args = parser.parse_args()

    connection = connect()
    try:
        pre_rows = extract(connection, args.unit, args.cylinders, args.sides, args.pre[0], args.pre[1], args.per_day)
        post_rows = extract(connection, args.unit, args.cylinders, args.sides, args.post[0], args.post[1], args.per_day)
    finally:
        connection.close()

    pre, post = summary(pre_rows), summary(post_rows)
    print(f"{args.unit}号机  修前 {args.pre[0]}~{args.pre[1]} ({len(pre_rows)}行)  修后 {args.post[0]}~{args.post[1]} ({len(post_rows)}行)")
    if not pre_rows or not post_rows:
        print("某一段无数据"); return

    print(f"\n{'缸侧':>6} " + ' '.join(f"{f[:11]:>18}" for f in FEATURES))
    for key in sorted(set(pre) & set(post)):
        cells = []
        for feature in FEATURES:
            a, b = pre[key].get(feature), post[key].get(feature)
            if a and b and abs(a) > 1e-9:
                cells.append(f"{a:7.3f}->{b:6.3f}({(b-a)/a*100:+5.1f}%)")
            else:
                cells.append(f"{'':>18}")
        print(f"{key[0]+key[1]:>6} " + ' '.join(f"{c:>18}" for c in cells))

    print("\n同级 peer 相对（修前 -> 修后）:")
    for feature in ("pressure_ratio", "p_discharge", "efficiency"):
        rel_pre = peer_relative(pre, feature)
        rel_post = peer_relative(post, feature)
        print(f"  {feature}:")
        for key in sorted(set(rel_pre) & set(rel_post)):
            print(f"     {key[0]}缸{key[1]}: {rel_pre[key]*100:+.1f}% -> {rel_post[key]*100:+.1f}%")


if __name__ == "__main__":
    main()