"""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()