#!/usr/bin/env python3 # -*- coding: utf-8 -*- r"""部件历史生成端 —— 补上"包内没有生成端"的 `m5_cms_tcm/component_history.json` (用户令 2026-09-19)。 ## 为什么要写它 用户令：「**所有的计算 均要形成 观澜的源代码**，确保 观澜 在别的电脑安装后，系统运行正常。」 `component_history.json` 是**随包快照**（族表 kind=shipped、gen=None，见 `scripts/products_reverse_audit.py::HUMAN_SNAPSHOTS`），随包件里的"在升/闭环/换件日期"含人工裁决与 厂家锚，**不是测量数据的函数**。于是另一台机器上装完、放数据、重算之后： `src/ontology/trend_ingest.py` 拿不到在升/闭环证据（图谱里两类 Evidence 全缺）、 `scripts/windscada_serve.py` 的决策链盘「振动在升/闭环」两格永远空。 本脚本把它**从已有产物算出来**（CMS 窗索引 + 工单台账），口径写在下面与产物 meta 里。 ## 产物与消费者 (字段契约) outputs/<场>/m5_cms_tcm/component_history.json meta built / windows / segments / segments_kind / 口径 / ★数据边界 components[] turbine, component, n_windows, n_replace, series, events summary['★当前在升 (末三窗/早三窗 ≥1.3)'] ← src/ontology/trend_ingest.py 逐行读 turbine, component, ratio, early, late, n_events, 工单可对 summary['★换件闭环 (检出→换件→恢复, 算法端到端验证)'] ← 同上 + scripts/windscada_serve.py 链盘 turbine, component, ratio, pre, post, replace_date, title, verdict summary[非★键] 消费端只认 ★ 前缀的键 ⇒ 非★键是给人/给诊断看的, 不被误读 ★ `component` 必须是 `src/ontology/trend_ingest.py::COMP_MAP` 里的中文名 (齿轮箱/发电机/主轴承/变桨/偏航/变流器)， 否则该行被消费者跳过。本器只出**齿轮箱/发电机/主轴承**三类 —— CMS 索引里只有这三类的测点 (Main_bearing_* / Gear_* / Generator_*)，变桨/偏航/变流器在本产物无振动测点，不臆造。 ## 分段 (segment) 口径 —— 两种模式, 自动选 参考口径是**导出窗**为段 (末三窗/早三窗; 健康期窗中位)。本机实况 (2026-09-19 实测)： `src/windcms/config.py::FARMS['rudong']['windows']` 只解析出**一个**窗 `w0316` (`outputs/rudong/m5_cms_tcm/windows/w0316/index.parquet`，2,066,686 行 × 54 列， trigger_time 2026-03-16 17:27 → 2026-04-21 10:18)。窗数 < 3 ⇒ 自动降为 **窗内周段** (ISO 周, 2026-W12 … 2026-W17 共 6 段)。两种模式都在本文件里实现： len(windows) >= 3 segments_kind='window' 段名=窗名, 段日=该窗 trigger_time 最早日 len(windows) < 3 segments_kind='week' 段名=ISO 周 (2026-W12), 段日=该周周一 **多窗口径未在本机取数验证**（本机只有 1 窗 ⇒ 该分支只有单测/纸面正确性，不给"已验证"的说法）。 ## 数据现实与诚实边界 (不造数) * **`★当前在升` 本机必然是空表**：判据要求"末三窗/早三窗"且**多窗口径**；本机 1 窗 ⇒ 不可算。 ★ 不降门槛、不换判据去凑一行。作为补充, 出**非★**键 `窗内周段趋势 (单窗补充口径, 仅参考不作判据)`：同一条比值算式 (末三段/早三段, 要求 6 段以取到 **不相交**的早/晚两组) 在周段上跑出来是什么, 就写什么。 * **`★换件闭环` 本机为空**：闭环要求"该台该部件的更换/换件工单，且段跨度里**既有段在换件日之前、 又有段在其后**"（前段须**结束**在工单日之前、后段须**开始**在工单日之后 —— 工单所在那一段两边都不算， 否则换后数据会混进"换前"，实测污染过 3 条）。工单台账 (`outputs/<场>/windscada/workorders.parquet`, 5,876 行) 实测跨 **2020-01-03 → 2026-07-08**，其中传动链件 903 行（去重后）/换件类 12 条 —— 但**全部落在段跨度 (2026-03-16~04-21) 之外** (换件工单最晚 2025-10-14)，窗内 0 条 ⇒ 无"换件前段"可比 ⇒ 空表。空表本身就是结论。 * 台账有**逐字重复行**（实测 2026-01-03 环网柜那条同一台同一分钟 7 行；换件类里 WTG23 2024-09-12 发电机轴承更换 4 行）⇒ 本器按 (台,日,标题,部件) 去重后再计数，原始/去重两个数都写进 meta.台账。 * 段内样本数一并写进 series 的 `n` —— 周段只有 2~142 个标量记录 (中位 14), 中位数据此散布, 读者可自行判断。 ## 用法 python scripts/component_history_build.py # 算并写盘 + 自登记 python scripts/component_history_build.py --dry-run # 只算不写, 打印漏斗 python scripts/component_history_build.py --farm rudong # 场名 (默认 rudong) """ from __future__ import annotations import argparse import json import pathlib import re import sys import time ROOT = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) # ★GBK 控制台/日志重定向不再因 '²' '⇒' 这类字符抛 UnicodeEncodeError (2026-09-19 远端实逮: # `baseline_38_build.py` 打印 'm/s²' 时在 cp936 下 rc=1, 整条恢复链少一件产物)。 # 只把**不可编码字符替换掉**, 不改流编码 —— 中文照旧可读。 try: sys.stdout.reconfigure(errors='replace') except Exception: pass import numpy as np # noqa: E402 import pandas as pd # noqa: E402 import pyarrow.parquet as pq # noqa: E402 from src import paths as P # noqa: E402 from src.derived_manifest import record as dm_record # noqa: E402 from src.windscada.config import farm as ws_farm # noqa: E402 (场名/台数: windscada 侧) from src.windcms import data as CD # noqa: E402 (窗发现: 唯一取用口) from src.windcms.config import KEY_SCALARS, SENSORS, SENSOR_CN # noqa: E402 from src.windcms.config import farm as cms_farm # noqa: E402 (windcms 场配置: FARMS[name]) # ── 标量集 (bounded): KEY_SCALARS 去掉 CrestFactor ──────────────────────────────── # 为什么去 CrestFactor: `src/windcms/report.py` 明写"Kurtosis/CrestFactor 判别力弱(回测实证, 不展示)"; # 且随包旧件的 fleet 键集合恰好 = {Kurtosis, Peak, Rms_HP, Rms_Vel} ∪ {iso_rms_vel (发电机), # rms_200 (主轴承)} —— 本器沿用同一集合, 便于与旧件对拍。 MEAS_SERIES = [m for m in KEY_SCALARS if m != 'CrestFactor'] # ── 部件 ↔ 测点 ──────────────────────────────────────────────────────────────── # 依据 `src/windcms/config.py`: 8 测点 = 2 主轴承 + 4 齿轮箱 + 2 发电机。 # `System Monitor` (磁盘/内存/健康度) 不是部件测点 ⇒ 排除 (它连 SENSORS 都不在)。 COMP_OF_SENSOR = { 'Main_bearing_front': '主轴承', 'Main_bearing_rear': '主轴承', 'Gear_planet': '齿轮箱', 'Gear_IMS': '齿轮箱', 'Gear_HS_rotor_side': '齿轮箱', 'Gear_HS_generator_side': '齿轮箱', 'Generator_DE': '发电机', 'Generator_NDE': '发电机', } # 段数门: 单窗口径下"末三/早三"需要 ≥3 段 (与旧件判据同名同门槛); 周段补充口径另要求 ≥6 段 # (早三/末三**不相交**, 否则末三段里含早三段的中间段, 比值被自己稀释)。 MIN_SEG_WINDOW = 3 MIN_SEG_WEEK = 6 RATIO_HI = 1.3 # 在升判据 (与旧件同名同值): 末/早 ≥ 1.3 RECOVER_HI = 0.5 # 闭环"恢复"判据: 换件后/换件前 ≤ 0.5 FLAT_LO, FLAT_HI = 0.9, 1.1 # 负面对照分桶: 持平带 # ── 工单 → (部件, 是否更换) 规则 (见模块 docstring「诚实边界」) ───────────────────── WO_COMP_KEYS = (('齿轮箱', ('齿轮箱', '齿圈', '行星', '高速轴', '中间轴', '齿轮油')), ('主轴承', ('主轴承', '主轴')), ('发电机', ('发电机',))) PART_PAT = re.compile(r'轴承|齿圈|行星|高速轴|中间轴|箱体|发电机本体') # 耗材类词: 出现即**不算换件** (油/油脂/加注/油泵/滤芯/密封/传感器 —— 换了也不该改变振动) CONSUMABLE_PAT = re.compile(r'油脂|润滑|油位|油泵|滤芯|密封|加注|油管') REPLACE_PAT = re.compile(r'更换|替换|换件') EVENT_CAP = 8 # 每 (台, 部件) 落进 events 的最多工单条数 (最近 N 条), 总数另存 n_events WANT_COLS = ['turbine', 'sensor_name', 'meas_name', 'trigger_time', 'scalar_value', 'ds_size'] # ── 工具 ─────────────────────────────────────────────────────────────────────── def _r6(v): """6 位有效数字的浮点 (0.0028182 → 0.0028182; 4.630123 → 4.63012)。""" if v is None: return None v = float(v) if not np.isfinite(v): return None return float(f'{v:.6g}') def _r3(v): if v is None: return None v = float(v) return float(f'{v:.3g}') if np.isfinite(v) else None def _py(o): """numpy 标量 → python 标量; NaN/Inf → None (JSON 严格可解析, 见 json.dumps(allow_nan=False))。""" if isinstance(o, dict): return {k: _py(v) for k, v in o.items()} if isinstance(o, (list, tuple)): return [_py(v) for v in o] if isinstance(o, (np.integer,)): return int(o) if isinstance(o, (np.floating, float)): f = float(o) return f if np.isfinite(f) else None if isinstance(o, (np.bool_,)): return bool(o) return o def load_scalars(cfg, meas): """各 CMS 窗的标量子集 —— 口径同 `src/windcms/data.py::_read_index/load_scalars` (只取 ds_size==1 的标量行, 只取 SENSORS 内的测点), 但**不落缓存**: 本器只读, 不在产物仓里另生成件。 窗的覆盖区间另取 `src.windcms.data.window_spans` (逐窗读 trigger_time; 页面同源) —— 本器只筛了 6 个标量 × 8 测点, 单看筛后行会把窗起点说晚 (w0316 全表起 03-16 17:27, 筛后起 03-17)。 """ ws = CD.windows(cfg) spans = CD.window_spans(cfg) parts, used = [], {} for w, info in ws.items(): p = pathlib.Path(info['index']) names = set(pq.ParquetFile(p).schema_arrow.names) cols = [c for c in WANT_COLS if c in names] d = pd.read_parquet(p, columns=cols) for c in ('ds_size', 'scalar_value'): if c not in d.columns: d[c] = np.nan d = d[(d.ds_size == 1) & d.meas_name.isin(meas) & d.sensor_name.isin(SENSORS)].copy() d['window'] = w parts.append(d) t = pd.to_datetime(d['trigger_time'], errors='coerce') sp = spans.get(w) or {} used[w] = dict(index=P.rel(p), rows_scalar=int(len(d)), rows_all=int(sp.get('rows') or 0), span_min=str(sp.get('t_min') or '')[:10], span_max=str(sp.get('t_max') or '')[:10], seg_min=str(t.min())[:10], seg_max=str(t.max())[:10], turbines=int(d['turbine'].nunique())) if not parts: return pd.DataFrame(columns=WANT_COLS + ['window', 't', 'val']), used d = pd.concat(parts, ignore_index=True) d['t'] = pd.to_datetime(d['trigger_time'], errors='coerce') d['val'] = pd.to_numeric(d['scalar_value'], errors='coerce') d = d[d['val'].notna() & d['t'].notna()].copy() return d, used def mark_segments(d, n_win, span_min=None): """加 `seg` 列 + 返回 (kind, {段名: 段日起日}, 段序)。见 docstring「分段口径」。 段日: 多窗口径 = 该窗全表覆盖起点 (`window_spans`, 与页面同源); 单窗口径 = 该 ISO 周的**周一** (取周一而不是"该周首条记录日", 是让段日只依赖日历, 不随哪个测点先出数而漂)。 """ span_min = span_min or {} if n_win >= 3: # 多窗口径: 段 = 导出窗 d['seg'] = d['window'].astype(str) kind = 'window' else: # 单窗口径: 段 = 窗内 ISO 周 iso = d['t'].dt.isocalendar() d['seg'] = iso['year'].astype(int).astype(str) + '-W' + iso['week'].astype(int).astype(str).str.zfill(2) kind = 'week' g = d.groupby('seg')['t'].min() order = [s for s, _ in sorted(g.items(), key=lambda kv: kv[1])] span_max = str(d['t'].max())[:10] if kind == 'week': seg_date = {s: str(pd.Timestamp.fromisocalendar(int(s[:4]), int(s[6:]), 1).date()) for s in order} seg_end = {s: str((pd.Timestamp.fromisocalendar(int(s[:4]), int(s[6:]), 1) + pd.Timedelta(days=6)).date()) for s in order} else: seg_date = {s: str(span_min.get(s) or str(g[s])[:10])[:10] for s in order} seg_end = {s: str(d[d.seg == s]['t'].max())[:10] for s in order} return kind, seg_date, order, seg_end def seg_table(d): """(台, 测点, 标量, 段) 的段内中位/样本数。""" g = d.groupby(['turbine', 'sensor_name', 'meas_name', 'seg'])['val'].agg(['median', 'size']) g = g.reset_index().rename(columns={'median': 'med', 'size': 'n'}) return g def trend_table(sg, order, k=3, min_seg=3): """(台, 测点, 标量) 的「末 K 段中位 / 早 K 段中位」比 —— 只对段数 ≥ min_seg 的组合出数。""" rows = [] idx = {s: i for i, s in enumerate(order)} for (t, sen, meas), sub in sg.groupby(['turbine', 'sensor_name', 'meas_name']): m = sub.assign(_i=sub['seg'].map(idx)).dropna(subset=['_i']).sort_values('_i') if len(m) < min_seg: continue vals = m['med'].to_numpy(dtype=float) early, late = float(np.median(vals[:k])), float(np.median(vals[-k:])) if not (early > 0): continue rows.append(dict(turbine=t, sensor=sen, component=COMP_OF_SENSOR.get(sen, ''), measure=meas, early=_r6(early), late=_r6(late), ratio=_r3(late / early), n_seg=int(len(m)), segs=list(m['seg']))) return pd.DataFrame(rows) # ── 工单台账 ─────────────────────────────────────────────────────────────────── def load_workorders(store): """工单台账 → (逐行事件表, 台账范围说明)。台账不存在 ⇒ 空表 + 如实说明 (不造事件)。""" p = pathlib.Path(store) / 'workorders.parquet' if not p.exists(): return pd.DataFrame(columns=['turbine', 'date', 'title', 'replaced', 'comps', 'raw']), \ dict(path=P.rel(p), exists=False, rows=0, note='台账不在位 ⇒ 全部 events 为空数组, 不是"无工单"'), w = pd.read_parquet(p) w['t_report'] = pd.to_datetime(w.get('t_report'), errors='coerce') txt = (w.get('故障名称', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('故障描述', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('故障位置二级', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('故障位置三级', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('排查项目', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('故障原因', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('维修对象', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('元器件名称', pd.Series('', index=w.index)).fillna('')) act = w.get('维修动作', pd.Series('', index=w.index)).fillna('').astype(str) obj = (w.get('故障名称', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('维修对象', pd.Series('', index=w.index)).fillna('') + '｜' + w.get('元器件名称', pd.Series('', index=w.index)).fillna('')) recs = [] for i, s in txt.items(): comps = [c for c, keys in WO_COMP_KEYS if any(k in s for k in keys)] if not comps: continue o = str(obj.iloc[i] if hasattr(obj, 'iloc') else '') repl = bool(REPLACE_PAT.search(act.iloc[i])) and bool(PART_PAT.search(o)) and not CONSUMABLE_PAT.search(o) title = str(w.get('故障名称', pd.Series('', index=w.index)).iloc[i] or '').strip() recs.append(dict(turbine=str(w['turbine'].iloc[i]), date=str(w['t_report'].iloc[i])[:10], title=title[:60], replaced=repl, comps=comps, act=str(act.iloc[i]), obj=o[:60])) ev = pd.DataFrame(recs) # ★ 台账里有**逐字重复行** (实测: 2026-01-03 环网柜 V 柜出线电缆接头三相击穿 同一台同一分钟 7 行; # 换件类里 WTG23 2024-09-12 发电机轴承更换 4 行)。不去重会让 n_replace / n_events 虚高。 n_raw = int(len(ev)) if n_raw: ev['_k'] = ev['turbine'] + '|' + ev['date'] + '|' + ev['title'] + '|' + ev['comps'].map(lambda L: ','.join(L)) ev = ev.drop_duplicates(subset=['_k']).drop(columns=['_k']).reset_index(drop=True) tv = w['t_report'].dropna() info = dict(path=P.rel(p), exists=True, rows=int(len(w)), span=[str(tv.min())[:10], str(tv.max())[:10]] if len(tv) else ['', ''], n_transmission=int(len(ev)), n_transmission_raw=n_raw, n_replace=int(ev['replaced'].sum()) if len(ev) else 0, replace_span=[str(pd.to_datetime(ev[ev.replaced].date).min())[:10], str(pd.to_datetime(ev[ev.replaced].date).max())[:10]] if len(ev) and ev.replaced.any() else None) return ev, info def events_of(ev, turbine, comp): """该 (台, 部件) 的工单事件 (按台/部件关键字映射) → (最近 EVENT_CAP 条明细, 总条数, 其中换件条数)。 ★ 总条数与换件条数在**全量**匹配上数 (明细只是最近的 EVENT_CAP 条): 早期版本用明细数当总数, 一旦该台该部件工单超过 8 条, n_replace 就会**少报**。 """ if not len(ev) or 'comps' not in ev.columns: return [], 0, 0 m = ev[ev['comps'].map(lambda L: comp in L) & (ev['turbine'] == turbine)].sort_values('date') out = [dict(turbine=turbine, comp=comp, date=r['date'], src='工单', title=r['title'], replaced=bool(r['replaced'])) for _, r in m.tail(EVENT_CAP).iterrows()] return out, int(len(m)), int(m['replaced'].sum()) def build(cfg, farm_name='rudong', dry=False): m5 = P.m5(farm_name) store = P.store(farm_name) t0 = time.time() meas = MEAS_SERIES d, used = load_scalars(cfg, meas) if not len(d): print('[X] 一个 CMS 窗索引都没读到 —— 先放数据并跑重算链 (scripts/rebuild_all.py); 本器不写空产物。') return None, 4 kind, seg_date, order, seg_end = mark_segments(d, len(used), {w: v['span_min'] for w, v in used.items()}) sg = seg_table(d) ev, wo_info = load_workorders(store) # 台名单以场配置为准 (38 台), 数据里没有的台如实留空 series (不静默丢台) turbines = [t for t in ws_farm(farm_name).get('turbines', [])] or sorted(d['turbine'].unique()) no_data = [t for t in turbines if t not in set(d['turbine'].unique())] # 数据覆盖区间取**全表** (src.windcms.data.window_spans, 与页面同源): 只筛 6 标量 × 8 测点会把窗起点说晚一天 span = [min(v['span_min'] for v in used.values()), max(v['span_max'] for v in used.values())] print(f'部件历史生成端 · 场={farm_name}') print(f' 窗 {len(used)} 个: ' + '; '.join( f'{w}(全表 {v["span_min"]}→{v["span_max"]}, {v["rows_all"]}行; 标量 {v["rows_scalar"]}行)' for w, v in used.items())) print(f' 标量行 (ds_size==1 ∧ {len(meas)} 标量 ∧ {len(SENSORS)} 测点): {len(d)}' + (f' · 场配置 {len(turbines)} 台, 无数据的台: {no_data}' if no_data else f' · 台数 {len(turbines)}')) print(f' 分段口径: segments_kind={kind} · 段 {len(order)} 个: {order[0]}…{order[-1]} ' f'(段日起 {seg_date[order[0]]}→{seg_date[order[-1]]}; 数据覆盖 {span[0]}→{span[1]})') print(f' 段内样本 (台×测点×标量×段) 中位: {int(sg["n"].median())} · 最少 {int(sg["n"].min())} · 最多 {int(sg["n"].max())}') print(f' 工单台账: {wo_info["path"]} 行={wo_info["rows"]} 跨度={wo_info.get("span")} ' f'传动链件={wo_info.get("n_transmission")} 换件={wo_info.get("n_replace")} ' f'换件跨度={wo_info.get("replace_span")}') if len(ev): print(' 换件类工单 (全量, 判据 candidate): ' + '; '.join( f'{r["turbine"]} {"/".join(r["comps"])} {r["date"]} {r["title"][:18]}' for _, r in ev[ev['replaced']].sort_values('date').iterrows())) # ── trend: 单窗口径的末三/早三 (判据) 与 周段的末三/早三 (补充, 不相交) ── if kind == 'window': tw = trend_table(sg, order, k=3, min_seg=MIN_SEG_WINDOW) else: tw = trend_table(sg, order, k=3, min_seg=MIN_SEG_WEEK) tw = tw.sort_values('ratio', ascending=False) if len(tw) else tw wo_ok = bool(wo_info.get('exists') and wo_info.get('span') and wo_info['span'][0] and wo_info['span'][0] <= span[1] and wo_info['span'][1] >= span[0]) # ── components ── comps_out = [] for t in turbines: for comp in ('主轴承', '齿轮箱', '发电机'): sens = [s for s, c in COMP_OF_SENSOR.items() if c == comp] series = {} n_seg_c = 0 for sen in sens: for ms in meas: sub = sg[(sg.turbine == t) & (sg.sensor_name == sen) & (sg.meas_name == ms)] if not len(sub): continue m = sub.set_index('seg')['med'].reindex(order).dropna() if not len(m): continue base = float(np.median(m.to_numpy(dtype=float))) if not (base > 0): continue entries = [] for s in order: if s not in m.index: continue v = float(m.loc[s]) n = int(sub.loc[sub.seg == s, 'n'].iloc[0]) entries.append(dict(seg=s, med=_r6(v), x_self=_r3(v / base), n=n)) series[f'{sen}|{ms}'] = entries n_seg_c = max(n_seg_c, len(entries)) e_list, n_ev, n_repl = events_of(ev, t, comp) comps_out.append(dict(turbine=t, component=comp, n_windows=int(n_seg_c), n_events=n_ev, n_replace=n_repl, events_capped=bool(n_ev > len(e_list)), series=series, events=e_list)) # ── summary ── # ★判据一: 只认多窗口径 (本机必然空) —— 不降门槛凑数 win_trend = trend_table(sg, order, k=3, min_seg=MIN_SEG_WINDOW) if kind == 'window' else pd.DataFrame() rising = [] if kind == 'window' and len(win_trend): for (t, comp), sub in win_trend.groupby(['turbine', 'component']): w = sub.sort_values('ratio', ascending=False).iloc[0] if w['ratio'] >= RATIO_HI: _, n_ev, _ = events_of(ev, t, comp) rising.append(dict(turbine=t, component=comp, ratio=float(w['ratio']), early=float(w['early']), late=float(w['late']), n_events=n_ev, 工单可对=wo_ok, sensor=w['sensor'], measure=w['measure'], n_windows=int(w['n_seg']))) rising = sorted(rising, key=lambda r: -r['ratio']) # 补充诊断 (非★, 消费者忽略): 周段末三/早三 —— 逐行带段内最小样本数, 让"薄段噪声"可见 diag = [] if len(tw): nmin = sg.groupby(['turbine', 'sensor_name', 'meas_name'])['n'].min() for _, r in tw[tw['ratio'] >= RATIO_HI].iterrows(): _, n_ev, _ = events_of(ev, r['turbine'], r['component']) diag.append(dict(turbine=r['turbine'], component=r['component'], sensor=r['sensor'], measure=r['measure'], early=float(r['early']), late=float(r['late']), ratio=float(r['ratio']), n_seg=int(r['n_seg']), seg_n_min=int(nmin.get((r['turbine'], r['sensor'], r['measure']), 0)), n_events=n_ev, 工单可对=wo_ok)) diag_all = len(diag) diag = diag[:20] # ★判据二: 换件闭环 —— 需要"换件日之前有段 ∧ 之后有段" loop, loop_skip, repl_pairs = [], [], [] repl = ev[ev['replaced']] if len(ev) and 'replaced' in ev.columns else ev fusion = {} fp = m5 / 'fusion_38.csv' if fp.exists(): # 检出候选旗标 (融合/CMS 面), 只作 verdict 的佐证 fu = pd.read_csv(fp) for _, r in fu.iterrows(): fusion[str(r.get('台'))] = dict(融合=str(r.get('融合') or ''), CMS=str(r.get('CMS') or '')) for _, r in (repl.iterrows() if len(repl) else []): for comp in r['comps']: repl_pairs.append((r['turbine'], comp, r['date'])) pre_segs = [s for s in order if seg_end[s] < r['date']] # 段**结束**在工单日之前 post_segs = [s for s in order if seg_date[s] > r['date']] # 段**开始**在工单日之后 if not pre_segs or not post_segs: loop_skip.append(dict(turbine=r['turbine'], component=comp, date=r['date'], title=r['title'], why=f'段跨度 {span[0]}~{span[1]} 内' + ('无换件前段' if not pre_segs else '无换件后段'))) continue sens = [s for s, c in COMP_OF_SENSOR.items() if c == comp] cand = [] for sen in sens: for ms in meas: sub = sg[(sg.turbine == r['turbine']) & (sg.sensor_name == sen) & (sg.meas_name == ms)] if not len(sub): continue a = sub[sub.seg.isin(pre_segs)]['med'] b = sub[sub.seg.isin(post_segs)]['med'] if len(a) and len(b) and float(np.median(a)) > 0: cand.append((ms, sen, float(np.median(a)), float(np.median(b)))) if not cand: loop_skip.append(dict(turbine=r['turbine'], component=comp, date=r['date'], title=r['title'], why='换件前后段内该部件无可用标量')) continue ms, sen, pre, post = max(cand, key=lambda c: c[2]) # 取换前幅值最大者 (最不利/最有分辨力) ratio = post / pre f = fusion.get(r['turbine']) or {} hit = bool(f) and (f.get('融合') not in ('', '正常') or '红' in f.get('CMS', '') and '红0' not in f.get('CMS', '')) if ratio <= RECOVER_HI: verdict = '恢复' if hit else '回落 (非检出台, 不作端到端实证)' elif ratio < FLAT_LO: verdict = '部分回落' elif ratio <= FLAT_HI: verdict = '持平' else: verdict = '未降/上升' loop.append(dict(turbine=r['turbine'], component=comp, ratio=_r3(ratio), pre=_r6(pre), post=_r6(post), replace_date=r['date'], title=r['title'], verdict=verdict, sensor=sen, measure=ms, 检出候选=hit, 融合=(f.get('融合') or None), pre_segs=pre_segs, post_segs=post_segs)) loop = sorted(loop, key=lambda r: r['ratio']) # 负面对照: 工单后振动未降/持平 (非换件的传动链工单, 换前后各有段) # 不可算的原因**分桶计数**: 单看一个 1512 的总数会把"工单在窗外"和"窗内但无前后段"混成一团。 neg_rows, neg_noc = [], dict(工单在段跨度外=0, 窗内但无前段或后段=0, 该部件无标量=0) nonrepl = ev[~ev['replaced']] if len(ev) and 'replaced' in ev.columns else ev for _, r in (nonrepl.iterrows() if len(nonrepl) else []): for comp in r['comps']: # ★ 前/后段都**不含工单所在那一段**: 段界线按"段结束 < 工单日"与"段开始 > 工单日"判, # 否则工单落在段中间时, 该段自己会被当成"换前", 把换后的数据混进 pre (实测污染过 3 条)。 pre_segs = [s for s in order if seg_end[s] < r['date']] post_segs = [s for s in order if seg_date[s] > r['date']] if r['date'] < seg_date[order[0]] or r['date'] > seg_end[order[-1]]: neg_noc['工单在段跨度外'] += 1 continue if not pre_segs or not post_segs: neg_noc['窗内但无前段或后段'] += 1 continue sens = [s for s, c in COMP_OF_SENSOR.items() if c == comp] cand = [] for sen in sens: for ms in meas: sub = sg[(sg.turbine == r['turbine']) & (sg.sensor_name == sen) & (sg.meas_name == ms)] if not len(sub): continue a = sub[sub.seg.isin(pre_segs)]['med'] b = sub[sub.seg.isin(post_segs)]['med'] if len(a) and len(b) and float(np.median(a)) > 0: cand.append((ms, sen, float(np.median(a)), float(np.median(b)))) if not cand: neg_noc['该部件无标量'] += 1 continue ms, sen, pre, post = max(cand, key=lambda c: c[2]) neg_rows.append(dict(turbine=r['turbine'], component=comp, date=r['date'], title=r['title'], sensor=sen, measure=ms, pre=_r6(pre), post=_r6(post), ratio=_r3(post / pre))) bucket = dict(下降=sum(1 for r in neg_rows if r['ratio'] < FLAT_LO), 持平=sum(1 for r in neg_rows if FLAT_LO <= r['ratio'] <= FLAT_HI), 上升=sum(1 for r in neg_rows if r['ratio'] > FLAT_HI)) neg_noc_total = sum(neg_noc.values()) n_comp_rows = sum(1 for c in comps_out for _ in c['series']) win_txt = '; '.join(f'{w} {v["span_min"]}~{v["span_max"]}' for w, v in used.items()) n_pair_inspan = sum(1 for _, _, dt in repl_pairs if span[0] <= dt <= span[1]) 判读 = ( f'分段口径 segments_kind={kind}: 本机只解析出 {len(used)} 个导出色窗 ({win_txt}) ' f'⇒ 以窗内 ISO 周段为分段 ({len(order)} 段 {order[0]}…{order[-1]}, {span[0]}~{span[1]}); ' f'多窗 (≥3) 时自动改回多窗口径。' f'★当前在升 (末三窗/早三窗 ≥{RATIO_HI}) 本机{("有 " + str(len(rising)) + " 行") if rising else "不可算 ⇒ 空表"}: ' f'该判据要求多窗口径且 ≥{MIN_SEG_WINDOW} 窗, 本机 {len(used)} 窗。' f'★换件闭环 本机{("有 " + str(len(loop)) + " 行") if loop else "不可算 ⇒ 空表"}: 要求换件工单在段跨度内' f'且其前后各有至少一段; 台账 {wo_info.get("span")} 共 {wo_info.get("n_transmission")} 条传动链工单, ' f'其中换件类 {wo_info.get("n_replace")} 条 (跨度 {wo_info.get("replace_span")}) 展开为 {len(repl_pairs)} 个 ' f'(台,部件) 组合, 段跨度内 {n_pair_inspan} 个 ⇒ 成闭环 {len(loop)} 行, 其余 {len(loop_skip)} 个因' f'"无换件前段/后段"逐条记入 meta.未成闭环。' f'补充口径 (非判据): 周段末三/早三比值 ≥{RATIO_HI} 的候选 {diag_all} 条 (列出前 {len(diag)} 条); ' f'周段每 (台,测点,标量) 只有 {int(sg["n"].min())}~{int(sg["n"].max())} 条标量记录/段 ' f'(中位 {int(sg["n"].median())}) ⇒ 该比值只说明窗内走向, 不作在升判据 (明细逐行给出 seg_n_min)。' f'负面对照: 传动链非换件工单里换前后各有段的 {len(neg_rows)} 条可算 (下降 {bucket["下降"]} · 持平 {bucket["持平"]} ' f'· 上升 {bucket["上升"]}), 其余 {neg_noc_total} 条不可算 ({neg_noc})。') seg_kind_txt = '导出窗' if kind == 'window' else f'窗内 ISO 周 (本机窗数 {len(used)} < 3 ⇒ 降级; 多窗时自动改回)' meta = dict( built=time.strftime('%Y-%m-%d'), farm=farm_name, by='scripts/component_history_build.py', windows={w: v['span_min'] for w, v in used.items()}, windows_detail={w: dict(rows_all=v['rows_all'], rows_scalar=v['rows_scalar'], span_min=v['span_min'], span_max=v['span_max'], seg_min=v['seg_min'], seg_max=v['seg_max'], turbines=v['turbines']) for w, v in used.items()}, segments_kind=kind, segments=seg_date, n_segments=len(order), 来源声明=('本文件由 scripts/component_history_build.py **现算**写成: 输入 = ' 'outputs/<场>/m5_cms_tcm/windows/*/index.parquet (data/raw 重算产物, 只读 ds_size==1 标量行) ' '+ outputs/<场>/windscada/workorders.parquet (工单台账) ' '+ outputs/<场>/m5_cms_tcm/fusion_38.csv (仅作"检出候选"佐证)。' '不含随包旧件 (人工裁决/厂家锚) 的数值搬运; 算不出来的 (在升/闭环) 一律空表 + 边界说明, 不填估值。' 'components[].n_windows = 该部件的**段数** (segments_kind="week" 时即窗内周段数, 不是导出窗数)。'), 口径=(f'标量: CMS 窗索引 ds_size==1 行, 标量集 = KEY_SCALARS 去 CrestFactor ({", ".join(meas)}); ' f'测点 → 部件: Main_bearing_*→主轴承, Gear_*→齿轮箱, Generator_*→发电机 (System Monitor 非部件测点, 排除)。' f'段: {seg_kind_txt}。' f'series 值 = 该段该 (测点|标量) 标量**中位**; 归一 x_self = 段中位 / 该台该测点该标量**全段中位数** ' f'(随包旧件同口径, 旧件写作 x_self; 本件不出 x_fleet —— 那需要跨台同工况对照)。' f'在升判据 = 末三窗中位 / 早三窗中位 ≥ {RATIO_HI}, 需多窗口径且 ≥{MIN_SEG_WINDOW} 窗。' f'闭环判据 = 换件工单日之前有段 ∧ 之后有段; pre/post = 该部件换前/后段中位 (取换前幅值最大的测点|标量), ' f'verdict: ratio ≤ {RECOVER_HI} 且该台为检出候选 ⇒ 恢复, 否则按 部分回落/持平/未降 如实记。' f'工单口径: 部件由关键词映射 (齿轮箱/齿圈/行星/高速轴/中间轴/齿轮油 → 齿轮箱; 主轴承/主轴 → 主轴承; ' f'发电机 → 发电机); replaced = 维修动作含 更换|替换|换件 ∧ 目标文本含部件词 ' f'(轴承|齿圈|行星|高速轴|中间轴|箱体|发电机本体) ∧ 目标文本无耗材词 ' f'(油脂|润滑|油位|油泵|滤芯|密封|加注|油管) —— 换油/换传感器不算换件。'), inputs=dict(scalars=[v['index'] for v in used.values()], workorders=wo_info.get('path'), fusion=P.rel(fp) if fp.exists() else None), 未成闭环=loop_skip[:40], 未成闭环总数=len(loop_skip), 台账=wo_info, ) meta['★数据边界'] = ( f'① 本机只有 {len(used)} 个 CMS 导出色窗 ({win_txt}) ⇒ 在升 (末三窗/早三窗) 与 多窗口径的健康期分段' f'**都不可算** ⇒ summary["★当前在升…"] 为空表; 本件以窗内周段 ({len(order)} 段) 作补充诊断, ' f'只描述窗内走向, 不作判据。' f'② 换件闭环需要"换件日前后各有段": 工单台账 {wo_info.get("path")} 跨度 {wo_info.get("span")} ' f'(传动链件 {wo_info.get("n_transmission")} 条, 其中换件类 {wo_info.get("n_replace")} 条, ' f'跨度 {wo_info.get("replace_span")}), 与段跨度 {seg_date[order[0]]}~{seg_date[order[-1]]} **无交叠** ' f'⇒ summary["★换件闭环…"] 为空表。★ 这两张空表是数据边界, 不是"没有在升机组/没有闭环案例"; ' f'站点需 (a) 补导 ≥3 个 CMS 色窗 (尤其换件日前后各一窗), (b) 补 2026 年传动链换件工单, ' f'二者到位后本脚本自动出数。' f'③ 段内样本薄: 周段每 (台,测点,标量) 仅 {int(sg["n"].min())}~{int(sg["n"].max())} 条标量记录 ' f'(中位 {int(sg["n"].median())}), 中位数据此散布 —— series 的 n 逐条给出, 读者自行判断可信度。') meta['★换件断点'] = '换件前后不可混算趋势; 本表在 events 里标 replaced=true, 读者须自行分段读 series' payload = dict( meta=meta, components=_py(comps_out), summary=_py({ '★换件闭环 (检出→换件→恢复, 算法端到端验证)': loop, '★当前在升 (末三窗/早三窗 ≥1.3)': rising, '窗内周段趋势 (单窗补充口径, 仅参考不作判据)': diag, '窗内周段趋势·候选总数': diag_all, '工单后振动未降/持平 (多为油位/传感器/加注类, 非轴承更换 ⇒ 振动本不该变)': dict( 可算=len(neg_rows), 不可算=neg_noc_total, 不可算原因=neg_noc, 分桶=bucket, 口径='逐条传动链非换件工单: 换件日前后各有至少一段 ⇒ 可算; 指标取该部件换前幅值最大的 (测点|标量) 的 ' f'post/pre 比值; 分桶 下降<{FLAT_LO} · 持平 {FLAT_LO}~{FLAT_HI} · 上升>{FLAT_HI}', 明细=sorted(neg_rows, key=lambda r: -r['ratio'])[:12]), '判读': 判读}), ) txt = json.dumps(payload, ensure_ascii=False, indent=1, allow_nan=False, default=str) out = m5 / 'component_history.json' print(f'\n 段表行 {len(sg)} · 部件行 {len(comps_out)} · series 键 {n_comp_rows} · ' f'events {sum(len(c["events"]) for c in comps_out)} 条') print(f' summary: ★换件闭环 {len(loop)} 行 · ★当前在升 {len(rising)} 行 · ' f'周段候选 {diag_all} 条(列 {len(diag)}) · 负面对照可算 {len(neg_rows)}/不可算 {neg_noc_total} {neg_noc}') if neg_rows: print(' 负面对照明细: ' + '; '.join(f'{r["turbine"]} {r["component"]} {r["date"]} {r["title"][:16]} ' f'{r["sensor"]}|{r["measure"]} {r["ratio"]}×' for r in sorted(neg_rows, key=lambda r: -r['ratio']))) if diag: print(' 周段候选 top: ' + '; '.join(f'{r["turbine"]} {r["component"]} {r["sensor"]}|{r["measure"]} ' f'{r["ratio"]}× ({r["early"]}→{r["late"]}, {r["n_seg"]}段)' for r in diag[:5])) if loop_skip: print(f' 换件工单未成闭环 {len(loop_skip)} 条, 例: ' + '; '.join(f'{s["turbine"]} {s["component"]} {s["date"]} {s["why"]}' for s in loop_skip[:3])) print(f' 产物体积预估 {len(txt.encode("utf-8")) / 1024:.0f} KB ({P.rel(out)}) · 用时 {time.time() - t0:.1f}s') if dry: print('\n[dry-run] 不写盘、不登记。') return payload, 0 m5.mkdir(parents=True, exist_ok=True) out.write_text(txt, encoding='utf-8') dm_record(P.out_root(farm_name), {out.relative_to(P.out_root(farm_name)).as_posix():'scripts/component_history_build.py (CMS 窗标量 ds_size==1: 段中位/x_self 序列 + ' '工单台账事件; 在升/闭环判据与空表情形见产物 meta.★数据边界)'}, by='scripts/component_history_build.py') print(f'\n[OK] 已写 {P.rel(out)} ({out.stat().st_size / 1024:.0f} KB) · 已自登记') return payload, 0 def main() -> int: ap = argparse.ArgumentParser(description='部件历史生成端 (m5_cms_tcm/component_history.json)') ap.add_argument('--farm', default='rudong') ap.add_argument('--dry-run', action='store_true', help='只算不写盘 (打印漏斗与体积预估)') args = ap.parse_args() cfg = cms_farm(args.farm) _, rc = build(cfg, args.farm, dry=args.dry_run) return rc if __name__ == '__main__': sys.exit(main())