Compressor/_tmp_temperature/rescan.py

#!/usr/bin/env python3
"""Re-run the temperature rules from the cached hourly long table (no DB scan).

Usage::

    python _tmp_temperature/rescan.py --out _tmp_temperature/out_t1 \
        --mild-peer 4 --mild-spread 4 --abnormal-peer 6 --abnormal-spread 8 \
        --severe-peer 10 --severe-spread 15
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

import pandas as pd

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from _tmp_temperature import hf_validation, unified
from _tmp_temperature.config import build_unit_config

LONG = Path(__file__).resolve().parent / "out" / "pks_temperature_hourly.csv"


def load_wide(path: Path) -> dict[str, list[dict]]:
    frame = pd.read_csv(path, low_memory=False)
    frame["hour_start"] = pd.to_datetime(frame["小时"])
    result: dict[str, list[dict]] = {}
    for unit, group in frame.groupby("机组"):
        keys = ["hour_start", "批次"]
        wide = group.pivot_table(index=keys, columns="点位", values="均值", aggfunc="first")
        coverage = group.groupby(keys)["运行覆盖率"].first()
        std = group.pivot_table(index=keys, columns="点位", values="标准差", aggfunc="first")
        minraw = group.pivot_table(index=keys, columns="点位", values="原始最小值", aggfunc="first")
        zeros = group.pivot_table(index=keys, columns="点位", values="零值样本数", aggfunc="first")
        high = group.pivot_table(index=keys, columns="点位", values="高报警样本数", aggfunc="first")
        high_high = group.pivot_table(index=keys, columns="点位", values="高高报警样本数", aggfunc="first")
        rows = []
        for (hour_start, batch), values in wide.iterrows():
            cov = float(coverage.loc[(hour_start, batch)])
            row = {
                "机组": unit,
                "import_batch_id": int(batch),
                "hour_start": hour_start.to_pydatetime(),
                "coverage": cov,
                # The long table is already filtered to valid running hours by
                # scan.py, so every row is valid here (coverage may be <0.8 for
                # the 4# cumulative-counter path).
                "valid": True,
            }
            for column, value in values.items():
                if pd.notna(value):
                    row[f"{column}__avg"] = float(value)
                svalue = std.loc[(hour_start, batch), column] if column in std.columns else float("nan")
                if pd.notna(svalue):
                    row[f"{column}__std"] = float(svalue)
                mvalue = minraw.loc[(hour_start, batch), column] if column in minraw.columns else float("nan")
                if pd.notna(mvalue):
                    row[f"{column}__minraw"] = float(mvalue)
                zvalue = zeros.loc[(hour_start, batch), column] if column in zeros.columns else float("nan")
                if pd.notna(zvalue):
                    row[f"{column}__zeros"] = int(zvalue)
                h = high.loc[(hour_start, batch), column] if column in high.columns else float("nan")
                hh = high_high.loc[(hour_start, batch), column] if column in high_high.columns else float("nan")
                row[f"{column}__high"] = 0 if pd.isna(h) else int(h)
                row[f"{column}__high_high"] = 0 if pd.isna(hh) else int(hh)
            rows.append(row)
        rows.sort(key=lambda item: (item["import_batch_id"], item["hour_start"]))
        result[unit] = rows
    return result


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--long", default=str(LONG))
    parser.add_argument("--out", required=True)
    parser.add_argument("--min-hours", type=int, default=1)
    parser.add_argument("--confirm-hours", type=int, default=3)
    for name in ("mild_peer", "mild_spread", "abnormal_peer", "abnormal_spread",
                 "severe_peer", "severe_spread", "mild_peer_low", "abnormal_peer_low",
                 "severe_peer_low", "mild_z", "abnormal_z", "severe_z",
                 "mild_delta", "abnormal_delta", "severe_delta"):
        parser.add_argument(f"--{name.replace('_', '-')}", type=float)
    args = parser.parse_args()

    overrides = {
        key: getattr(args, key)
        for key in ("mild_peer", "mild_spread", "abnormal_peer", "abnormal_spread",
                    "severe_peer", "severe_spread", "mild_peer_low", "abnormal_peer_low",
                    "severe_peer_low", "mild_z", "abnormal_z", "severe_z",
                    "mild_delta", "abnormal_delta", "severe_delta", "confirm_hours")
        if getattr(args, key) is not None
    }

    by_unit = load_wide(Path(args.long))
    all_qc, all_features, all_segments, all_cycles, all_channels, all_observations, all_hf = [], [], [], [], [], [], []
    for unit, rows in by_unit.items():
        number = int(unit[0])
        config = build_unit_config(number)
        qc, features, segments, cycles, channels, observations, hf_segments = unified.run(rows, config, overrides, args.min_hours)
        print(f"{unit}: 有效小时 {sum(1 for row in rows if row['valid']):,} 段 {len(segments):,}")
        all_qc += qc
        all_features += features
        all_segments += segments
        all_cycles += cycles
        all_channels += channels
        all_observations += observations
        all_hf += hf_segments

    output = Path(args.out)
    confirmed_hf, hf_candidates = hf_validation.filter_hf(all_hf)
    unified.write_results(output, all_qc, all_features, all_segments, all_cycles, all_channels, all_observations,
                          confirmed_hf, hf_candidates)
    for level in ("轻微", "异常", "严重异常"):
        print(f"  {level}: {sum(1 for segment in all_segments if segment['温度阶段'] == level):,}")
    print(f"  观察: {len(all_observations):,}")
    print(f"  高频候选: {len(hf_candidates):,}  高频确认: {len(confirmed_hf):,}")
    print(f"  合并段: {len(all_segments) + len(confirmed_hf):,}")
    print(f"written -> {output}")


if __name__ == "__main__":
    main()