"""Unified cylinder-temperature evolution detector. One pass over the hourly wide table produces: * always-on same-stage difference features; * short-term (1h/3h/6h) same-stage relative rise/fall; * 7#/8#/9# 三级 high-frequency fluctuation (separate table); * independent channel anomalies (shutdown / sudden drop / frozen / dead); * confirmed hourly levels, event segments and one evolution-cycle row per (run cycle x stage). Pressure and vibration are intentionally *not* touched here. """ from __future__ import annotations from collections import defaultdict, deque from datetime import datetime, timedelta from statistics import median import math import numpy as np from .algorithm import ( _assign_cycles, _baseline, _global_offsets, _mad, _median, _snapshot, channel_qc, clean_rows, linear_slope, output_sort_key, write_csv, ) from .config import ( TEMP_BASELINE_HOURS, TEMP_MIN_HISTORY_HOURS, TEMP_MIN_STEADY, TEMP_SENSOR_MAX, TEMP_SENSOR_MIN, UnitConfig, ) SEV_LABELS = {0: "正常", 1: "观察", 2: "轻微", 3: "异常", 4: "严重异常"} LEVELS = ("观察", "轻微", "异常", "严重异常") EVENT_LEVELS = ("轻微", "异常", "严重异常") DIRECTIONS = ("偏高", "偏低", "混合") def _fmt(value, digits=2): return "" if value is None else round(float(value), digits) def _part_and_point(label, config: UnitConfig) -> tuple[str, str]: """Derive the fault-table/3D part label and PKS point from a cylinder label.""" text = str(label or "").strip() if not text: return "", "" point = next((column for column, name in config.labels.items() if name == text), "") part = text for suffix in ("排气温度", "温度"): if part.endswith(suffix): part = part[: -len(suffix)] break return part, point def _valid(value): return value is not None and TEMP_SENSOR_MIN <= value <= TEMP_SENSOR_MAX and value > 0 def _value_at(history: deque, target: datetime, tolerance_hours: float = 0.6): for moment, value in reversed(history): delta = abs((moment - target).total_seconds()) / 3600.0 if delta <= tolerance_hours: return value if moment < target - timedelta(hours=tolerance_hours + 0.1): break return None def _sev_level(value, mild, abnormal, severe) -> int: if value is None: return 0 if value >= severe: return 4 if value >= abnormal: return 3 if value >= mild: return 2 return 0 def _sev_rate(value, observe, mild, abnormal, severe) -> int: if value is None: return 0 if value >= severe: return 4 if value >= abnormal: return 3 if value >= mild: return 2 if value >= observe: return 1 return 0 def _max_candidate(candidates: list[tuple[int, str, str]]): """Pick the highest severity, preferring level over short on ties.""" candidates = [item for item in candidates if item[0] > 0] if not candidates: return 0, "", "" return max(candidates, key=lambda item: (item[0], 1 if item[1] == "level" else 0)) # --------------------------------------------------------------------------- # channel anomalies # --------------------------------------------------------------------------- def mark_channel_anomalies(rows: list[dict], config: UnitConfig, thresholds: dict) -> list[dict]: ambient = thresholds.get("channel_ambient", 45.0) drop_delta = thresholds.get("channel_drop_delta", 8.0) min_drop = thresholds.get("channel_min_drop", 20.0) zero_min = int(thresholds.get("channel_zero_min", 5)) recover_hours = int(thresholds.get("channel_recover_hours", 3)) frozen_hours = int(thresholds.get("channel_frozen_hours", 12)) frozen_eps = thresholds.get("channel_frozen_eps", 0.1) peer_variation = thresholds.get("channel_peer_variation", 3.0) anomalies: list[dict] = [] stage_max = [] for row in rows: values = {} for stage, columns in config.stages.items(): nums = [row.get(f"{column}__avg") for column in columns if _valid(row.get(f"{column}__avg"))] values[stage] = max(nums) if nums else None stage_max.append(values) for stage, columns in config.stages.items(): for column in columns: series = [ (index, row["hour_start"], row.get(f"{column}__avg")) for index, row in enumerate(rows) if row.get(f"{column}__avg") is not None and TEMP_SENSOR_MIN <= row.get(f"{column}__avg") <= TEMP_SENSOR_MAX ] for position, (index, moment, value) in enumerate(series): peak = stage_max[index].get(stage) if peak is None or peak < TEMP_MIN_STEADY: continue previous = [item[2] for item in series[max(0, position - 6):position] if moment - item[1] <= timedelta(hours=6) and item[2] > 0] following = [item[2] for item in series[position + 1:position + 7] if item[1] - moment <= timedelta(hours=6) and item[2] > 0] reference_values = previous + following reference = median(reference_values) if len(reference_values) >= 3 else None next_values = [item[2] for item in series[position + 1:position + 1 + recover_hours] if item[1] - moment <= timedelta(hours=recover_hours)] recovers = any(value2 >= max(60.0, (reference or 60.0) - 10) for value2 in next_values) minraw = rows[index].get(f"{column}__minraw") zeros = rows[index].get(f"{column}__zeros") or 0 if minraw is not None and minraw <= ambient and zeros >= zero_min and peak >= 60: kind = "停机/通道异常" if recovers else "通道坏道" rows[index][f"{column}__dropout"] = True anomalies.append(_anomaly_record(config, rows[index], column, kind, minraw, reference)) continue if minraw is not None and reference is not None and minraw < reference - min_drop and recovers and peak >= 60: rows[index][f"{column}__dropout"] = True anomalies.append(_anomaly_record(config, rows[index], column, "通道骤降", minraw, reference)) continue if value <= ambient and reference is not None and value < reference - 20 and peak >= 60: kind = "停机/通道异常" if recovers else "通道坏道" rows[index][f"{column}__dropout"] = True anomalies.append(_anomaly_record(config, rows[index], column, kind, value, reference)) elif value > 0 and reference is not None and value < reference - drop_delta: if recovers: rows[index][f"{column}__dropout"] = True anomalies.append(_anomaly_record(config, rows[index], column, "通道骤降", value, reference)) # else: persistent relative low -> keep as process low # Rows whose hourly average is entirely empty (all samples 0) are # not in ``series``; still record them from minraw / zero count. for index, row in enumerate(rows): if row.get(f"{column}__dropout"): continue minraw = row.get(f"{column}__minraw") zeros = row.get(f"{column}__zeros") or 0 peak = stage_max[index].get(stage) if minraw is not None and minraw <= ambient and zeros >= zero_min and peak and peak >= 60: row[f"{column}__dropout"] = True anomalies.append(_anomaly_record(config, row, column, "停机/通道异常", minraw, None)) # frozen channel: identical for >= N hours while peers vary run_start = 0 for position in range(1, len(series) + 1): if position == len(series) or abs(series[position][2] - series[position - 1][2]) > frozen_eps: run_length = position - run_start if run_length >= frozen_hours: peer_values = [] for item in series[run_start:position]: index = item[0] for peer in columns: if peer == column: continue peer_value = rows[index].get(f"{peer}__avg") if _valid(peer_value): peer_values.append(peer_value) if peer_values and max(peer_values) - min(peer_values) >= peer_variation: for item in series[run_start:position]: index = item[0] rows[index][f"{column}__dropout"] = True anomalies.append(_anomaly_record( config, rows[index], column, "通道冻值", item[2], median(peer_values) )) run_start = position return anomalies def _anomaly_record(config: UnitConfig, row: dict, column: str, kind: str, value, peer) -> dict: meta = config.point_meta.get(column, {}) return { "机组": config.name, "批次": row["import_batch_id"], "点位": column, "点位描述": meta.get("label", column), "级": meta.get("stage", ""), "缸": meta.get("cylinder", ""), "异常类型": kind, "小时": row["hour_start"], "值": value, "同级基准": peer, } def build_channel_segments(anomalies: list[dict], thresholds: dict) -> list[dict]: group: dict[tuple, list[dict]] = defaultdict(list) for item in anomalies: group[(item["机组"], item["批次"], item["点位"])].append(item) segments = [] for (unit, batch, column), items in group.items(): items.sort(key=lambda item: item["小时"]) current = [] for item in items: if not current or item["小时"] - current[-1]["小时"] <= timedelta(hours=6): current.append(item) else: segments.append(_channel_segment(current, unit, batch, column)) current = [item] if current: segments.append(_channel_segment(current, unit, batch, column)) return sorted(segments, key=output_sort_key) def _channel_segment(items: list[dict], unit: str, batch: int, column: str) -> dict: priority = {"停机/通道异常": 4, "通道坏道": 3, "通道冻值": 2, "通道骤降": 1} kind = max({item["异常类型"] for item in items}, key=lambda value: priority.get(value, 0)) first, last = items[0], items[-1] values = [item["值"] for item in items if item["值"] is not None] peers = [item["同级基准"] for item in items if item["同级基准"] is not None] duration = int((last["小时"] - first["小时"]).total_seconds() / 3600) + 1 significant = ( duration >= 2 or kind in ("通道坏道", "通道冻值") or (kind == "停机/通道异常" and values and min(values) <= 0) ) return { "机组": unit, "批次": batch, "点位": column, "点位描述": first["点位描述"], "级": first["级"], "缸": first["缸"], "异常类型": kind, "开始小时": first["小时"].strftime("%Y-%m-%d %H:%M:%S"), "结束小时": last["小时"].strftime("%Y-%m-%d %H:%M:%S"), "持续自然小时数": duration, "有效小时数": len(items), "是否显著": "是" if significant else "否", "最小值": _fmt(min(values)) if values else "", "最大值": _fmt(max(values)) if values else "", "中位数": _fmt(median(values)) if values else "", "同级基准中位": _fmt(median(peers)) if peers else "", "原因": { "停机/通道异常": "单通道突然接近环境温度并恢复，按停机/采集异常处理，不参与温度故障", "通道骤降": "单通道短时骤降并恢复，按通道异常处理，不参与同级温差", "通道坏道": "单通道长期接近环境温度，按坏道处理", "通道冻值": "单通道长期不变而同级其他缸在变化，按冻值处理", }.get(kind, ""), "_dt": first["小时"], } # --------------------------------------------------------------------------- # unified features # --------------------------------------------------------------------------- def compute_features(rows: list[dict], config: UnitConfig, bad_channels: set[str], thresholds: dict) -> list[dict]: valid = list(rows) 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_std: dict[str, deque] = {column: deque() for column in config.columns} history_spread: dict[str, deque] = {stage: deque() for stage in config.stages} last_dt: dict[str, datetime | None] = {column: None for column in config.columns} rise_count: dict[str, int] = {column: 0 for column in config.columns} features: list[dict] = [] for row in valid: dt: datetime = row["hour_start"] window_start = dt - timedelta(hours=TEMP_BASELINE_HOURS) for values in (list(history.values()) + list(history_peer.values()) + list(history_std.values()) + list(history_spread.values())): while values and values[0][0] < window_start: values.popleft() snapshots = {} stage_medians = {} 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 snapshots[stage] = snapshot stage_medians[stage] = median(snapshot[0].values()) for stage, (values, stage_median, spread, peer) in snapshots.items(): spread_values = [item[1] for item in history_spread[stage]] spread_base = _median(spread_values) if len(spread_values) >= TEMP_MIN_HISTORY_HOURS else global_spread.get(stage) spread_delta = spread - spread_base if spread is not None and spread_base is not None else None per_column: dict[str, dict] = {} for column, value in values.items(): own = [item[1] for item in history[column]] tod = [value for moment, value in history[column] if moment.hour == dt.hour] if len(tod) >= 3: base = median(tod) scale = max(1.4826 * _mad(tod, base), abs(base) * 0.005, 0.5) else: base, scale = _baseline(own) 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[1] 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 change = {} for hours in (1, 3, 6): past = _value_at(history[column], dt - timedelta(hours=hours)) change[hours] = value - past if past is not None else None slope = linear_slope(own[-5:] + [value]) if len(own) >= 2 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 std_value = row.get(f"{column}__std") std_history = [item[1] for item in history_std[column] if item[1] is not None] std_base = _median(std_history) if len(std_history) >= TEMP_MIN_HISTORY_HOURS else None hf_ratio = (std_value / max(std_base, thresholds["hf_base_floor"])) if (std_value is not None and std_base is not None) else None hf_z = None if std_value is not None and std_base is not None and len(std_history) >= TEMP_MIN_HISTORY_HOURS: log_values = [math.log(value + 0.01) for value in std_history] log_center = median(log_values) log_scale = max(1.4826 * median([abs(value - log_center) for value in log_values]), 0.1) hf_z = (math.log(std_value + 0.01) - log_center) / log_scale recent_values = own[-6:] recent_median = median(recent_values) if recent_values else None rise_values = [ item[1] for item in history[column] if item[0] >= dt - timedelta(hours=thresholds["rise_baseline_hours"]) ] rise_base = median(rise_values) if len(rise_values) >= thresholds["rise_min_history_hours"] else None rise_value = value - rise_base if rise_base is not None else None per_column[column] = { "value": value, "base": base, "delta": delta, "z": z, "peer_dev": peer_dev, "peer_anom": peer_anom, "d1": change[1], "d3": change[3], "d6": change[6], "slope": slope, "rise": rise_count[column], "std": std_value, "std_base": std_base, "hf_ratio": hf_ratio, "hf_z": hf_z, "recent_median": recent_median, "rise_base": rise_base, "rise_value": rise_value, } median_rate = {} for hours in (1, 3, 6): rate_values = [item[f"d{hours}"] for item in per_column.values() if item[f"d{hours}"] is not None] median_rate[hours] = median(rate_values) if rate_values else None for column, item in per_column.items(): for hours in (1, 3, 6): peer_rate = item[f"d{hours}"] - median_rate[hours] if item[f"d{hours}"] is not None and median_rate[hours] is not None else None item[f"peer_d{hours}"] = peer_rate # ---- grading ---- 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_high = max((item["peer_anom"] for item in per_column.values() if item["peer_anom"] is not None), default=None) min_peer = min((item["peer_anom"] for item in per_column.values() if item["peer_anom"] is not None), default=0.0) peer_low = abs(min(0.0, min_peer)) rate_max = {hours: max((item[f"peer_d{hours}"] for item in per_column.values() if item[f"peer_d{hours}"] is not None and item["recent_median"] is not None and item["value"] > item["recent_median"]), default=None) for hours in (1, 3, 6)} rate_min = {hours: min((item[f"peer_d{hours}"] for item in per_column.values() if item[f"peer_d{hours}"] is not None and item["recent_median"] is not None and item["value"] < item["recent_median"]), default=None) for hours in (1, 3, 6)} # A drop followed by a rebound (or a rise followed by a fall) is a # load/transition/channel artifact, not a one-sided short-term fault. rebound = ( (rate_max[1] is not None and rate_min[3] is not None and rate_max[1] >= 4 and rate_min[3] <= -4) or (rate_min[1] is not None and rate_max[3] is not None and rate_min[1] <= -4 and rate_max[3] >= 4) or (rate_max[1] is not None and rate_min[6] is not None and rate_max[1] >= 4 and rate_min[6] <= -4) or (rate_min[1] is not None and rate_max[6] is not None and rate_min[1] <= -4 and rate_max[6] >= 4) ) if rebound: rate_max = {hours: None for hours in (1, 3, 6)} rate_min = {hours: None for hours in (1, 3, 6)} 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) spread_high_column = max(values, key=values.get) if values else None spread_low_column = min(values, key=values.get) if values else None hf_columns = [column for column in per_column if per_column[column]["hf_ratio"] is not None] hf_column = max(hf_columns, key=lambda column: per_column[column]["hf_ratio"], default=None) hf_ratio = per_column[hf_column]["hf_ratio"] if hf_column else None hf_std = per_column[hf_column]["std"] if hf_column else None hf_base = per_column[hf_column]["std_base"] if hf_column else None hf_zs = [per_column[column]["hf_z"] for column in per_column if per_column[column]["hf_z"] is not None] hf_z = max(hf_zs) if hf_zs else None hf_sev = 0 if hf_std is not None and hf_std >= thresholds["hf_min_std"]: hf_sev = max( _sev_level(hf_ratio, thresholds["hf_ratio_mild"], thresholds["hf_ratio_abnormal"], thresholds["hf_ratio_severe"]), _sev_level(hf_z, thresholds["hf_z_mild"], thresholds["hf_z_abnormal"], thresholds["hf_z_severe"]), ) hf_reason = "" if hf_sev: label = config.labels.get(hf_column, "") if hf_column else "" if hf_ratio is not None and hf_std is not None and hf_base is not None: hf_reason = f"{label}高频波动×{hf_ratio:.1f}（基线{hf_base:.2f}，实测{hf_std:.2f}）" elif hf_z is not None: hf_reason = f"{label}高频波动z={hf_z:.1f}" rise_values = [item["rise_value"] for item in per_column.values() if item["rise_value"] is not None] stage_rise = median(rise_values) if rise_values else None relative_rises = { column: item["rise_value"] - stage_rise for column, item in per_column.items() if item["rise_value"] is not None and stage_rise is not None } if len(values) >= 2: rise_single_column = max(relative_rises, key=relative_rises.get, default=None) rise_single = relative_rises[rise_single_column] if rise_single_column else None rise_stage_value = stage_rise else: rise_single_column = max( (column for column in per_column if per_column[column]["rise_value"] is not None), key=lambda column: per_column[column]["rise_value"], default=None) rise_single = per_column[rise_single_column]["rise_value"] if rise_single_column else None rise_stage_value = rise_single previous_values = [] for column in per_column: previous = _value_at(history[column], dt - timedelta(hours=1)) if previous is not None: previous_values.append(previous) previous_stage_median = median(previous_values) if previous_values else None stage_rising = previous_stage_median is not None and stage_median > previous_stage_median rise_flat = stage_rise is not None and abs(stage_rise) <= thresholds["rise_stage_flat_limit"] # Only a single-cylinder rise *relative to its stage peers* is a # fault candidate. A whole-stage rise is the normal daily load # ramp; it is kept as a feature, not an event. if len(values) >= 2: rise_single_sev = _sev_level( rise_single, thresholds["rise_single_mild"], thresholds["rise_single_abnormal"], thresholds["rise_single_severe"] ) if (stage_rising and rise_flat) else 0 rise_sev = rise_single_sev if rise_sev: rise_type = "单缸抬升" rise_label = config.labels.get(rise_single_column, "") if rise_single_column else "" rise_reason = f"6h整级抬升{rise_stage_value:+.1f}℃；{rise_label}相对同级{rise_single:+.1f}℃" else: rise_type = "" rise_label = "" rise_reason = "" else: rise_sev = 0 rise_type = "" rise_label = "" rise_reason = "" def peer_reason(column, anomaly, sign): if column is None or anomaly is None: return "" item = per_column[column] dev = item["peer_dev"] base = dev - item["peer_anom"] label = config.labels.get(column, column) return f"{label}较同级中位{dev:+.1f}℃（平时{base:+.1f}℃），较平时{sign}{anomaly:.1f}℃" def spread_reason(): if spread is None or spread_delta is None: return "" high = config.labels.get(spread_high_column, "") low = config.labels.get(spread_low_column, "") base = f"{spread_base:.1f}" if spread_base is not None else "-" return f"级差{spread:.1f}℃（{high}-{low}，平时{base}℃）较平时+{spread_delta:.1f}℃" if len(values) >= 2: high_candidates = [ (_sev_level(peer_high, thresholds["mild_peer"], thresholds["abnormal_peer"], thresholds["severe_peer"]), "level", peer_reason(peak_peer_column, peer_high, "+")), (_sev_level(spread_delta, thresholds["mild_spread"], thresholds["abnormal_spread"], thresholds["severe_spread"]), "level", spread_reason()), ] low_candidates = [ (_sev_level(peer_low, thresholds["mild_peer_low"], thresholds["abnormal_peer_low"], thresholds["severe_peer_low"]), "level", peer_reason(low_peer_column, peer_low, "-")), ] for hours in (1, 3, 6): if rate_max[hours] is not None: high_candidates.append((_sev_rate(rate_max[hours], thresholds[f"observe_rate{hours}"], thresholds[f"mild_rate{hours}"], thresholds[f"abnormal_rate{hours}"], thresholds[f"severe_rate{hours}"]), "short", f"{hours}h同级升{rate_max[hours]:.1f}℃")) if rate_min[hours] is not None: low_candidates.append((_sev_rate(-rate_min[hours], thresholds[f"observe_rate{hours}"], thresholds[f"mild_rate{hours}"], thresholds[f"abnormal_rate{hours}"], thresholds[f"severe_rate{hours}"]), "short", f"{hours}h同级降{-rate_min[hours]:.1f}℃")) high_sev, high_source, high_reason = _max_candidate(high_candidates) low_sev, low_source, low_reason = _max_candidate(low_candidates) else: # Single-cylinder stage: no same-stage peers. The old # cross-stage residual feature is intentionally removed; only # high-frequency fluctuation is kept for the tertiary stage. high_sev = low_sev = 0 high_source = low_source = "" high_reason = low_reason = "" if high_sev and low_sev: base_direction, base_severity, base_source = "混合", max(high_sev, low_sev), "混合" base_reason = "；".join(item for item in (high_reason, low_reason) if item) elif high_sev: base_direction, base_severity, base_source, base_reason = "偏高", high_sev, high_source, high_reason elif low_sev: base_direction, base_severity, base_source, base_reason = "偏低", low_sev, low_source, low_reason else: base_direction, base_severity, base_source, base_reason = "", 0, "", "" direction, severity, source, reason = base_direction, base_severity, base_source, base_reason if base_severity == 0 and rise_sev > 0: direction, severity, source, reason = "偏高", rise_sev, "rise", rise_reason if severity == 0: feature_type = "" elif source == "hf": feature_type = "高频波动" elif source == "rise": feature_type = rise_type elif direction == "混合": feature_type = "混合" elif source == "short": feature_type = "短时突升" if direction == "偏高" else "短时突降" else: feature_type = "持续偏高" if direction == "偏高" else "持续偏低" confidence = "高" if source == "level" else "中" 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) 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] 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(deltas) if deltas else None), "最大自身偏离缸": config.labels.get(strongest, "") if strongest else "", "最大同级偏离": _fmt(peer_high), "最大同级偏离缸": config.labels.get(peak_peer_column, "") if peak_peer_column else "", "最大同级偏低": _fmt(peer_low if peer_low else None), "最大同级偏低缸": config.labels.get(low_peer_column, "") if low_peer_column else "", "最大稳健z": _fmt(max(zs) if zs else None), "高频波动倍数": _fmt(hf_ratio), "高频波动z": _fmt(hf_z), "高频标准差": _fmt(hf_std), "高频基线标准差": _fmt(hf_base), "高频波动缸": config.labels.get(hf_column, "") if hf_column else "", "6小时整级抬升": _fmt(rise_stage_value), "6小时单缸抬升": _fmt(rise_single), "6小时单缸相对抬升": _fmt(relative_rises.get(rise_single_column)) if rise_single_column else "", "6小时基线": _fmt(per_column[rise_single_column]["rise_base"]) if rise_single_column else "", "抬升类型": rise_type, "抬升缸": rise_label, "最强缸": 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(per_column[strongest]["slope"], 4) if strongest else "", "连续上升小时数": per_column[strongest]["rise"] if strongest else 0, "同级相对1h": _fmt(rate_max[1] if rate_max[1] is not None and (rate_max[1] or 0) >= (abs(rate_min[1] or 0)) else rate_min[1]), "同级相对3h": _fmt(rate_max[3] if rate_max[3] is not None and (rate_max[3] or 0) >= (abs(rate_min[3] or 0)) else rate_min[3]), "同级相对6h": _fmt(rate_max[6] if rate_max[6] is not None and (rate_max[6] or 0) >= (abs(rate_min[6] or 0)) else rate_min[6]), "各缸温度": "；".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), "方向": direction, "特征类型": feature_type, "来源": source, "置信度": confidence, "原始等级": SEV_LABELS[severity], "判定依据": reason, "温度等级": "", "是否事件": "", "首次短时触发": "", "首次持续触发": "", "关联通道异常": "", "_raw_sev": severity, "_direction": direction, "_source": source, "_hf_sev": hf_sev, "_hf_ratio": hf_ratio, "_hf_z": hf_z, "_hf_std": hf_std, "_hf_base": hf_base, "_hf_column": hf_column, "_rise_sev": rise_sev, "_rise_type": rise_type, "_rise_stage": rise_stage_value, "_rise_single": rise_single, "_type": feature_type, "_confidence": confidence, "_dt": dt, "_strongest": strongest, }) # append histories after all computations for this stage for column, value in values.items(): history[column].append((dt, value)) history_peer[column].append((dt, peer[column])) history_std[column].append((dt, per_column[column]["std"])) last_dt[column] = dt if spread is not None: history_spread[stage].append((dt, spread)) return features # --------------------------------------------------------------------------- # confirmation / segments / cycles # --------------------------------------------------------------------------- def confirm_levels(features: list[dict], thresholds: dict) -> None: confirm_hours = int(thresholds.get("confirm_hours", 3)) event_hours = int(thresholds.get("event_hours", 2)) hf_persist = int(thresholds.get("hf_persist_hours", 3)) rise_persist = int(thresholds.get("rise_persist_hours", 2)) grouped: dict[tuple, list[dict]] = defaultdict(list) for row in features: grouped[(row["机组"], row["批次"], row["级"])].append(row) for rows in grouped.values(): rows.sort(key=lambda row: row["_dt"]) index = 0 while index < len(rows): if rows[index]["_raw_sev"] == 0: rows[index]["温度等级"] = "正常" rows[index]["是否事件"] = "否" index += 1 continue end = index while end + 1 < len(rows) and rows[end + 1]["_raw_sev"] > 0: if rows[end + 1]["_dt"] - rows[end]["_dt"] != timedelta(hours=1): break end += 1 run = rows[index:end + 1] run_length = len(run) level_rows = [row for row in run if row["_source"] == "level" and row["_raw_sev"] > 0] short_rows = [row for row in run if row["_source"] == "short" and row["_raw_sev"] > 0] hf_rows = [row for row in run if row["_source"] == "hf" and row["_raw_sev"] > 0] rise_rows = [row for row in run if row["_source"] == "rise" and row["_raw_sev"] > 0] max_level = max((row["_raw_sev"] for row in level_rows), default=0) max_short = max((row["_raw_sev"] for row in short_rows), default=0) strong_short = [row for row in short_rows if row["_raw_sev"] >= 2] short_streak = best_short_streak = 0 previous_direction = None for row in strong_short: if row["_direction"] == previous_direction and previous_direction in ("偏高", "偏低"): short_streak += 1 else: short_streak = 1 previous_direction = row["_direction"] best_short_streak = max(best_short_streak, short_streak) max_hf = max((row["_raw_sev"] for row in hf_rows), default=0) strong_hf = [row for row in hf_rows if row["_raw_sev"] >= 2] max_rise = max((row["_raw_sev"] for row in rise_rows), default=0) strong_rise = [row for row in rise_rows if row["_raw_sev"] >= 2] final = 0 if max_level: if run_length >= confirm_hours: final = max(final, max_level) elif run_length >= event_hours: if max_level >= 3: final = max(final, 3) elif max_level >= 2: final = max(final, 2) elif max_level >= 2: final = max(final, 1) if max_short >= 2 and best_short_streak >= 2: final = max(final, max_short) if max_hf >= 2 and len(strong_hf) >= hf_persist: final = max(final, max_hf) if max_rise >= 2 and len(strong_rise) >= rise_persist: final = max(final, max_rise) if final == 0: final = 1 peak = max(run, key=lambda row: (row["_raw_sev"], 1 if row["_source"] == "level" else 0)) active_directions = {row["_direction"] for row in run if row["_direction"] and row["_raw_sev"] >= 2} if "混合" in active_directions or len(active_directions) > 1: direction = "混合" else: direction = peak["_direction"] or "" first_short = next((row["小时"] for row in run if row["_source"] == "short" and row["_raw_sev"] >= 2), "") first_level = next((row["小时"] for row in run if row["_source"] in ("level", "rise") and row["_raw_sev"] >= 2), "") for row in run: row["温度等级"] = SEV_LABELS[final] row["是否事件"] = "是" if final >= 2 else "否" row["方向"] = direction row["首次短时触发"] = first_short row["首次持续触发"] = first_level if final >= 2 and direction == "混合": row["特征类型"] = "混合" elif final >= 2 and row["_type"]: row["特征类型"] = row["_type"] index = end + 1 def build_observations(features: list[dict], segments: list[dict], config: UnitConfig) -> list[dict]: """One row per 1h observation hour, linked to the next event within 24h.""" events: dict[tuple, list[dict]] = defaultdict(list) for segment in segments: if segment["是否事件"] == "是": events[(segment["机组"], segment["级"])].append(segment) rows = [] for row in features: if row["温度等级"] != "观察": continue candidates = [] for segment in events.get((row["机组"], row["级"]), []): delta = (segment["_dt"] - row["_dt"]).total_seconds() / 3600 if 0 <= delta <= 24: candidates.append(segment) event = min(candidates, key=lambda item: item["_dt"]) if candidates else None rows.append({ "机组": row["机组"], "批次": row["批次"], "周期编号": row["周期编号"], "小时": row["小时"], "级": row["级"], "方向": row["方向"], "特征类型": row["特征类型"], "置信度": row["置信度"], "最高温度": row["最高温度"], "级差": row["级差"], "级差变化": row["级差变化"], "最大同级偏离": row["最大同级偏离"], "最大同级偏低": row["最大同级偏低"], "最大同级相对1h": row["同级相对1h"], "最大同级相对3h": row["同级相对3h"], "最大同级相对6h": row["同级相对6h"], "判定依据": row["判定依据"], "关联周期编号": event["周期编号"] if event else "", "关联事件开始": event["开始小时"] if event else "", "关联事件等级": event["温度阶段"] if event else "", "是否升级为事件": "是" if event else "否", "_dt": row["_dt"], }) return sorted(rows, key=output_sort_key) def build_segments(features: list[dict], config: UnitConfig, channel_segments: list[dict], thresholds: dict, min_hours: int = 1) -> list[dict]: channel_by_key: dict[tuple, set[str]] = defaultdict(set) for segment in channel_segments: channel_by_key[(segment["机组"], segment["级"])].add(segment["异常类型"]) grouped: dict[tuple, list[dict]] = defaultdict(list) for row in features: if row["温度等级"] in EVENT_LEVELS: grouped[(row["机组"], row["批次"], row["级"], row["温度等级"])].append(row) segments = [] for (unit, batch, stage, level), rows in grouped.items(): rows.sort(key=lambda row: row["_dt"]) current = [] for row in rows: if not current or row["_dt"] - current[-1]["_dt"] <= timedelta(hours=thresholds.get("merge_gap_hours", 6)): current.append(row) else: segments.append(_build_segment(current, unit, batch, stage, level, channel_by_key, config)) current = [row] if current: segments.append(_build_segment(current, unit, batch, stage, level, channel_by_key, 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, channel_by_key: dict, config: UnitConfig) -> dict: first, last = rows[0], rows[-1] peak = max(rows, key=lambda row: (row["_raw_sev"], 1 if row["_source"] == "level" else 0)) peak_label = peak.get("最强缸") or peak.get("最高温度缸") or "" spread_rows = [row for row in rows if row["级差"] not in ("", None)] spread_peak = max(spread_rows, key=lambda row: row["级差"]) if spread_rows else None peer_rows = [row for row in rows if row["最大同级偏离"] not in ("", None)] peer_peak = max(peer_rows, key=lambda row: row["最大同级偏离"]) if peer_rows else None low_rows = [row for row in rows if row["最大同级偏低"] not in ("", None)] low_peak = max(low_rows, key=lambda row: row["最大同级偏低"]) if low_rows else None short_times = [row["小时"] for row in rows if row["_source"] == "short" and row["_raw_sev"] >= 2] level_times = [row["小时"] for row in rows if row["_source"] == "level" and row["_raw_sev"] >= 2] directions = {row["方向"] for row in rows if row["方向"]} direction = "混合" if "混合" in directions or len(directions) > 1 else (next(iter(directions), "")) return { "机组": unit, "批次": batch, "周期编号": first["周期编号"], "级": stage, "温度阶段": level, "阶段部位": _part_and_point(peak_label, config)[0], "阶段点位": _part_and_point(peak_label, config)[1], "是否事件": "是" if level in EVENT_LEVELS else "否", "是否形成确认等级": "是" if level in ("异常", "严重异常") else "否", "方向": direction or peak["_direction"], "特征类型": "混合" if direction == "混合" else peak["_type"], "置信度": peak["_confidence"], "开始小时": first["小时"], "结束小时": last["小时"], "持续自然小时数": int((last["_dt"] - first["_dt"]).total_seconds() / 3600) + 1, "有效运行小时数": len(rows), "峰值时刻": peak["小时"], "峰值缸": peak_label, "最高温度": max((row["最高温度"] for row in rows if row["最高温度"] not in ("", None)), default=""), "最大自身偏离": peak["最大自身偏离"], "最大级差": spread_peak["级差"] if spread_peak else "", "最大级差基线": spread_peak["级差基线"] if spread_peak else "", "最大级差变化": spread_peak["级差变化"] if spread_peak else "", "最大同级偏离": peer_peak["最大同级偏离"] if peer_peak else "", "最大同级偏低": low_peak["最大同级偏低"] if low_peak else "", "最大稳健z": peak["最大稳健z"], "最大同级相对1h": max((row["同级相对1h"] for row in rows if row["同级相对1h"] not in ("", None)), default=""), "最大同级相对3h": max((row["同级相对3h"] for row in rows if row["同级相对3h"] not in ("", None)), default=""), "最大同级相对6h": max((row["同级相对6h"] for row in rows if row["同级相对6h"] not in ("", None)), default=""), "首次短时触发": min(short_times) if short_times else "", "首次持续触发": min(level_times) if level_times 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=""), "阶段高报警小时数": sum(1 for row in rows if row.get("超告警缸") not in ("", None)), "阶段高高报警小时数": sum(1 for row in rows if row.get("超高高缸") not in ("", None)), "关联通道异常": "、".join(sorted(channel_by_key.get((unit, stage), set()))), "阶段触发依据": "；".join( list(dict.fromkeys( row["判定依据"] for row in rows if row["判定依据"] )) [:3]), "_dt": first["_dt"], } def build_hf_segments(features: list[dict], config: UnitConfig, thresholds: dict) -> list[dict]: """High-frequency fluctuation events, tertiary stage only.""" hf_persist = int(thresholds.get("hf_persist_hours", 3)) rise_persist = int(thresholds.get("rise_persist_hours", 2)) grouped: dict[tuple, list[dict]] = defaultdict(list) for row in features: if row["级"] != "三级" or row.get("_hf_sev", 0) <= 0: continue grouped[(row["机组"], row["批次"], row["级"])].append(row) raw_segments = [] for (unit, batch, stage), rows in grouped.items(): rows.sort(key=lambda row: row["_dt"]) current = [] for row in rows: if not current or row["_dt"] - current[-1]["_dt"] == timedelta(hours=1): current.append(row) else: if len(current) >= hf_persist: raw_segments.append(_build_hf_segment(current, unit, batch, stage, config)) current = [row] if len(current) >= hf_persist: raw_segments.append(_build_hf_segment(current, unit, batch, stage, config)) raw_segments.sort(key=lambda row: (row["机组"], row["批次"], row["_level"], row["_dt"])) merged = [] for segment in raw_segments: if (merged and merged[-1]["机组"] == segment["机组"] and merged[-1]["批次"] == segment["批次"] and merged[-1]["_level"] == segment["_level"] and segment["_dt"] - merged[-1]["_dt"] <= timedelta(hours=6)): merged[-1]["结束小时"] = segment["结束小时"] merged[-1]["有效运行小时数"] += segment["有效运行小时数"] merged[-1]["持续自然小时数"] += segment["持续自然小时数"] merged[-1]["最高温度"] = max(merged[-1]["最高温度"], segment["最高温度"]) merged[-1]["高频波动倍数"] = max(merged[-1]["高频波动倍数"], segment["高频波动倍数"]) merged[-1]["高频波动z"] = max(merged[-1]["高频波动z"], segment["高频波动z"]) merged[-1]["高频标准差"] = max(merged[-1]["高频标准差"], segment["高频标准差"]) merged[-1]["阶段触发依据"] = segment["阶段触发依据"] else: merged.append(segment) return sorted(merged, key=output_sort_key) def _build_hf_segment(rows: list[dict], unit: str, batch: int, stage: str, config: UnitConfig) -> dict: peak = max(rows, key=lambda row: (row["_hf_sev"], row.get("_hf_ratio") or 0)) level = peak["_hf_sev"] column = peak.get("_hf_column") or "" label = config.labels.get(column, "") ratio = peak.get("_hf_ratio") or 0 z_value = peak.get("_hf_z") std_value = peak.get("_hf_std") or 0 base_value = peak.get("_hf_base") or 0 reason = f"{label}高频波动×{ratio:.1f}（基线{base_value:.2f}，实测{std_value:.2f}）" if z_value is not None: reason += f"，z={z_value:.1f}" return { "机组": unit, "批次": batch, "周期编号": rows[0]["周期编号"], "级": stage, "温度阶段": SEV_LABELS[level], "阶段部位": _part_and_point(label, config)[0], "阶段点位": _part_and_point(label, config)[1], "是否事件": "是", "是否形成确认等级": "是" if SEV_LABELS[level] in ("异常", "严重异常") else "否", "方向": "波动", "特征类型": "高频波动", "置信度": "中", "开始小时": rows[0]["小时"], "结束小时": rows[-1]["小时"], "持续自然小时数": int((rows[-1]["_dt"] - rows[0]["_dt"]).total_seconds() / 3600) + 1, "有效运行小时数": len(rows), "峰值时刻": peak["小时"], "峰值缸": label, "最高温度": max((row["最高温度"] for row in rows if row["最高温度"] not in ("", None)), default=""), "高频波动倍数": round(ratio, 2), "高频波动z": round(z_value, 2) if z_value is not None else "", "高频标准差": round(std_value, 2), "高频基线标准差": round(base_value, 2), "高频波动缸": label, "阶段触发依据": reason, "_dt": rows[0]["_dt"], "_level": level, } def summarize_cycles(features: list[dict], segments: list[dict], config: UnitConfig) -> list[dict]: grouped: dict[tuple, list[dict]] = defaultdict(list) for row in features: grouped[(row["机组"], row["批次"], row["周期编号"], row["级"])].append(row) result = [] for (unit, batch, cycle_id, stage), rows in grouped.items(): rows.sort(key=lambda row: row["_dt"]) stage_segments = [segment for segment in segments if segment["周期编号"] == cycle_id and segment["级"] == stage] sequence = [] previous = None for row in rows: label = row["温度等级"] if label != previous: sequence.append(label) previous = label if len(sequence) > 1 and sequence[-1] == "正常": sequence[-1] = "恢复" peak = max(rows, key=lambda row: (row["_raw_sev"], 1 if row["_source"] == "level" else 0)) summary = { "机组": unit, "批次": batch, "周期编号": cycle_id, "级": stage, "周期开始小时": rows[0]["小时"], "周期结束小时": rows[-1]["小时"], "有效运行小时数": len(rows), "阶段演化顺序": "→".join(sequence) if sequence else "正常", "方向": peak["方向"] if peak["_raw_sev"] else "", "特征类型": peak["_type"] if peak["_raw_sev"] else "", "置信度": peak["置信度"] if peak["_raw_sev"] else "", "峰值时刻": peak["小时"] if peak["_raw_sev"] else "", "峰值缸": peak["最强缸"] if peak["_raw_sev"] else "", "首次短时触发": min((row["首次短时触发"] for row in rows if row["首次短时触发"]), default=""), "首次持续触发": min((row["首次持续触发"] for row in rows if row["首次持续触发"]), default=""), "关联通道异常": "、".join(sorted({segment["关联通道异常"] for segment in stage_segments if segment["关联通道异常"]})), } 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[f"是否达到{level}"] = "是" if selected else "否" 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="") summary["周期最大同级偏低"] = max((row["最大同级偏低"] for row in rows if row["最大同级偏低"] not in ("", None)), default="") result.append(summary) return result # --------------------------------------------------------------------------- # outputs # --------------------------------------------------------------------------- FEATURE_FIELDS = [ "机组", "批次", "周期编号", "小时", "级", "有效缸数", "运行覆盖率", "最高温度", "最低温度", "最高温度缸", "最高报警阈值", "高高报警阈值", "超告警缸", "超高高缸", "级差", "级差基线", "级差变化", "最大自身偏离", "最大自身偏离缸", "最大同级偏离", "最大同级偏离缸", "最大同级偏低", "最大同级偏低缸", "最大稳健z", "高频波动倍数", "高频波动z", "高频标准差", "高频基线标准差", "高频波动缸", "6小时整级抬升", "6小时单缸抬升", "6小时单缸相对抬升", "6小时基线", "抬升类型", "抬升缸", "最强缸", "近6小时斜率", "连续上升小时数", "同级相对1h", "同级相对3h", "同级相对6h", "各缸温度", "各缸自身偏离", "各缸同级偏离", "方向", "特征类型", "来源", "置信度", "温度等级", "原始等级", "是否事件", "首次短时触发", "首次持续触发", "关联通道异常", "判定依据", ] CANONICAL_FIELDS = [ "机组", "批次", "周期编号", "级", "温度阶段", "阶段部位", "阶段点位", "开始小时", "结束小时", "持续自然小时数", "有效运行小时数", "峰值时刻", "峰值缸", "最高温度", "最大自身偏离", "最大级差", "最大级差基线", "最大级差变化", "最大同级偏离", "最大同级偏低", "最大稳健z", "最大同级相对1h", "最大同级相对3h", "最大同级相对6h", "首次短时触发", "首次持续触发", "阶段最高报警阈值", "阶段高高报警阈值", "阶段高报警小时数", "阶段高高报警小时数", "方向", "特征类型", "置信度", "高频波动倍数", "高频波动z", "高频标准差", "高频基线标准差", "高频波动缸", "去趋势残差标准差", "每小时反复升降次数", "是否形成确认等级", "关联通道异常", "阶段触发依据", ] SEGMENT_FIELDS = CANONICAL_FIELDS HF_FIELDS = CANONICAL_FIELDS HF_CANDIDATE_FIELDS = CANONICAL_FIELDS + ["是否确认"] CYCLE_FIELDS = [ "机组", "批次", "周期编号", "级", "周期开始小时", "周期结束小时", "有效运行小时数", "阶段演化顺序", "方向", "特征类型", "置信度", "峰值时刻", "峰值缸", "首次短时触发", "首次持续触发", "关联通道异常", "观察阶段开始小时", "观察阶段结束小时", "观察阶段运行小时数", "是否达到观察", "轻微阶段开始小时", "轻微阶段结束小时", "轻微阶段运行小时数", "是否达到轻微", "异常阶段开始小时", "异常阶段结束小时", "异常阶段运行小时数", "是否达到异常", "严重异常阶段开始小时", "严重异常阶段结束小时", "严重异常阶段运行小时数", "是否达到严重异常", "周期最高温度", "周期最大级差", "周期最大级差变化", "周期最大同级偏离", "周期最大同级偏低", ] CHANNEL_FIELDS = [ "机组", "批次", "点位", "点位描述", "级", "缸", "异常类型", "开始小时", "结束小时", "持续自然小时数", "有效小时数", "是否显著", "最小值", "最大值", "中位数", "同级基准中位", "原因", ] QC_FIELDS = [ "机组", "点位", "点位描述", "运行小时数", "最小值", "最大值", "中位数", "不同值个数", "最长不变小时", "零值比例", "越界比例", "判定", "原因", ] SAME_STAGE_FIELDS = [ "机组", "批次", "周期编号", "小时", "级", "有效缸数", "级差", "级差基线", "级差变化", "最大同级偏离", "最大同级偏低", "各缸温度", "各缸同级偏离", ] SHORT_FIELDS = [ "机组", "批次", "周期编号", "小时", "级", "同级相对1h", "同级相对3h", "同级相对6h", "方向", "最强缸", ] OBSERVATION_FIELDS = [ "机组", "批次", "周期编号", "小时", "级", "方向", "特征类型", "置信度", "最高温度", "级差", "级差变化", "最大同级偏离", "最大同级偏低", "最大同级相对1h", "最大同级相对3h", "最大同级相对6h", "判定依据", "关联周期编号", "关联事件开始", "关联事件等级", "是否升级为事件", ] def write_results(output_dir, qc, features, segments, cycles, channel_segments, observations, hf_segments, hf_candidates=None): output_dir.mkdir(parents=True, exist_ok=True) candidates = hf_candidates if hf_candidates is not None else hf_segments canonical = clean_rows(segments) + clean_rows(hf_segments) canonical.sort(key=output_sort_key) canonical_hf = [row for row in canonical if row.get("特征类型") == "高频波动"] write_csv(output_dir / "温度小时特征.csv", clean_rows(features), FEATURE_FIELDS, {field: field for field in FEATURE_FIELDS}) write_csv(output_dir / "温度异常段.csv", canonical, CANONICAL_FIELDS, {field: field for field in CANONICAL_FIELDS}) write_csv(output_dir / "温度观察段.csv", clean_rows(observations), OBSERVATION_FIELDS, {field: field for field in OBSERVATION_FIELDS}) write_csv(output_dir / "温度高频波动段.csv", canonical_hf, HF_FIELDS, {field: field for field in HF_FIELDS}) write_csv(output_dir / "温度高频候选.csv", clean_rows(candidates), HF_CANDIDATE_FIELDS, {field: field for field in HF_CANDIDATE_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(channel_segments), CHANNEL_FIELDS, {field: field for field in CHANNEL_FIELDS}) write_csv(output_dir / "温度通道质控.csv", clean_rows(qc), QC_FIELDS, {field: field for field in QC_FIELDS}) same_stage = [{**row} for row in features] write_csv(output_dir / "同级温差特征.csv", clean_rows(same_stage), SAME_STAGE_FIELDS, {field: field for field in SAME_STAGE_FIELDS}) short = [{**row} for row in features] write_csv(output_dir / "温度短时特征.csv", clean_rows(short), SHORT_FIELDS, {field: field for field in SHORT_FIELDS}) def run(rows: list[dict], config: UnitConfig, thresholds: dict | None = None, min_hours: int = 1): merged = {**config.thresholds, **(thresholds or {})} valid = [row for row in rows if row.get("valid")] bad_channels, qc = channel_qc(valid, config) anomalies = mark_channel_anomalies(valid, config, merged) channel_segments = build_channel_segments(anomalies, merged) valid = _assign_cycles(valid, config) features = compute_features(valid, config, bad_channels, merged) confirm_levels(features, merged) segments = build_segments(features, config, channel_segments, merged, min_hours) observations = build_observations(features, segments, config) hf_segments = build_hf_segments(features, config, merged) cycles = summarize_cycles(features, segments, config) return qc, features, segments, cycles, channel_segments, observations, hf_segments __all__ = ["run", "write_results"]