"""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", ]