Compressor/backend/indicator_features/bins_common.py

#!/usr/bin/env python3
"""Shared metric registry for the per-angle-bin indicator analysis.

Two families:

* ``BIN_METRICS``   : computed on the raw samples **inside each angular bin**;
* ``CYCLE_METRICS`` : computed **once per cycle** (plateau levels, valve-event
  angles, corner slopes) and repeated on every bin row of that cycle.

``VOLATILE`` metrics are amplitude/irregularity-like (a change means the local
waveform got rougher/noisier); the rest are level/shape-like.
"""
from __future__ import annotations

BIN_METRICS = (
    # 旧的 6 个
    "mean", "range", "std", "mad", "slope", "roughness",
    # 分布
    "p10", "p50", "p90", "iqr", "skew", "kurt",
    # 导数 / 振荡
    "d_abs_mean", "d_abs_max", "d_rms", "d2_rms", "zero_cross", "n_extrema",
    # 形状 / 复杂度
    "d2_abs_max", "tv", "mean_cross", "mobility", "complexity",
)

CYCLE_METRICS = (
    "p_suction", "p_discharge", "pressure_ratio", "suction_frac", "discharge_frac",
    "angle_min", "angle_max", "cross_up", "cross_down",
    "max_rise", "max_fall", "max_rise_angle", "max_fall_angle",
)

META_COLUMNS = (
    "unit", "cylinder", "stage", "side", "point_name", "wave_file_id",
    "sample_time", "rpm", "n_samples", "p_min", "p_max", "p_span",
    "bin", "bin_start", "n_raw",
)

VOLATILE = (
    "std", "mad", "range", "roughness",
    "d_abs_mean", "d_abs_max", "d_rms", "d2_rms", "zero_cross", "n_extrema",
    "d2_abs_max", "tv", "mean_cross", "mobility", "complexity",
)

# 整周期指标里偏“波动/不稳定”的（其余按水平处理）
CYCLE_VOLATILE = ("suction_frac", "discharge_frac", "max_rise", "max_fall")

# 角度型指标：差值与中位数必须按 0-360 环形处理（否则 wrap 会产生假跳变）
ANGLE_METRICS = ("angle_min", "angle_max", "cross_up", "cross_down", "max_rise_angle", "max_fall_angle")

# 随压力幅度（p_span）成比例的量；--normalize span 时除以本周期 p_span 以消工况
BIN_SCALED = (
    "mean", "range", "std", "mad", "slope", "roughness",
    "p10", "p50", "p90", "iqr",
    "d_abs_mean", "d_abs_max", "d_rms", "d2_rms", "d2_abs_max", "tv",
)
CYCLE_SCALED = ("p_suction", "p_discharge", "max_rise", "max_fall")


def apply_normalization(rows, mode: str) -> None:
    """In place: divide pressure-scale metrics by each cycle's own ``p_span``."""
    if mode != "span":
        return
    scaled = set(BIN_SCALED) | set(CYCLE_SCALED)
    for row in rows:
        span = row.get("p_span") or 0.0
        for name in scaled:
            if row.get(name) is None:
                continue
            row[name] = row[name] / span if span > 1e-9 else None

ALL_METRICS = BIN_METRICS + CYCLE_METRICS
FIELDNAMES = META_COLUMNS + BIN_METRICS + CYCLE_METRICS