Compressor/_tmp_temperature/algorithm.py

"""Cylinder-temperature feature extraction, grading and event segmentation.

The scan is deliberately rule based: it reuses the pressure/vibration skeleton
(valid running hours -> robust baseline -> relative deviation -> confirmed
levels -> merged events) but adds the signals that matter for cylinder body
temperature:

* each cylinder's own recent baseline (level + robust z);
* same-stage spread, compared with its own recent normal spread;
* each cylinder's deviation from the stage median, compared with its own
  historic offset (removes the chronic "this cylinder always runs hotter"
  bias);
* engineering high / high-high limits from ``site_point``.

Thresholds are configurable so they can be calibrated against labelled faults.
"""
from __future__ import annotations

from collections import defaultdict, deque
from datetime import datetime, timedelta
from statistics import median

from .config import (
    TEMP_BASELINE_HOURS,
    TEMP_CONFIRM_HOURS,
    TEMP_DROPOUT_REF,
    TEMP_MERGE_GAP_HOURS,
    TEMP_MIN_HISTORY_HOURS,
    TEMP_MIN_STEADY,
    TEMP_RESTART_GAP_HOURS,
    TEMP_SENSOR_MAX,
    TEMP_SENSOR_MIN,
    UnitConfig,
)


def linear_slope(values):
    pairs = [(index, value) for index, value in enumerate(values) if value is not None]
    if len(pairs) < 3:
        return None
    xbar = sum(x for x, _ in pairs) / len(pairs)
    ybar = sum(y for _, y in pairs) / len(pairs)
    denominator = sum((x - xbar) ** 2 for x, _ in pairs)
    if denominator == 0:
        return None
    return sum((x - xbar) * (y - ybar) for x, y in pairs) / denominator


def unit_sort_key(value):
    import re

    match = re.search(r"[4-9]", str(value or ""))
    return int(match.group()) if match else 999


def output_sort_key(row: dict):
    time_field = next(
        (field for field in ("小时", "开始小时", "周期开始小时", "开始日期", "日期") if row.get(field)),
        None,
    )
    time_value = str(row.get(time_field, "")) if time_field else ""
    return (unit_sort_key(row.get("机组")), time_value, int(row.get("批次") or 0), str(row.get("周期编号", "")))


def clean_rows(rows):
    cleaned = []
    for row in rows:
        item = {}
        for key, value in row.items():
            if key.startswith("_"):
                continue
            if isinstance(value, float):
                item[key] = round(value, 6)
            else:
                item[key] = value
        cleaned.append(item)
    return cleaned


def write_csv(path, rows, fields, headers):
    import csv
    from pathlib import Path

    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    rows = sorted(rows, key=output_sort_key)
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
        writer.writerow({field: headers.get(field, field) for field in fields})
        writer.writerows(rows)

LEVELS = ("轻微", "异常", "严重")


def _fmt(value, digits=2):
    return "" if value is None else round(float(value), digits)


def _median(values):
    values = [value for value in values if value is not None]
    return median(values) if values else None


def _mad(values, center):
    return median(abs(value - center) for value in values) if values else 0.0


def _baseline(values) -> tuple[float | None, float | None]:
    values = [value for value in values if value is not None]
    if len(values) < TEMP_MIN_HISTORY_HOURS:
        return None, None
    center = median(values)
    # Temperature sensors can be very stable; floor the scale at 0.5 degC and
    # 0.5% of the level so that z stays meaningful without exploding.
    scale = max(1.4826 * _mad(values, center), abs(center) * 0.005, 0.5)
    return center, scale


def _longest_constant_run(values) -> int:
    longest = current = 1
    for previous, value in zip(values, values[1:]):
        if round(previous, 3) == round(value, 3):
            current += 1
            longest = max(longest, current)
        else:
            current = 1
    return longest if values else 0


def channel_qc(rows: list[dict], config: UnitConfig) -> tuple[set[str], list[dict]]:
    """Flag frozen / dead / out-of-range cylinder channels."""
    bad: set[str] = set()
    report: list[dict] = []
    for column in config.columns:
        values = [row.get(f"{column}__avg") for row in rows if row.get(f"{column}__avg") is not None]
        label = config.labels.get(column, column)
        if not values:
            bad.add(column)
            report.append({"机组": config.name, "点位": column, "点位描述": label,
                           "运行小时数": 0, "判定": "无数据"})
            continue
        count = len(values)
        zero_ratio = sum(1 for value in values if value == 0) / count
        out_ratio = sum(1 for value in values if value < TEMP_SENSOR_MIN or value > TEMP_SENSOR_MAX) / count
        distinct = len({round(value, 3) for value in values})
        constant_run = _longest_constant_run(values)
        reasons = []
        if zero_ratio > 0.20:
            reasons.append(f"零值比例{zero_ratio:.1%}")
        if out_ratio > 0.05:
            reasons.append(f"越界比例{out_ratio:.1%}")
        if distinct <= 3 and count > 50:
            reasons.append(f"仅{distinct}个不同值")
        if constant_run > 168:
            reasons.append(f"连续{constant_run}小时不变")
        if reasons:
            bad.add(column)
        report.append({
            "机组": config.name,
            "点位": column,
            "点位描述": label,
            "运行小时数": count,
            "最小值": _fmt(min(values)),
            "最大值": _fmt(max(values)),
            "中位数": _fmt(median(values)),
            "不同值个数": distinct,
            "最长不变小时": constant_run,
            "零值比例": round(zero_ratio, 4),
            "越界比例": round(out_ratio, 4),
            "判定": "坏道" if reasons else "正常",
            "原因": "；".join(reasons),
        })
    return bad, report


def _assign_cycles(rows: list[dict], config: UnitConfig) -> list[dict]:
    """Attach a per-batch run-cycle id; gaps > 12h start a new cycle."""
    ordered = sorted(rows, key=lambda row: (row["import_batch_id"], row["hour_start"]))
    cycle = 0
    previous_batch = None
    previous_dt = None
    for row in ordered:
        batch = row["import_batch_id"]
        dt = row["hour_start"]
        if (
            previous_batch is None
            or batch != previous_batch
            or previous_dt is None
            or dt - previous_dt > timedelta(hours=TEMP_RESTART_GAP_HOURS)
        ):
            cycle += 1
        row["_cycle_id"] = f"{config.unit}-T{cycle:03d}"
        previous_batch, previous_dt = batch, dt
    return ordered


def _snapshot(row: dict, columns: list[str]) -> tuple[dict, float, float | None, dict] | None:
    raw = {}
    for column in columns:
        if row.get(f"{column}__dropout"):
            continue
        value = row.get(f"{column}__avg")
        if value is None:
            continue
        if value <= 0 or value < TEMP_SENSOR_MIN or value > TEMP_SENSOR_MAX:
            continue
        raw[column] = value
    if not raw:
        return None
    if max(raw.values()) < TEMP_MIN_STEADY:
        return None
    # When the stage is clearly hot, a channel reading ambient is a stopped or
    # dead sensor, not a real cylinder; drop it before stage statistics.
    if max(raw.values()) >= TEMP_DROPOUT_REF:
        values = {column: value for column, value in raw.items() if value >= TEMP_MIN_STEADY}
    else:
        values = raw
    if not values:
        return None
    stage_median = median(values.values())
    spread = max(values.values()) - min(values.values()) if len(values) >= 2 else None
    peer = {column: value - stage_median for column, value in values.items()}
    return values, stage_median, spread, peer


def _mark_dropouts(rows: list[dict], config: UnitConfig) -> None:
    """Mark abrupt downward channel glitches (a channel dropping alone).

    A real load cut moves all cylinders together and the whole stage falls
    below the steady floor, so it is filtered elsewhere.  A single channel that
    suddenly reads far below its own neighbours (and below 60 degC) is a sensor
    / acquisition dropout and must not inflate the same-stage spread.
    """
    for column in config.columns:
        series = []
        for index, row in enumerate(rows):
            value = row.get(f"{column}__avg")
            if value is None or value <= 0 or value < TEMP_SENSOR_MIN or value > TEMP_SENSOR_MAX:
                continue
            series.append((index, row["hour_start"], value))
        for position, (index, moment, value) in enumerate(series):
            previous = [item[2] for item in series[max(0, position - 6):position]
                        if moment - item[1] <= timedelta(hours=24)]
            following = [item[2] for item in series[position + 1:position + 7]
                         if item[1] - moment <= timedelta(hours=24)]
            reference = previous + following
            if len(reference) < 3:
                continue
            center = median(reference)
            # A transient drop of >15 degC that recovers within the local
            # window is an acquisition dropout; a persistent drop keeps the
            # reference low as well and therefore survives.
            if value < center - 15:
                rows[index][f"{column}__dropout"] = True


def _global_offsets(rows: list[dict], config: UnitConfig) -> tuple[dict, dict]:
    spreads: dict[str, list] = defaultdict(list)
    peers: dict[str, list] = defaultdict(list)
    for row in rows:
        for stage, columns in config.stages.items():
            snapshot = _snapshot(row, columns)
            if not snapshot:
                continue
            _, _, spread, peer = snapshot
            if spread is not None:
                spreads[stage].append(spread)
            for column, value in peer.items():
                peers[column].append(value)
    return (
        {stage: _median(values) for stage, values in spreads.items()},
        {column: _median(values) for column, values in peers.items()},
    )


def _grade(values: dict, metrics: dict, config: UnitConfig, thresholds: dict) -> tuple[str, list[str]]:
    """Grade one stage-hour from same-stage relative signals only.

    Engineering PVHigh/PVHighHigh are deliberately *not* used as triggers: they
    are the plant's existing absolute alarms, and a hot summer day pushes every
    cylinder above them.  The added value here is the same-stage difference.
    """
    reasons: list[str] = []
    severe = abnormal = mild = False

    spread_delta = metrics.get("spread_delta")
    if spread_delta is not None:
        if spread_delta >= thresholds["severe_spread"]:
            severe = True
            reasons.append(f"级差较平时+{spread_delta:.1f}℃")
        elif spread_delta >= thresholds["abnormal_spread"]:
            abnormal = True
            reasons.append(f"级差较平时+{spread_delta:.1f}℃")
        elif spread_delta >= thresholds["mild_spread"]:
            mild = True
            reasons.append(f"级差较平时+{spread_delta:.1f}℃")

    peer_high = metrics.get("max_peer_anom")
    if peer_high is not None:
        if peer_high >= thresholds["severe_peer"]:
            severe = True
            reasons.append(f"同级偏高较平时+{peer_high:.1f}℃")
        elif peer_high >= thresholds["abnormal_peer"]:
            abnormal = True
            reasons.append(f"同级偏高较平时+{peer_high:.1f}℃")
        elif peer_high >= thresholds["mild_peer"]:
            mild = True
            reasons.append(f"同级偏高较平时+{peer_high:.1f}℃")

    peer_low = metrics.get("max_peer_low")
    if peer_low is not None:
        if peer_low >= thresholds["severe_peer_low"]:
            severe = True
            reasons.append(f"同级偏低较平时-{peer_low:.1f}℃")
        elif peer_low >= thresholds["abnormal_peer_low"]:
            abnormal = True
            reasons.append(f"同级偏低较平时-{peer_low:.1f}℃")
        elif peer_low >= thresholds["mild_peer_low"]:
            mild = True
            reasons.append(f"同级偏低较平时-{peer_low:.1f}℃")

    # Own-baseline level/z is load-driven, so it is reported but not used as a
    # trigger; a single-cylinder stage (7#/8#/9# 三级) has no same-stage peer and
    # therefore stays quiet unless the same-stage logic above can fire.
    if severe:
        return "严重", reasons
    if abnormal:
        return "异常", reasons
    if mild:
        return "轻微", reasons
    return "正常", reasons


def compute_features(
    rows: list[dict],
    config: UnitConfig,
    bad_channels: set[str],
    thresholds: dict | None = None,
) -> list[dict]:
    thresholds = {**config.thresholds, **(thresholds or {})}
    valid = [row for row in _assign_cycles(rows, config) if row.get("valid")]
    _mark_dropouts(valid, config)
    global_spread, global_peer = _global_offsets(valid, config)

    history: dict[str, deque] = {column: deque() for column in config.columns}
    history_peer: dict[str, deque] = {column: deque() for column in config.columns}
    history_spread: dict[str, deque] = {stage: deque() for stage in config.stages}
    rise_count: dict[str, int] = {column: 0 for column in config.columns}
    last_dt: dict[str, datetime | None] = {column: None for column in config.columns}
    last_stage_dt: dict[str, datetime | None] = {stage: None for stage in config.stages}

    features: list[dict] = []
    for row in valid:
        dt: datetime = row["hour_start"]
        window_start = dt - timedelta(hours=TEMP_BASELINE_HOURS)
        for column, values in history.items():
            while values and values[0][0] < window_start:
                values.popleft()
        for column, values in history_peer.items():
            while values and values[0][0] < window_start:
                values.popleft()
        for values in history_spread.values():
            while values and values[0][0] < window_start:
                values.popleft()

        for stage, columns in config.stages.items():
            usable = [column for column in columns if column not in bad_channels]
            snapshot = _snapshot(row, usable)
            if not snapshot:
                continue
            values, stage_median, spread, peer = snapshot

            spread_values = [value for _, value in history_spread[stage]]
            spread_base = _median(spread_values) if len(spread_values) >= TEMP_MIN_HISTORY_HOURS else global_spread.get(stage)
            spread_delta = None
            if spread is not None and spread_base is not None:
                spread_delta = spread - spread_base

            per_column: dict[str, dict] = {}
            for column, value in values.items():
                own_values = [item for _, item in history[column]]
                # Daily load cycles move every cylinder together.  A time-of-day
                # baseline removes that common ramp, which is essential for a
                # single-cylinder stage (7#/8#/9# 三级) that has no same-stage peer.
                tod_values = [item for moment, item in history[column] if moment.hour == dt.hour]
                if len(tod_values) >= 3:
                    base = median(tod_values)
                    scale = max(1.4826 * _mad(tod_values, base), abs(base) * 0.005, 0.5)
                else:
                    base, scale = _baseline(own_values)
                delta = value - base if base is not None else None
                z = delta / scale if (delta is not None and scale) else None

                peer_values = [item for _, item in history_peer[column]]
                peer_base = _median(peer_values) if len(peer_values) >= TEMP_MIN_HISTORY_HOURS else global_peer.get(column, 0.0)
                peer_dev = peer[column]
                peer_anom = peer_dev - peer_base if peer_base is not None else None

                series = own_values[-5:] + [value]
                slope = linear_slope(series) if len(series) >= 3 else None

                if delta is not None and delta > 1.0:
                    rise_count[column] = rise_count[column] + 1 if last_dt[column] and dt - last_dt[column] == timedelta(hours=1) else 1
                else:
                    rise_count[column] = 0

                per_column[column] = {
                    "value": value, "base": base, "delta": delta, "z": z,
                    "peer_dev": peer_dev, "peer_anom": peer_anom,
                    "slope": slope, "rise": rise_count[column],
                }

                history[column].append((dt, value))
                history_peer[column].append((dt, peer_dev))
                last_dt[column] = dt

            if spread is not None:
                history_spread[stage].append((dt, spread))

            deltas = [item["delta"] for item in per_column.values() if item["delta"] is not None]
            zs = [item["z"] for item in per_column.values() if item["z"] is not None]
            peer_anoms = [item["peer_anom"] for item in per_column.values() if item["peer_anom"] is not None]
            insufficient = not deltas

            max_delta = max(deltas) if deltas else None
            max_z = max(zs) if zs else None
            max_peer_high = max(peer_anoms) if peer_anoms else None
            max_peer_low = abs(min(peer_anoms)) if peer_anoms else None
            strongest = max(
                (column for column in per_column if per_column[column]["delta"] is not None),
                key=lambda column: per_column[column]["delta"],
                default=None,
            )
            peak_peer_column = max(
                (column for column in per_column if per_column[column]["peer_anom"] is not None),
                key=lambda column: per_column[column]["peer_anom"],
                default=None,
            )
            low_peer_column = min(
                (column for column in per_column if per_column[column]["peer_anom"] is not None),
                key=lambda column: per_column[column]["peer_anom"],
                default=None,
            )
            slope = per_column[strongest]["slope"] if strongest else None
            rise = per_column[strongest]["rise"] if strongest else None

            high_values = [config.limits.get(column, {}).get("high") for column in values]
            high_values = [value for value in high_values if value is not None]
            high_high_values = [config.limits.get(column, {}).get("high_high") for column in values]
            high_high_values = [value for value in high_high_values if value is not None]

            metrics = {
                "spread": spread, "spread_base": spread_base, "spread_delta": spread_delta,
                "max_delta": max_delta, "max_z": max_z, "max_peer_anom": max_peer_high,
                "slope": slope,
            }
            if insufficient:
                raw = "数据不足"
                reasons = ["有效基线不足24小时"]
            else:
                raw, reasons = _grade(values, metrics, config, thresholds)

            features.append({
                "机组": config.name,
                "批次": row["import_batch_id"],
                "周期编号": row["_cycle_id"],
                "小时": dt.strftime("%Y-%m-%d %H:%M:%S"),
                "级": stage,
                "有效缸数": len(values),
                "运行覆盖率": round(row.get("coverage") or 0.0, 4),
                "最高温度": _fmt(max(values.values())),
                "最低温度": _fmt(min(values.values())),
                "最高温度缸": config.labels.get(max(values, key=values.get), ""),
                "级差": _fmt(spread),
                "级差基线": _fmt(spread_base),
                "级差变化": _fmt(spread_delta),
                "最大自身偏离": _fmt(max_delta),
                "最大自身偏离缸": config.labels.get(strongest, "") if strongest else "",
                "最大同级偏离": _fmt(max_peer_high),
                "最大同级偏离缸": config.labels.get(peak_peer_column, "") if peak_peer_column else "",
                "最大同级偏低": _fmt(max_peer_low),
                "最大同级偏低缸": config.labels.get(low_peer_column, "") if low_peer_column else "",
                "最大稳健z": _fmt(max_z),
                "最强缸": config.labels.get(strongest, "") if strongest else "",
                "最高报警阈值": _fmt(min(high_values)) if high_values else "",
                "高高报警阈值": _fmt(min(high_high_values)) if high_high_values else "",
                "超告警缸": "、".join(
                    config.labels.get(column, column) for column in values
                    if (config.limits.get(column, {}).get("high") is not None
                        and values[column] >= config.limits[column]["high"])
                ),
                "超高高缸": "、".join(
                    config.labels.get(column, column) for column in values
                    if (config.limits.get(column, {}).get("high_high") is not None
                        and values[column] >= config.limits[column]["high_high"])
                ),
                "近6小时斜率": _fmt(slope, 4),
                "连续上升小时数": rise or 0,
                "各缸温度": "；".join(f"{config.labels.get(column, column)}={value:.2f}" for column, value in values.items()),
                "各缸自身偏离": "；".join(f"{config.labels.get(column, column)}={per_column[column]['delta']:+.2f}"
                                          for column in values if per_column[column]["delta"] is not None),
                "各缸同级偏离": "；".join(f"{config.labels.get(column, column)}={per_column[column]['peer_dev']:+.2f}"
                                          for column in values),
                "原始等级": raw,
                "判定依据": "；".join(reasons),
                "_dt": dt,
                "_strongest": strongest,
            })

    _confirm_levels(features, thresholds)
    return features


def _confirm_levels(features: list[dict], thresholds: dict) -> None:
    confirm_hours = int(thresholds.get("confirm_hours", TEMP_CONFIRM_HOURS))
    grouped: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        grouped[(row["机组"], row["批次"], row["级"])].append(row)

    order = {"轻微": 1, "异常": 2, "严重": 3}
    for rows in grouped.values():
        rows.sort(key=lambda row: row["_dt"])
        index = 0
        while index < len(rows):
            raw = rows[index]["原始等级"]
            if raw in ("正常", "数据不足"):
                rows[index]["温度等级"] = raw
                index += 1
                continue
            end = index
            while end + 1 < len(rows):
                candidate = rows[end + 1]["原始等级"]
                if candidate in ("正常", "数据不足"):
                    break
                if rows[end + 1]["_dt"] - rows[end]["_dt"] != timedelta(hours=1):
                    break
                end += 1
            run = rows[index:end + 1]
            run_length = len(run)
            highest = max((row["原始等级"] for row in run), key=lambda value: order[value])
            if run_length >= confirm_hours:
                final = highest
            elif highest == "严重":
                final = "严重" if any(row["超高高缸"] for row in run) else "异常"
            else:
                final = "轻微"
            for row in run:
                row["温度等级"] = final
            index = end + 1


def build_segments(features: list[dict], config: UnitConfig, min_hours: int = 2) -> list[dict]:
    grouped: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        if row["温度等级"] in LEVELS:
            grouped[(row["机组"], row["批次"], row["级"], row["温度等级"])].append(row)

    segments: list[dict] = []
    for (unit, batch, stage, level), rows in grouped.items():
        rows.sort(key=lambda row: row["_dt"])
        current: list[dict] = []
        for row in rows:
            if not current or row["_dt"] - current[-1]["_dt"] <= timedelta(hours=TEMP_MERGE_GAP_HOURS):
                current.append(row)
            else:
                segments.append(_build_segment(current, unit, batch, stage, level, config))
                current = [row]
        if current:
            segments.append(_build_segment(current, unit, batch, stage, level, config))
    segments = [segment for segment in segments if segment["有效运行小时数"] >= min_hours]
    return sorted(segments, key=output_sort_key)


def _build_segment(rows: list[dict], unit: str, batch: int, stage: str, level: str, config: UnitConfig) -> dict:
    first, last = rows[0], rows[-1]
    strongest = [row for row in rows if row["最强缸"]]
    peak_row = max(rows, key=lambda row: row["最大自身偏离"] or -1e9)
    spread_rows = [row for row in rows if row["级差"] not in ("", None)]
    peak_spread_row = max(spread_rows, key=lambda row: row["级差"]) if spread_rows else None
    peer_rows = [row for row in rows if row["最大同级偏离"] not in ("", None)]
    peak_peer_row = max(peer_rows, key=lambda row: row["最大同级偏离"]) if peer_rows else None
    z_rows = [row for row in rows if row["最大稳健z"] not in ("", None)]
    peak_z_row = max(z_rows, key=lambda row: row["最大稳健z"]) if z_rows else None

    return {
        "机组": unit,
        "批次": batch,
        "周期编号": first["周期编号"],
        "级": stage,
        "温度阶段": level,
        "开始小时": first["小时"],
        "结束小时": last["小时"],
        "持续自然小时数": int((last["_dt"] - first["_dt"]).total_seconds() / 3600) + 1,
        "有效运行小时数": len(rows),
        "峰值时刻": peak_row["小时"],
        "峰值缸": peak_row["最强缸"],
        "最高温度": max(row["最高温度"] for row in rows if row["最高温度"] not in ("", None)),
        "最大自身偏离": peak_row["最大自身偏离"],
        "最大级差": peak_spread_row["级差"] if peak_spread_row else "",
        "最大级差基线": peak_spread_row["级差基线"] if peak_spread_row else "",
        "最大级差变化": peak_spread_row["级差变化"] if peak_spread_row else "",
        "最大同级偏离": peak_peer_row["最大同级偏离"] if peak_peer_row else "",
        "最大稳健z": peak_z_row["最大稳健z"] if peak_z_row else "",
        "阶段最高报警阈值": min((row["最高报警阈值"] for row in rows if row["最高报警阈值"] not in ("", None)), default=""),
        "阶段高高报警阈值": min((row["高高报警阈值"] for row in rows if row["高高报警阈值"] not in ("", None)), default=""),
        "阶段超告警缸": "、".join(sorted({item for row in rows for item in str(row["超告警缸"]).split("、") if item})),
        "阶段超高高缸": "、".join(sorted({item for row in rows for item in str(row["超高高缸"]).split("、") if item})),
        "最强缸": max(set(row["最强缸"] for row in strongest), key=lambda value: sum(1 for row in strongest if row["最强缸"] == value)) if strongest else "",
        "阶段触发依据": "；".join(row["判定依据"] for row in rows[:3] if row["判定依据"]),
        "_dt": first["_dt"],
    }


def summarize_cycles(features: list[dict], segments: list[dict], config: UnitConfig) -> list[dict]:
    by_cycle: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        by_cycle[(row["机组"], row["批次"], row["周期编号"], row["级"])].append(row)

    result: list[dict] = []
    for (unit, batch, cycle_id, stage), rows in by_cycle.items():
        rows.sort(key=lambda row: row["_dt"])
        stage_segments = [segment for segment in segments
                          if segment["周期编号"] == cycle_id and segment["级"] == stage]
        summary = {
            "机组": unit,
            "批次": batch,
            "周期编号": cycle_id,
            "级": stage,
            "周期开始小时": rows[0]["小时"],
            "周期结束小时": rows[-1]["小时"],
            "有效运行小时数": len(rows),
            "阶段演化顺序": "→".join(segment["温度阶段"] for segment in stage_segments) or "正常",
        }
        for level in LEVELS:
            selected = [segment for segment in stage_segments if segment["温度阶段"] == level]
            summary[f"{level}阶段开始小时"] = selected[0]["开始小时"] if selected else ""
            summary[f"{level}阶段结束小时"] = selected[-1]["结束小时"] if selected else ""
            summary[f"{level}阶段运行小时数"] = sum(segment["有效运行小时数"] for segment in selected)
        summary["周期最高温度"] = max((row["最高温度"] for row in rows if row["最高温度"] not in ("", None)), default="")
        summary["周期最大级差"] = max((row["级差"] for row in rows if row["级差"] not in ("", None)), default="")
        summary["周期最大级差变化"] = max((row["级差变化"] for row in rows if row["级差变化"] not in ("", None)), default="")
        summary["周期最大自身偏离"] = max((row["最大自身偏离"] for row in rows if row["最大自身偏离"] not in ("", None)), default="")
        result.append(summary)
    return result


def run(rows: list[dict], config: UnitConfig, thresholds: dict | None = None, min_hours: int = 2):
    valid = [row for row in rows if row.get("valid")]
    bad_channels, qc_report = channel_qc(valid, config)
    features = compute_features(rows, config, bad_channels, thresholds)
    segments = build_segments(features, config, min_hours)
    cycles = summarize_cycles(features, segments, config)
    return bad_channels, qc_report, features, segments, cycles


FEATURE_FIELDS = [
    "机组", "批次", "周期编号", "小时", "级", "有效缸数", "运行覆盖率",
    "最高温度", "最低温度", "最高温度缸", "最高报警阈值", "高高报警阈值", "超告警缸", "超高高缸",
    "级差", "级差基线", "级差变化",
    "最大自身偏离", "最大自身偏离缸", "最大同级偏离", "最大同级偏离缸", "最大同级偏低", "最大同级偏低缸", "最大稳健z", "最强缸",
    "近6小时斜率", "连续上升小时数",
    "各缸温度", "各缸自身偏离", "各缸同级偏离",
    "温度等级", "原始等级", "判定依据",
]

SEGMENT_FIELDS = [
    "机组", "批次", "周期编号", "级", "温度阶段", "开始小时", "结束小时",
    "持续自然小时数", "有效运行小时数", "峰值时刻", "峰值缸", "最高温度",
    "最大自身偏离", "最大级差", "最大级差基线", "最大级差变化", "最大同级偏离", "最大稳健z",
    "阶段最高报警阈值", "阶段高高报警阈值", "阶段超告警缸", "阶段超高高缸", "最强缸", "阶段触发依据",
]

CYCLE_FIELDS = [
    "机组", "批次", "周期编号", "级", "周期开始小时", "周期结束小时", "有效运行小时数", "阶段演化顺序",
    "轻微阶段开始小时", "轻微阶段结束小时", "轻微阶段运行小时数",
    "异常阶段开始小时", "异常阶段结束小时", "异常阶段运行小时数",
    "严重阶段开始小时", "严重阶段结束小时", "严重阶段运行小时数",
    "周期最高温度", "周期最大级差", "周期最大级差变化", "周期最大自身偏离",
]

QC_FIELDS = [
    "机组", "点位", "点位描述", "运行小时数", "最小值", "最大值", "中位数",
    "不同值个数", "最长不变小时", "零值比例", "越界比例", "判定", "原因",
]


def write_results(output_dir, qc_report, features, segments, cycles):
    output_dir.mkdir(parents=True, exist_ok=True)
    write_csv(output_dir / "温度小时特征.csv", clean_rows(features), FEATURE_FIELDS, {field: field for field in FEATURE_FIELDS})
    write_csv(output_dir / "温度异常段.csv", clean_rows(segments), SEGMENT_FIELDS, {field: field for field in SEGMENT_FIELDS})
    write_csv(output_dir / "温度演化周期.csv", clean_rows(cycles), CYCLE_FIELDS, {field: field for field in CYCLE_FIELDS})
    write_csv(output_dir / "温度通道质控.csv", clean_rows(qc_report), QC_FIELDS, {field: field for field in QC_FIELDS})


__all__ = [
    "LEVELS", "channel_qc", "compute_features", "build_segments", "summarize_cycles", "run",
    "write_results", "FEATURE_FIELDS", "SEGMENT_FIELDS", "CYCLE_FIELDS", "QC_FIELDS",
]