Compressor/backend/indicator_features/new_unit_events.py

"""Sample and rank 4/6 unit valve events without scanning wave_sample_one."""
from __future__ import annotations

from collections import defaultdict
from datetime import datetime, timedelta

import numpy as np

from extract_from_db import fetch_cycles, fetch_file_rows, point_name, sample_per_day
from indicator_features.indicator_features import CYLINDER_BORE_MM, _plateau_levels, smooth, volume_from_angle

CYLINDERS = ("一", "二", "三", "四", "五", "六")
SIDES = ("盖侧", "轴侧")
BAD = {
    "4": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
    "6": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
}
EVENTS = [
    {
        "key": "4号 2026-06-10 气阀事件",
        "unit": "4", "center": "2026-06-10", "pre_end": "2026-06-10", "post_start": "2026-06-10",
        "start": "2026-05-11", "end": "2026-07-11", "markers": ["2026-06-10"],
    },
    {
        "key": "6号 2026-04-22~04-26 气阀维修过程",
        "unit": "6", "center": "2026-04-24", "pre_end": "2026-04-22", "post_start": "2026-04-26",
        "start": "2026-03-23", "end": "2026-05-27", "markers": ["2026-04-22", "2026-04-26"],
    },
]


def _median(items, key):
    return np.median([item[key] for item in items], axis=0).round(6).tolist() if items else []


def collect_events(connection, per_day=6, bins=90):
    result = {}
    for config in EVENTS:
        center = datetime.fromisoformat(config["center"])
        pre_end = datetime.fromisoformat(config["pre_end"])
        post_start = datetime.fromisoformat(config["post_start"])
        points = {}
        for cylinder in CYLINDERS:
            for side in SIDES:
                if (cylinder, side) in BAD[config["unit"]]:
                    continue
                files = fetch_file_rows(
                    connection, point_name(config["unit"], cylinder, side), config["start"], config["end"]
                )
                picked = sample_per_day(files, per_day)
                cycles = fetch_cycles(connection, [int(row["id"]) for row in picked], 100) if picked else {}
                samples = []
                for row in picked:
                    signal = cycles.get(int(row["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)
                    span = discharge - suction
                    if not np.isfinite(span) or span < .3:
                        continue
                    indices = np.arange(bins) * len(pressure) / bins
                    absolute = np.interp(indices, np.arange(len(pressure)), pressure)
                    profile = (absolute - suction) / span
                    volume = np.interp(
                        indices, np.arange(len(pressure)),
                        volume_from_angle(len(pressure), CYLINDER_BORE_MM[cylinder]),
                    )
                    time = row["sample_time"]
                    if not isinstance(time, datetime):
                        time = datetime.fromisoformat(str(time))
                    if time < pre_end:
                        phase = "修前"
                    elif time >= post_start:
                        phase = "修后"
                    else:
                        phase = "过程内"
                    samples.append({
                        "date": time.strftime("%Y-%m-%d %H:%M:%S"),
                        "day": (time.date() - center.date()).days,
                        "phase": phase,
                        "profile": profile.round(6).tolist(),
                        "absolute": absolute.round(6).tolist(),
                        "volume": volume.round(6).tolist(),
                    })
                samples.sort(key=lambda sample: sample["date"])
                pre = [sample for sample in samples if sample["phase"] == "修前"]
                post = [sample for sample in samples if sample["phase"] == "修后"]
                if not pre or not post:
                    continue
                before = np.median([sample["profile"] for sample in pre], axis=0)
                after = np.median([sample["profile"] for sample in post], axis=0)
                points[(cylinder, side)] = {
                    "samples": samples, "pre": pre, "post": post,
                    "before": before, "after": after, "delta": after - before,
                }

        common = np.median([point["delta"] for point in points.values()], axis=0)
        cylinder_scores = defaultdict(list)
        for (cylinder, _), point in points.items():
            point["total"] = float(np.linalg.norm(point["delta"]))
            point["residual"] = float(np.linalg.norm(point["delta"] - common))
            cylinder_scores[cylinder].append(point["residual"])
        ranking = sorted(
            ((cylinder, max(scores)) for cylinder, scores in cylinder_scores.items()),
            key=lambda item: item[1], reverse=True,
        )
        event = {
            "label": config["key"], "unit": config["unit"], "markers": config["markers"],
            "ranking": [{"cylinder": cylinder, "residual": round(score, 5)} for cylinder, score in ranking],
            "sides": {},
        }
        for (cylinder, side), point in points.items():
            template = point["before"]
            for sample in point["samples"]:
                sample["distance"] = round(float(np.linalg.norm(np.asarray(sample["profile"]) - template)), 5)
            name = f"{cylinder}缸{side}"
            event["sides"][name] = {
                "samples": point["samples"],
                "counts": {"pre": len(point["pre"]), "post": len(point["post"])},
                "preMedian": point["before"].round(6).tolist(),
                "postMedian": point["after"].round(6).tolist(),
                "preAbsoluteMedian": _median(point["pre"], "absolute"),
                "postAbsoluteMedian": _median(point["post"], "absolute"),
                "trend": [{"date": sample["date"], "day": sample["day"], "distance": sample["distance"]}
                          for sample in point["samples"]],
                "totalChange": round(point["total"], 5),
                "localResidual": round(point["residual"], 5),
                "markerDays": [
                    (datetime.fromisoformat(marker).date() - center.date()).days
                    for marker in config["markers"]
                ],
            }
        coverage = sorted({sample["date"][:10] for point in event["sides"].values() for sample in point["samples"]})
        event["label"] += f" · 实际功图 {coverage[0]} 至 {coverage[-1]}"
        result[config["key"]] = event
    return result


def top_cylinder_event(event, limit=3):
    cylinders = {item["cylinder"] for item in event["ranking"][:limit]}
    selected = dict(event)
    selected["sides"] = {
        name: point for name, point in event["sides"].items()
        if name[0] in cylinders
    }
    return selected