Compressor/_tmp_temperature/config.py

"""Unit / point configuration for the PKS cylinder-temperature scan.

Only cylinder-body temperature points are configured here (exhaust temperature,
and optionally packing temperature).  The stage/cylinder mapping is the one
confirmed by the plant:

* 4#/5#/6#: two stages; cylinders 1/2/3 -> stage 1, cylinders 4/5/6 -> stage 2.
  The point names use stage-local numbering (``一级1#`` / ``二级1#``).
* 7#/8#/9#: three stages; cylinders 2/4/6 -> stage 1, 3/5 -> stage 2, 1 -> stage 3.

Run state:
* 7#/8#/9# expose speed in ``YSJ_41`` (``>0`` means running).
* 4#/5#/6# have no speed signal and no run-status table, so a data-derived
  proxy is used: motor/body vibration or first-stage suction pressure.
"""
from __future__ import annotations

from dataclasses import dataclass, field

from models.common.db import connect, number

# --- local constants (kept here so models/common stays untouched) ---
EXPECTED_PER_HOUR = 720
TEMP_BASELINE_HOURS = 168
TEMP_MIN_HISTORY_HOURS = 24
TEMP_CONFIRM_HOURS = 3
TEMP_MERGE_GAP_HOURS = 6
TEMP_RESTART_GAP_HOURS = 12
TEMP_MIN_HOUR_COVERAGE = 0.80
TEMP_SENSOR_MIN = -30.0
TEMP_SENSOR_MAX = 350.0
TEMP_MIN_STEADY = 45.0
TEMP_DROPOUT_REF = 60.0

# ``(stage, cylinder, column)`` for each unit's cylinder exhaust temperatures.
CYLINDERS: dict[int, list[tuple[str, str, str]]] = {
    4: [
        ("一级", "1#", "YSJ_9"),
        ("一级", "2#", "YSJ_10"),
        ("一级", "3#", "YSJ_11"),
        ("二级", "1#", "YSJ_13"),
        ("二级", "2#", "YSJ_14"),
        ("二级", "3#", "YSJ_15"),
    ],
    5: [
        ("一级", "1#", "YSJ_18"),
        ("一级", "2#", "YSJ_20"),
        ("一级", "3#", "YSJ_22"),
        ("二级", "1#", "YSJ_17"),
        ("二级", "2#", "YSJ_19"),
        ("二级", "3#", "YSJ_21"),
    ],
    6: [
        ("一级", "1#", "YSJ_18"),
        ("一级", "2#", "YSJ_20"),
        ("一级", "3#", "YSJ_22"),
        ("二级", "1#", "YSJ_17"),
        ("二级", "2#", "YSJ_19"),
        ("二级", "3#", "YSJ_21"),
    ],
    7: [
        ("一级", "2列", "YSJ_16"),
        ("一级", "4列", "YSJ_17"),
        ("一级", "6列", "YSJ_18"),
        ("二级", "3列", "YSJ_19"),
        ("二级", "5列", "YSJ_20"),
        ("三级", "1列", "YSJ_21"),
    ],
    8: [
        ("一级", "2列", "YSJ_16"),
        ("一级", "4列", "YSJ_17"),
        ("一级", "6列", "YSJ_18"),
        ("二级", "3列", "YSJ_19"),
        ("二级", "5列", "YSJ_20"),
        ("三级", "1列", "YSJ_21"),
    ],
    9: [
        ("一级", "2列", "YSJ_16"),
        ("一级", "4列", "YSJ_17"),
        ("一级", "6列", "YSJ_18"),
        ("二级", "3列", "YSJ_19"),
        ("二级", "5列", "YSJ_20"),
        ("三级", "1列", "YSJ_21"),
    ],
}

STAGE_ORDER: dict[int, list[str]] = {
    4: ["一级", "二级"],
    5: ["一级", "二级"],
    6: ["一级", "二级"],
    7: ["一级", "二级", "三级"],
    8: ["一级", "二级", "三级"],
    9: ["一级", "二级", "三级"],
}

# Hourly run-state specification.
#   rpm   -> ``YSJ_41 > 0``
#   proxy -> any of the listed columns above ``threshold``
RUN_SPEC: dict[int, dict] = {
    # 4# has a cumulative running-hour counter: +1 per running hour, flat when
    # stopped.  Body vibration / inlet pressure are the fallback if the counter
    # is missing or resets.
    4: {
        "type": "cumulative",
        "column": "YSJ_31",
        "sql": "(IFNULL(YSJ_37,0) > 0.5 OR IFNULL(YSJ_38,0) > 0.5 OR IFNULL(YSJ_2,0) > 0.5)",
    },
    # 5#/6# have no speed or cumulative counter.  Use 2-of-4:
    #   discharge pressure, inlet pressure, body vibration, valve position.
    5: {
        "type": "multi",
        "sql": "((IFNULL(YSJ_5,0) > 0.5) + (IFNULL(YSJ_3,0) > 0.3) + "
               "((IFNULL(YSJ_10,0) > 0.5) OR (IFNULL(YSJ_11,0) > 0.5) OR "
               "(IFNULL(YSJ_12,0) > 0.5) OR (IFNULL(YSJ_13,0) > 0.5)) + "
               "(IFNULL(YSJ_14,100) < 90)) >= 2",
    },
    6: {
        "type": "multi",
        "sql": "((IFNULL(YSJ_5,0) > 0.5) + (IFNULL(YSJ_3,0) > 0.3) + "
               "((IFNULL(YSJ_10,0) > 0.5) OR (IFNULL(YSJ_11,0) > 0.5) OR "
               "(IFNULL(YSJ_12,0) > 0.5) OR (IFNULL(YSJ_13,0) > 0.5)) + "
               "(IFNULL(YSJ_14,100) < 90)) >= 2",
    },
    7: {"type": "rpm", "column": "YSJ_41"},
    8: {"type": "rpm", "column": "YSJ_41"},
    9: {"type": "rpm", "column": "YSJ_41"},
}

# Optional packing-temperature groups for 5#/6# (cylinders 1..6 -> stage 1/2).
PACKING: dict[int, list[tuple[str, str, str]]] = {
    5: [
        ("一级", "1#", "YSJ_39"),
        ("一级", "2#", "YSJ_40"),
        ("一级", "3#", "YSJ_41"),
        ("二级", "4#", "YSJ_42"),
        ("二级", "5#", "YSJ_43"),
        ("二级", "6#", "YSJ_44"),
    ],
    6: [
        ("一级", "1#", "YSJ_39"),
        ("一级", "2#", "YSJ_40"),
        ("一级", "3#", "YSJ_41"),
        ("二级", "4#", "YSJ_42"),
        ("二级", "5#", "YSJ_43"),
        ("二级", "6#", "YSJ_44"),
    ],
}

DEFAULT_THRESHOLDS: dict[str, float] = {
    # Same-stage spread change versus its own recent normal spread.
    "mild_spread": 4.0,
    "abnormal_spread": 8.0,
    "severe_spread": 15.0,
    # Cylinder deviation from stage median, versus its own recent normal offset.
    "mild_peer": 4.0,
    "abnormal_peer": 6.0,
    "severe_peer": 10.0,
    # Low-side peer deviation (a cylinder falling relative to its stage peers).
    "mild_peer_low": 5.0,
    "abnormal_peer_low": 8.0,
    "severe_peer_low": 12.0,
    # Own-baseline robust z (reported; used only for single-cylinder stages).
    "mild_z": 3.0,
    "abnormal_z": 5.0,
    "severe_z": 8.0,
    # Own-baseline absolute rise (degC).
    "mild_delta": 5.0,
    "abnormal_delta": 10.0,
    "severe_delta": 15.0,
    # Short-term same-stage relative change (degC over 1/3/6h).
    "observe_rate1": 3.0, "mild_rate1": 4.0, "abnormal_rate1": 6.0, "severe_rate1": 9.0,
    "observe_rate3": 4.0, "mild_rate3": 5.0, "abnormal_rate3": 7.0, "severe_rate3": 11.0,
    "observe_rate6": 6.0, "mild_rate6": 7.0, "abnormal_rate6": 9.0, "severe_rate6": 13.0,
    # Intra-hour high-frequency fluctuation: std relative to its normal baseline.
    "hf_ratio_mild": 2.5, "hf_ratio_abnormal": 3.5, "hf_ratio_severe": 5.0,
    "hf_z_mild": 6.0, "hf_z_abnormal": 9.0, "hf_z_severe": 12.0,
    "hf_min_std": 0.8, "hf_base_floor": 0.3,
    "hf_persist_hours": 3,
    # 12h relative rise (single-cylinder and stage-level), integrated into events.
    "rise_single_mild": 3.0, "rise_single_abnormal": 4.5, "rise_single_severe": 6.0,
    "rise_stage_mild": 3.0, "rise_stage_abnormal": 5.0, "rise_stage_severe": 7.0,
    "rise_stage_flat_limit": 3.0,
    "rise_persist_hours": 2, "rise_baseline_hours": 6, "rise_min_history_hours": 4,
    # Channel / shutdown.
    "channel_ambient": 45.0, "channel_drop_delta": 8.0,
    "channel_min_drop": 20.0, "channel_zero_min": 5,
    "channel_recover_hours": 3, "channel_frozen_hours": 12,
    "channel_frozen_eps": 0.1, "channel_peer_variation": 3.0,
    # Persistence.
    "observe_hours": 1, "event_hours": 2, "confirm_hours": 3,
    "merge_gap_hours": 6,
}


def _number(value):
    return number(value)


def normalize_devices(values) -> list[str]:
    """Accept 4#~9# / 4 / 4号机 and return ``['4#', ...]``."""
    result = []
    for value in values:
        digits = "".join(ch for ch in str(value) if ch.isdigit())
        if not digits or not 4 <= int(digits) <= 9:
            raise ValueError(f"无法识别的压缩机编号：{value!r}（应为 4#~9#）")
        device = f"{int(digits)}#"
        if device not in result:
            result.append(device)
    return result


def load_device_batches(device: str) -> list[int]:
    """Resolve a device code to its successful PKS import batches."""
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT id FROM compressor_data_import "
                "WHERE data_type = 'PKS' AND directory_path = %s "
                "AND is_imported = 1 AND execution_status = '成功' ORDER BY id",
                (device,),
            )
            batches = [int(row["id"]) for row in cursor.fetchall()]
    finally:
        connection.close()
    if not batches:
        raise RuntimeError(f"{device} 没有成功的 PKS 导入批次")
    return batches


def _parse_limits(row: dict) -> tuple[float | None, float | None]:
    high = high_high = None
    for index in range(1, 5):
        kind = str(row.get(f"AlarmType{index}") or "").strip()
        limit = _number(row.get(f"AlarmLimit{index}"))
        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
    if high is None and high_high is not None:
        high = high_high
    return high, high_high


def load_point_limits(unit: int, columns: list[str]) -> dict[str, dict]:
    """Return ``{column: {description, units, high, high_high}}`` from site_point."""
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT ItemName, ItemDescription, Units, "
                "AlarmType1, AlarmType2, AlarmType3, AlarmType4, "
                "AlarmLimit1, AlarmLimit2, AlarmLimit3, AlarmLimit4 "
                "FROM site_point WHERE ItemName LIKE %s",
                (f"YSJ{unit}\\_%",),
            )
            rows = {row["ItemName"]: row for row in cursor.fetchall()}
    finally:
        connection.close()

    result = {}
    for column in columns:
        index = column.split("_")[1]
        row = rows.get(f"YSJ{unit}_{index}")
        if not row:
            result[column] = {"description": column, "units": "", "high": None, "high_high": None}
            continue
        high, high_high = _parse_limits(row)
        result[column] = {
            "description": row.get("ItemDescription") or column,
            "units": row.get("Units") or "",
            "high": high,
            "high_high": high_high,
        }
    return result


@dataclass
class UnitConfig:
    unit: int
    name: str
    batches: list[int]
    run: dict
    stages: dict[str, list[str]] = field(default_factory=dict)
    labels: dict[str, str] = field(default_factory=dict)
    limits: dict[str, dict] = field(default_factory=dict)
    point_meta: dict[str, dict] = field(default_factory=dict)
    thresholds: dict[str, float] = field(default_factory=lambda: dict(DEFAULT_THRESHOLDS))

    @property
    def columns(self) -> list[str]:
        return [column for columns in self.stages.values() for column in columns]

    @property
    def stage_order(self) -> list[str]:
        return list(self.stages)


def build_unit_config(unit: int, batches: list[int] | None = None, include_packing: bool = False) -> UnitConfig:
    if batches is None:
        batches = load_device_batches(f"{unit}#")

    points = list(CYLINDERS[unit])
    if include_packing and unit in PACKING:
        points.extend(PACKING[unit])

    stages: dict[str, list[str]] = {stage: [] for stage in STAGE_ORDER[unit]}
    labels: dict[str, str] = {}
    point_meta: dict[str, dict] = {}
    exhaust = {c for _, _, c in CYLINDERS[unit]}
    for stage, cylinder, column in points:
        stages[stage].append(column)
        label = f"{stage}{cylinder}缸排气温度" if column in exhaust else f"{stage}{cylinder}缸填料温度"
        labels[column] = label
        point_meta[column] = {"stage": stage, "cylinder": cylinder, "label": label}

    limits = load_point_limits(unit, list(labels))
    return UnitConfig(
        unit=unit,
        name=f"{unit}号机",
        batches=list(batches),
        run=RUN_SPEC[unit],
        stages=stages,
        labels=labels,
        limits=limits,
        point_meta=point_meta,
    )


__all__ = [
    "CYLINDERS", "STAGE_ORDER", "RUN_SPEC", "PACKING", "DEFAULT_THRESHOLDS",
    "UnitConfig", "build_unit_config", "load_point_limits",
    "normalize_devices", "load_device_batches",
    "EXPECTED_PER_HOUR", "TEMP_BASELINE_HOURS", "TEMP_MIN_HISTORY_HOURS",
    "TEMP_CONFIRM_HOURS", "TEMP_MERGE_GAP_HOURS", "TEMP_RESTART_GAP_HOURS",
    "TEMP_MIN_HOUR_COVERAGE", "TEMP_SENSOR_MIN", "TEMP_SENSOR_MAX",
    "TEMP_MIN_STEADY", "TEMP_DROPOUT_REF",
]