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