Compressor/_tmp_temperature/scan.py

#!/usr/bin/env python3
"""Cylinder-temperature scan (4#~9#) with rule-based 轻微/异常/严重异常 events.

Examples
--------
    python _tmp_temperature/scan.py --device 6# --start "2026-01-01" --end "2026-03-01"
    python _tmp_temperature/scan.py --device 4# 5# 6# 7# 8# 9#
    python _tmp_temperature/scan.py --device 5# --include-packing

All outputs stay inside ``_tmp_temperature/out/`` so the existing ``models/``
tree is untouched until the rules are validated.
"""
from __future__ import annotations

import argparse
import sys
from datetime import datetime, timedelta
from pathlib import Path

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

from _tmp_temperature import unified as temperature_algorithm
from _tmp_temperature import hf_validation
from _tmp_temperature import data as temperature_data
from _tmp_temperature.config import (
    CYLINDERS,
    PACKING,
    UnitConfig,
    build_unit_config,
    load_device_batches,
    normalize_devices,
)

OUT_DIR = Path(__file__).resolve().parent / "out"


def parse_time(value: str) -> datetime:
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(value, fmt)
        except ValueError:
            continue
    raise ValueError(f"无法解析时间：{value!r}")


def _auto_range(config: UnitConfig):
    from models.common.db import connect

    placeholders = ",".join(["%s"] * len(config.batches))
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                f"SELECT MIN(sample_time) AS mn, MAX(sample_time) AS mx "
                f"FROM pks_long_sample WHERE import_batch_id IN ({placeholders})",
                config.batches,
            )
            row = cursor.fetchone()
    finally:
        connection.close()
    start = row["mn"].replace(minute=0, second=0, microsecond=0)
    end = row["mx"].replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
    return start, end


def threshold_overrides(args) -> dict:
    mapping = {
        "mild_spread": args.mild_spread, "abnormal_spread": args.abnormal_spread, "severe_spread": args.severe_spread,
        "mild_peer": args.mild_peer, "abnormal_peer": args.abnormal_peer, "severe_peer": args.severe_peer,
        "mild_z": args.mild_z, "abnormal_z": args.abnormal_z, "severe_z": args.severe_z,
        "mild_delta": args.mild_delta, "abnormal_delta": args.abnormal_delta, "severe_delta": args.severe_delta,
        "confirm_hours": args.confirm_hours,
    }
    return {key: value for key, value in mapping.items() if value is not None}


def point_meta(config: UnitConfig, include_packing: bool) -> dict[str, dict]:
    points = list(CYLINDERS[config.unit])
    if include_packing and config.unit in PACKING:
        points.extend(PACKING[config.unit])
    return {
        column: {"stage": stage, "cylinder": cylinder, "label": config.labels.get(column, column)}
        for stage, cylinder, column in points
    }


def hourly_long_rows(rows: list[dict], config: UnitConfig, include_packing: bool) -> list[dict]:
    meta = point_meta(config, include_packing)
    output = []
    for row in rows:
        if not row.get("valid"):
            continue
        hour = row["hour_start"].strftime("%Y-%m-%d %H:%M:%S")
        for column, info in meta.items():
            if row.get(f"{column}__avg") is None:
                continue
            output.append({
                "机组": config.name,
                "批次": row["import_batch_id"],
                "小时": hour,
                "点位": column,
                "点位描述": info["label"],
                "级": info["stage"],
                "缸": info["cylinder"],
                "运行覆盖率": round(row.get("coverage") or 0.0, 4),
                "均值": row.get(f"{column}__avg"),
                "最小值": row.get(f"{column}__min"),
                "最大值": row.get(f"{column}__max"),
                "标准差": row.get(f"{column}__std"),
                "原始最小值": row.get(f"{column}__minraw"),
                "零值样本数": row.get(f"{column}__zeros"),
                "高报警样本数": row.get(f"{column}__high"),
                "高高报警样本数": row.get(f"{column}__high_high"),
            })
    return output


def main() -> None:
    parser = argparse.ArgumentParser(description="PKS 缸体温度扫描（4#~9#）")
    parser.add_argument("--device", nargs="+", default=["4#", "5#", "6#", "7#", "8#", "9#"])
    parser.add_argument("--start")
    parser.add_argument("--end")
    parser.add_argument("--chunk-days", type=int, default=30)
    parser.add_argument("--include-packing", action="store_true", help="5#/6# 同时扫描气缸填料温度")
    parser.add_argument("--output-dir", default=str(OUT_DIR))
    parser.add_argument("--no-hourly-long", action="store_true")
    for name in ("mild_spread", "abnormal_spread", "severe_spread",
                 "mild_peer", "abnormal_peer", "severe_peer",
                 "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)
    parser.add_argument("--confirm-hours", type=int)
    parser.add_argument("--min-hours", type=int, default=1, help="最短输出段小时数（默认1，观察段也输出）")
    args = parser.parse_args()

    selected = normalize_devices(args.device)
    overrides = threshold_overrides(args)

    all_rows: list[dict] = []
    all_qc: list[dict] = []
    all_features: list[dict] = []
    all_segments: list[dict] = []
    all_cycles: list[dict] = []
    all_channels: list[dict] = []
    all_observations: list[dict] = []
    all_hf: list[dict] = []
    all_long: list[dict] = []

    for device in selected:
        unit = int(device[0])
        config = build_unit_config(unit, include_packing=args.include_packing)
        if args.start or args.end:
            start = parse_time(args.start) if args.start else None
            end = parse_time(args.end) if args.end else None
            if start is None or end is None:
                auto_start, auto_end = _auto_range(config)
                start = start or auto_start
                end = end or auto_end
        else:
            start, end = _auto_range(config)

        print(f"[{device}] 批次 {config.batches}  {start} ~ {end}  扫描 {len(config.columns)} 个缸温点 ...", flush=True)
        rows = temperature_data.fetch_unit_hourly(config, start, end, args.chunk_days)
        qc, features, segments, cycles, channels, observations, hf_segments = temperature_algorithm.run(rows, config, overrides, args.min_hours)
        print(
            f"[{device}] 有效小时 {sum(1 for row in rows if row.get('valid')):,}  "
            f"特征 {len(features):,}  段 {len(segments):,}  通道异常 {len(channels):,}",
            flush=True,
        )
        all_rows.extend(rows)
        all_qc.extend(qc)
        all_features.extend(features)
        all_segments.extend(segments)
        all_cycles.extend(cycles)
        all_channels.extend(channels)
        all_observations.extend(observations)
        all_hf.extend(hf_segments)
        if not args.no_hourly_long:
            all_long.extend(hourly_long_rows(rows, config, args.include_packing))

    output_dir = Path(args.output_dir)
    confirmed_hf, hf_candidates = hf_validation.filter_hf(all_hf)
    temperature_algorithm.write_results(output_dir, all_qc, all_features, all_segments, all_cycles, all_channels,
                                         all_observations, confirmed_hf, hf_candidates)
    if not args.no_hourly_long:
        fields = ["机组", "批次", "小时", "点位", "点位描述", "级", "缸", "运行覆盖率",
                  "均值", "最小值", "最大值", "标准差", "原始最小值", "零值样本数", "高报警样本数", "高高报警样本数"]
        temperature_algorithm.write_csv(output_dir / "pks_temperature_hourly.csv", all_long, fields, {field: field for field in fields})
        print(f"温度小时长表：{len(all_long):,} 行 -> {output_dir / 'pks_temperature_hourly.csv'}")

    print(f"结果目录：{output_dir}")
    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):,}")
    for path in ("温度小时特征.csv", "温度异常段.csv", "温度演化周期.csv", "温度通道质控.csv"):
        print(f"  {path}")


if __name__ == "__main__":
    main()