Compressor/backend/app/services/temperature_transition_service.py

"""Temperature event waveform data for the review page.

The labels come from ``_tmp_temperature/out/温度异常段.csv``; the waveform is
queried live from ``pks_long_sample`` at 5-second resolution.
"""
from __future__ import annotations

import csv
import math
import re
from datetime import datetime, timedelta
from pathlib import Path
from statistics import median
from typing import Any

from ..db import get_connection

ROOT = Path(__file__).resolve().parents[3]
TEMPERATURE_EVENTS_CSV = ROOT / "_tmp_temperature" / "out" / "温度异常段.csv"
TEMPERATURE_OBS_CSV = ROOT / "_tmp_temperature" / "out" / "温度观察段.csv"
STEP_SECONDS = 5
PAD_HOURS = 6

CYLINDERS: dict[int, dict[str, list[tuple[str, str]]]] = {
    4: {
        "一级": [("YSJ_9", "一级1#"), ("YSJ_10", "一级2#"), ("YSJ_11", "一级3#")],
        "二级": [("YSJ_13", "二级1#"), ("YSJ_14", "二级2#"), ("YSJ_15", "二级3#")],
    },
    5: {
        "一级": [("YSJ_18", "一级1#"), ("YSJ_20", "一级2#"), ("YSJ_22", "一级3#")],
        "二级": [("YSJ_17", "二级1#"), ("YSJ_19", "二级2#"), ("YSJ_21", "二级3#")],
    },
    6: {
        "一级": [("YSJ_18", "一级1#"), ("YSJ_20", "一级2#"), ("YSJ_22", "一级3#")],
        "二级": [("YSJ_17", "二级1#"), ("YSJ_19", "二级2#"), ("YSJ_21", "二级3#")],
    },
    7: {
        "一级": [("YSJ_16", "一级2列"), ("YSJ_17", "一级4列"), ("YSJ_18", "一级6列")],
        "二级": [("YSJ_19", "二级3列"), ("YSJ_20", "二级5列")],
        "三级": [("YSJ_21", "三级1列")],
    },
    8: {
        "一级": [("YSJ_16", "一级2列"), ("YSJ_17", "一级4列"), ("YSJ_18", "一级6列")],
        "二级": [("YSJ_19", "二级3列"), ("YSJ_20", "二级5列")],
        "三级": [("YSJ_21", "三级1列")],
    },
    9: {
        "一级": [("YSJ_16", "一级2列"), ("YSJ_17", "一级4列"), ("YSJ_18", "一级6列")],
        "二级": [("YSJ_19", "二级3列"), ("YSJ_20", "二级5列")],
        "三级": [("YSJ_21", "三级1列")],
    },
}

COLORS = ["#1f6feb", "#e8890c", "#2e9e5b", "#8e44ad", "#c0392b", "#16a085"]
BAND = "rgba(255,193,7,0.16)"


def _to_float(value: Any) -> float | None:
    if value is None:
        return None
    text = str(value).strip()
    if not text or text.lower() == "nan":
        return None
    try:
        number = float(text)
    except ValueError:
        return None
    return number if math.isfinite(number) else None


def _temp_value(value: Any) -> float | None:
    """Cylinder temperature: treat 0 as a missing/dropout sample for plotting."""
    number = _to_float(value)
    if number is None or number == 0:
        return None
    return number


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


def _fmt_time(value: datetime) -> str:
    return value.strftime("%Y-%m-%d %H:%M:%S")


def _unit_number(value: str) -> int:
    match = re.search(r"(\d+)", value or "")
    return int(match.group(1)) if match else 4


def _padded(values: list[float]) -> tuple[float, float]:
    if not values:
        return 0.0, 1.0
    low, high = min(values), max(values)
    if high - low < 1e-9:
        return low - 1.0, high + 1.0
    span = high - low
    return low - span * 0.08, high + span * 0.08


def _limits(unit: int, columns: list[str]) -> dict[str, dict[str, float | None]]:
    result: dict[str, dict[str, float | None]] = {
        column: {"high": None, "high_high": None} for column in columns
    }
    with get_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT ItemName, AlarmType1, AlarmLimit1, AlarmType2, AlarmLimit2, "
                "AlarmType3, AlarmLimit3, AlarmType4, AlarmLimit4 "
                "FROM site_point WHERE ItemName LIKE %s",
                (f"YSJ{unit}\\_%",),
            )
            for row in cursor.fetchall():
                name = row.get("ItemName") or ""
                index = name.split("_")[-1]
                column = f"YSJ_{index}"
                if column not in result:
                    continue
                high = high_high = None
                for item in range(1, 5):
                    kind = str(row.get(f"AlarmType{item}") or "")
                    limit = _to_float(row.get(f"AlarmLimit{item}"))
                    if not kind or limit is None:
                        continue
                    if "HighHigh" in kind:
                        high_high = limit if high_high is None else max(high_high, limit)
                    elif "High" in kind and high is None:
                        high = limit
                result[column] = {"high": high, "high_high": high_high}
    return result


def _read_csv(path: Path) -> list[dict[str, str]]:
    if not path.exists():
        return []
    with path.open(encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))


def _load_faults() -> list[dict[str, Any]]:
    with get_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT fault_date, fault_category, fault_description, unit_name "
                "FROM compressor_fault ORDER BY fault_date"
            )
            return list(cursor.fetchall())


def _fault_matches(unit: str, start: datetime, end: datetime, faults: list[dict[str, Any]]) -> list[str]:
    unit_name = unit.replace("号机", "号机组")
    low = (start - timedelta(days=7)).date()
    high = (end + timedelta(days=1)).date()
    matches = []
    for row in faults:
        if (row.get("unit_name") or "") != unit_name:
            continue
        date = row.get("fault_date")
        if date is None or date < low or date > high:
            continue
        text = row.get("fault_description") or row.get("fault_category") or ""
        matches.append(f"{date} {text}")
    return matches


def _event_matches(unit: str, stage: str, start: datetime, end: datetime,
                   event_rows: list[dict[str, str]]) -> list[str]:
    matches = []
    for row in event_rows:
        if (row.get("特征类型") or "") == "高频波动":
            continue
        if (row.get("机组") or "") != unit or (row.get("级") or "") != stage:
            continue
        try:
            event_start = _parse_time(row.get("开始小时") or "")
            event_end = _parse_time(row.get("结束小时") or "") + timedelta(hours=1)
        except ValueError:
            continue
        if event_start <= end + timedelta(hours=1) and event_end >= start - timedelta(hours=1):
            matches.append(
                f"{event_start.strftime('%m-%d %H:%M')} "
                f"{row.get('温度阶段') or ''}-{row.get('特征类型') or ''}"
            )
    return matches


def _observation_matches(unit: str, stage: str, start: datetime, end: datetime,
                         observation_rows: list[dict[str, str]]) -> list[str]:
    matches = []
    for row in observation_rows:
        if (row.get("机组") or "") != unit or (row.get("级") or "") != stage:
            continue
        try:
            moment = _parse_time(row.get("小时") or "")
        except ValueError:
            continue
        if start - timedelta(hours=1) <= moment <= end + timedelta(hours=1):
            matches.append(
                f"{moment.strftime('%m-%d %H:%M')} "
                f"{row.get('方向') or ''}-{row.get('特征类型') or ''}"
            )
    return matches


def _labels() -> list[dict[str, Any]]:
    event_rows = _read_csv(TEMPERATURE_EVENTS_CSV)
    observation_rows = _read_csv(TEMPERATURE_OBS_CSV)
    faults = _load_faults()

    items: list[dict[str, Any]] = []
    for index, row in enumerate(event_rows):
        unit = (row.get("机组") or "").strip()
        start_text = row.get("开始小时") or ""
        end_text = row.get("结束小时") or ""
        level = row.get("温度阶段") or ""
        feature = row.get("特征类型") or ""
        grade = row.get("级") or ""
        is_hf = feature == "高频波动"
        items.append({
            "key": f"event:{index}",
            "title": f"{unit}-{grade}-{level}-{feature}-{start_text[:16]}",
            "kind": "temperature_hf" if is_hf else "temperature_event",
            "kindLabel": "高频波动" if is_hf else "温度事件",
            "device": unit,
            "deviceCode": f"{_unit_number(unit)}#",
            "batch": row.get("批次") or "",
            "cycle": row.get("周期编号") or "",
            "grade": grade,
            "stage": level,
            "transition": "",
            "level": level,
            "feature": feature,
            "direction": row.get("方向") or "",
            "start": start_text,
            "end": end_text,
            "duration": _to_float(row.get("持续自然小时数")),
            "maxSpread": None if is_hf else _to_float(row.get("最大级差变化")),
            "maxPeer": None if is_hf else _to_float(row.get("最大同级偏离")),
            "maxPeerLow": None if is_hf else _to_float(row.get("最大同级偏低")),
            "relatedChannel": row.get("关联通道异常") or "",
            "evidence": row.get("阶段触发依据") or "",
            "oldEvent": "高频波动" if is_hf else f"{level}-{feature}",
            "note": "",
        })

    for item in items:
        try:
            start = _parse_time(item["start"])
            end = _parse_time(item["end"])
        except ValueError:
            start = end = datetime.min
        unit = item["device"]
        grade = item["grade"]
        matches_faults = _fault_matches(unit, start, end, faults) if start != datetime.min else []
        matches_events = _event_matches(unit, grade, start, end, event_rows) if start != datetime.min else []
        matches_observations = _observation_matches(unit, grade, start, end, observation_rows) if start != datetime.min else []
        item.update({
            "faultHit": "是" if matches_faults else "否",
            "faultDescriptions": "；".join(matches_faults),
            "eventHit": "是" if matches_events else "否",
            "eventDescriptions": "；".join(matches_events),
            "observationHit": "是" if matches_observations else "否",
            "observationDescriptions": "；".join(matches_observations),
        })

    items.sort(key=lambda item: item.get("start") or "")
    return items


def _summary(items: list[dict[str, Any]]) -> dict[str, Any]:
    def count(key: str) -> dict[str, int]:
        result: dict[str, int] = {}
        for item in items:
            name = str(item.get(key) or "")
            if not name:
                continue
            result[name] = result.get(name, 0) + 1
        return dict(sorted(result.items()))

    return {
        "total": len(items),
        "byKind": count("kindLabel"),
        "byDevice": count("device"),
        "byTransition": count("transition"),
        "byGrade": count("grade"),
        "byLevel": count("level"),
    }


def _query(unit: int, batch: int, start: datetime, end: datetime) -> list[dict[str, Any]]:
    stages = CYLINDERS[unit]
    temp_columns = [column for values in stages.values() for column, _ in values]
    proxies = {
        4: ["YSJ_31", "YSJ_37", "YSJ_38", "YSJ_2"],
        5: ["YSJ_5", "YSJ_6", "YSJ_3", "YSJ_10", "YSJ_11", "YSJ_12", "YSJ_13", "YSJ_14"],
        6: ["YSJ_5", "YSJ_6", "YSJ_3", "YSJ_10", "YSJ_11", "YSJ_12", "YSJ_13", "YSJ_14"],
        7: ["YSJ_41"],
        8: ["YSJ_41"],
        9: ["YSJ_41"],
    }[unit]
    columns = list(dict.fromkeys(temp_columns + proxies))
    expression = ",\n      ".join(f"AVG(`{column}`) AS `{column}`" for column in columns)
    sql = (
        "SELECT FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV %s) * %s) AS t,\n      "
        + expression
        + "\nFROM pks_long_sample\n"
        "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time <= %s\n"
        "GROUP BY FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV %s) * %s)\n"
        "ORDER BY t"
    )
    with get_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                sql,
                (STEP_SECONDS, STEP_SECONDS, batch, start, end, STEP_SECONDS, STEP_SECONDS),
            )
            return cursor.fetchall()


def _series_points(values: list[float | None]) -> list[list[Any]]:
    return [[index, value] for index, value in enumerate(values)]


def _subplot(title: str, unit: str, series: list[dict[str, Any]], ref_lines: list[dict[str, Any]],
             y_min: float, y_max: float) -> dict[str, Any]:
    return {
        "title": title,
        "unit": unit,
        "band": BAND,
        "refLines": ref_lines,
        "yMin": round(y_min, 4),
        "yMax": round(y_max, 4),
        "series": series,
    }


def _build_chart(label: dict[str, Any]) -> dict[str, Any]:
    unit = _unit_number(label["device"])
    batch = int(label.get("batch") or 0)
    start = _parse_time(label["start"])
    end = _parse_time(label["end"]) + timedelta(hours=1)
    range_start = start - timedelta(hours=PAD_HOURS)
    range_end = end + timedelta(hours=PAD_HOURS)
    rows = _query(unit, batch, range_start, range_end)

    if not rows:
        return {
            "key": label["key"], "title": label["title"],
            "subtitle": f"{label.get('kindLabel') or '标签'} {_fmt_time(start)} ~ {_fmt_time(end)} ｜ 无数据",
            "stage": label["stage"], "kind": label["kind"],
            "x": {"max": 0, "times": [], "gaps": [], "shutdowns": []},
            "fault": {"start": 0, "end": 0}, "yellowBands": [], "subplots": [], "empty": True,
            "relatedChannel": label.get("relatedChannel") or "",
            "faultHit": label.get("faultHit") or "否",
            "faultDescriptions": label.get("faultDescriptions") or "",
            "evidence": label.get("evidence") or "",
            "breakdown": {},
        }

    # Insert visual gaps for long stops; keep 5s resolution otherwise.
    display: list[dict[str, Any] | None] = []
    gap_indexes: list[int] = []
    shutdowns: list[dict[str, Any]] = []
    previous = None
    gap_seconds = max(90, STEP_SECONDS * 3)
    gap_slots = 4
    for row in rows:
        moment = row["t"]
        if previous is not None and (moment - previous).total_seconds() > gap_seconds:
            left = len(display)
            display.extend([None] * gap_slots)
            right = len(display) - 1
            gap_indexes.append((left + right) // 2)
            shutdowns.append({
                "shutdown": previous.strftime("%Y-%m-%d %H:%M:%S"),
                "startup": moment.strftime("%Y-%m-%d %H:%M:%S"),
                "startIndex": left,
                "endIndex": right,
            })
        display.append(row)
        previous = moment

    times = [item["t"].strftime("%Y-%m-%d %H:%M:%S") if item else None for item in display]

    # Stage values and same-stage metrics at each timestamp.
    stages = CYLINDERS[unit]
    stage_values: dict[str, list[list[float | None]]] = {}
    for stage, columns in stages.items():
        stage_values[stage] = [
            [_temp_value(item.get(column)) if item else None for column, _ in columns]
            for item in display
        ]

    spread: list[float | None] = []
    peer_high: list[float | None] = []
    peer_low: list[float | None] = []
    for index, item in enumerate(display):
        if item is None:
            spread.append(None)
            peer_high.append(None)
            peer_low.append(None)
            continue
        values = [
            value
            for stage in stages
            for value in stage_values[stage][index]
            if value is not None
        ]
        if not values:
            spread.append(None)
            peer_high.append(None)
            peer_low.append(None)
            continue
        spread.append(max(values) - min(values))
        highs, lows = [], []
        for stage in stages:
            stage_vals = [value for value in stage_values[stage][index] if value is not None]
            if len(stage_vals) < 2:
                continue
            center = median(stage_vals)
            highs.append(max(stage_vals) - center)
            lows.append(center - min(stage_vals))
        peer_high.append(max(highs) if highs else None)
        peer_low.append(max(lows) if lows else None)

    limits = _limits(unit, [column for values in stages.values() for column, _ in values])

    subplots: list[dict[str, Any]] = []
    for stage_index, (stage, columns) in enumerate(stages.items()):
        series = []
        all_values = []
        for index, (column, name) in enumerate(columns):
            values = [
                stage_values[stage][point][index] if point is not None else None
                for point in range(len(display))
            ]
            all_values.extend(value for value in values if value is not None)
            series.append({
                "name": name,
                "color": COLORS[index % len(COLORS)],
                "data": _series_points(values),
            })
        limits_for_stage = [limits[column] for column, _ in columns]
        high_values = [item["high"] for item in limits_for_stage if item["high"] is not None]
        high_high_values = [item["high_high"] for item in limits_for_stage if item["high_high"] is not None]
        refs = []
        if high_values:
            refs.append({"value": min(high_values), "label": "高报"})
        if high_high_values:
            refs.append({"value": min(high_high_values), "label": "高高报"})
        low, high = _padded(all_values)
        subplots.append(_subplot(
            f"{stage}缸排气温度", "℃", series, refs,
            min(low, 0 if low < 0 else low), high,
        ))

    # Same-stage metrics.
    metric_values = [value for value in spread + peer_high + peer_low if value is not None]
    low, high = _padded(metric_values)
    subplots.append(_subplot(
        "同级温差指标", "℃",
        [
            {"name": "级差", "color": "#1f6feb", "data": _series_points(spread)},
            {"name": "同级偏高", "color": "#e8890c", "data": _series_points(peer_high)},
            {"name": "同级偏低", "color": "#2e9e5b", "data": _series_points(peer_low)},
        ],
        [{"value": 4, "label": "轻微"}, {"value": 8, "label": "异常"}, {"value": 15, "label": "严重异常"}],
        low, max(high, 16),
    ))

    # Run-state / operating signals.
    def column_values(column: str) -> list[float | None]:
        return [_to_float(item.get(column)) if item else None for item in display]

    aux_series: list[tuple[str, list[float | None], str]] = []
    if unit == 4:
        body = []
        for first, second in zip(column_values("YSJ_37"), column_values("YSJ_38")):
            candidates = [value for value in (first, second) if value is not None]
            body.append(max(candidates) if candidates else None)
        inlet = column_values("YSJ_2")
        run_state = []
        for body_value, inlet_value in zip(body, inlet):
            candidates = [value for value in (body_value, inlet_value) if value is not None]
            run_state.append((1.0 if max(candidates) > 0.5 else 0.0) if candidates else None)
        aux_series = [
            ("运行状态", run_state, "#8e44ad"),
            ("机身振动", body, "#e8890c"),
            ("一级进气压力", inlet, "#2e9e5b"),
        ]
    elif unit in (5, 6):
        body = []
        for values in zip(*(column_values(column) for column in ("YSJ_10", "YSJ_11", "YSJ_12", "YSJ_13"))):
            candidates = [value for value in values if value is not None]
            body.append(max(candidates) if candidates else None)
        for title, columns, colors in [
            ("机身振动", [("YSJ_10", "机身1"), ("YSJ_11", "机身2"), ("YSJ_12", "机身3"), ("YSJ_13", "机身4")], COLORS),
            ("排气压力", [("YSJ_5", "一级排气"), ("YSJ_6", "二级排气")], ["#1f6feb", "#e8890c"]),
            ("调节阀开度", [("YSJ_14", "调节阀")], ["#2e9e5b"]),
        ]:
            for index, (column, name) in enumerate(columns):
                aux_series.append((name, column_values(column), colors[index % len(colors)]))
    else:
        aux_series = [("转速", column_values("YSJ_41"), "#8e44ad")]

    if aux_series:
        def _normalize(values: list[float | None]) -> list[float | None]:
            present = [value for value in values if value is not None]
            if not present:
                return values
            low, high = min(present), max(present)
            if high - low < 1e-9:
                return [None if value is None else 0.5 for value in values]
            return [None if value is None else (value - low) / (high - low) for value in values]

        normalized = [
            {"name": f"{name}(归一)", "color": color, "data": _series_points(_normalize(values))}
            for name, values, color in aux_series
        ]
        subplots.append(_subplot(
            "开关机辅助信号（归一化）", "", normalized, [], -0.05, 1.05,
        ))

    fault_start = 0
    fault_end = len(display) - 1
    seen = False
    for index, item in enumerate(display):
        if item is None:
            continue
        if not seen and item["t"] >= start:
            fault_start = index
            seen = True
        if item["t"] <= end:
            fault_end = index
    if not seen:
        fault_start = 0

    # Detail for the selected stage at the peak of the transition/event band.
    breakdown: dict[str, Any] = {}
    grade = label.get("grade") or next(iter(stages))
    if grade in stages:
        columns = stages[grade]
        band = [index for index in range(fault_start, fault_end + 1) if display[index] is not None]
        if band:
            peak = max(
                band,
                key=lambda index: max(spread[index] or 0.0, peer_high[index] or 0.0, peer_low[index] or 0.0),
            )
            values = []
            for offset, (column, name) in enumerate(columns):
                value = stage_values[grade][peak][offset]
                if value is not None:
                    values.append((name, value))
            if values:
                center = median([value for _, value in values])
                cylinders = [
                    {"name": name, "value": value, "dev": value - center}
                    for name, value in values
                ]
                high = max(cylinders, key=lambda item: item["dev"])
                low = min(cylinders, key=lambda item: item["dev"])
                breakdown = {
                    "time": times[peak] or "",
                    "stage": grade,
                    "median": center,
                    "cylinders": cylinders,
                    "high": high,
                    "low": low,
                    "spread": high["value"] - low["value"],
                    "single": len(cylinders) < 2,
                }

    return {
        "key": label["key"],
        "title": label["title"],
        "subtitle": (
            f"{label.get('kindLabel') or '标签'} {_fmt_time(start)} ~ {_fmt_time(end)} ｜ "
            f"展示 {_fmt_time(range_start)} ~ {_fmt_time(range_end)} ｜ "
            f"{label.get('oldEvent') or ''}"
        ),
        "stage": label["stage"],
        "kind": label["kind"],
        "x": {
            "max": len(display) - 1,
            "times": times,
            "gaps": gap_indexes,
            "shutdowns": shutdowns,
        },
        "fault": {"start": fault_start, "end": fault_end},
        "yellowBands": [],
        "subplots": subplots,
        "empty": False,
        "relatedChannel": label.get("relatedChannel") or "",
        "faultHit": label.get("faultHit") or "否",
        "faultDescriptions": label.get("faultDescriptions") or "",
        "evidence": label.get("evidence") or "",
        "breakdown": breakdown,
    }


class TemperatureTransitionService:
    def labels(self) -> dict[str, Any]:
        items = _labels()
        return {"items": items, "summary": _summary(items)}

    def series(self, key: str) -> dict[str, Any]:
        for item in _labels():
            if item["key"] == key:
                return _build_chart(item)
        raise KeyError(key)


temperature_transition_service = TemperatureTransitionService()