Compressor/_tmp_temperature/hf_validation.py

"""Validate high-frequency candidates with 5-second raw data.

The hourly ``STDDEV`` used by the first pass also reacts to a slow one-sided
ramp inside an hour.  This module removes a rolling slow trend and counts
short reversals, then keeps only candidates that look like repeated up/down
movement rather than a load ramp.
"""
from __future__ import annotations

import re
from datetime import datetime, timedelta

import numpy as np
import pandas as pd
from scipy.signal import find_peaks

from models.common.db import connect

DEFAULT_POINT = "YSJ_21"
TREND_WINDOW = 121          # 10-minute rolling median at 5s resolution
MIN_PROMINENCE = 0.3        # degC; prevent sensor noise from counting
MIN_REVERSALS_PER_HOUR = 5.0
MIN_RESIDUAL_STD = 0.5      # degC


def _parse(value: str) -> datetime:
    return datetime.strptime(str(value).strip(), "%Y-%m-%d %H:%M:%S")


def _point(row: dict) -> str:
    point = str(row.get("阶段点位") or DEFAULT_POINT).strip()
    return point if re.fullmatch(r"YSJ_\d+", point) else DEFAULT_POINT


def _metrics(row: dict) -> dict:
    batch = int(row["批次"])
    start = _parse(row["开始小时"])
    end = _parse(row["结束小时"]) + timedelta(hours=1)
    point = _point(row)
    query_start = start - timedelta(hours=1)
    query_end = end + timedelta(hours=1)
    with connect() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                f"SELECT sample_time, `{point}` AS value "
                "FROM pks_long_sample "
                "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time <= %s "
                "ORDER BY sample_time",
                (batch, query_start, query_end),
            )
            rows = cursor.fetchall()
    if not rows:
        return {"去趋势残差标准差": "", "每小时反复升降次数": "", "是否确认": "否"}

    times = pd.to_datetime([item["sample_time"] for item in rows])
    values = np.asarray([float(item["value"] or 0.0) for item in rows], dtype=float)
    values[values <= 0] = np.nan
    valid = ~np.isnan(values)
    times = times[valid]
    values = values[valid]
    if len(values) < 30:
        return {"去趋势残差标准差": "", "每小时反复升降次数": "", "是否确认": "否"}

    series = pd.Series(values)
    trend = series.rolling(TREND_WINDOW, center=True, min_periods=30).median().to_numpy()
    residual = values - trend
    window = (times >= start) & (times < end)
    residual = residual[window]
    segment_times = times[window]
    if len(residual) < 30:
        return {"去趋势残差标准差": "", "每小时反复升降次数": "", "是否确认": "否"}

    residual_std = float(np.nanstd(residual))
    prominence = max(MIN_PROMINENCE, residual_std * 0.7)
    peaks, _ = find_peaks(residual, prominence=prominence, distance=12)
    troughs, _ = find_peaks(-residual, prominence=prominence, distance=12)
    hours = max((segment_times[-1] - segment_times[0]).total_seconds() / 3600.0, 1e-9)
    reversals = float(len(peaks) + len(troughs)) / hours
    confirmed = residual_std >= MIN_RESIDUAL_STD and reversals >= MIN_REVERSALS_PER_HOUR
    row["阶段触发依据"] = (
        f"{row.get('阶段触发依据') or ''}；去趋势残差{residual_std:.2f}℃、"
        f"反复升降{reversals:.1f}次/小时"
    ).strip("；")
    return {
        "去趋势残差标准差": round(residual_std, 2),
        "每小时反复升降次数": round(reversals, 1),
        "是否确认": "是" if confirmed else "否",
    }


def annotate(hf_rows: list[dict]) -> list[dict]:
    for row in hf_rows:
        row.update(_metrics(row))
    return hf_rows


def filter_hf(hf_rows: list[dict]) -> tuple[list[dict], list[dict]]:
    """Return ``(confirmed, candidates_with_metrics)``.

    ``candidates_with_metrics`` keeps every first-pass candidate and its
    validation metrics; ``confirmed`` only keeps candidates passing the
    residual/reversal test.
    """
    candidates = annotate(hf_rows)
    confirmed = [row for row in candidates if row.get("是否确认") == "是"]
    return confirmed, candidates


__all__ = ["annotate", "filter_hf", "MIN_RESIDUAL_STD", "MIN_REVERSALS_PER_HOUR"]