#!/usr/bin/env python3 # -*- coding: utf-8 -*- r"""从 data/raw 重算/更新产物 (2026-09-11) —— 维护页那四条「摄入命令」的总入口。 ## 背景 `docs/说明书…v0.2.md` §11 写着「**不含**从原始数据重生成产物的链 (P0)」, 于是随包产物只能吃预生成的那份 (台账止 2024-11-21、报警止 2026-07-15), 现场把新表放进 `data/raw/<场站>/…` 也不会更新。 本脚本把这条链补上, 顺序与依赖关系如下 (都按场配置 `src.windscada.config`): ① scripts/windscada_alarms_ingest.py 故障报警/*.xls → alarms.parquet ② scripts/windscada_workorder_ingest.py 风机故障记录/** → workorders.parquet ③ scripts/windscada_watch_channels_build.py 油样报告/** → oil_samples_index.parquet ── 以上三步只吃现场台账 (xls/xlsx/pdf), 快, 秒级到分钟级 ── ④ (--scada) 包内已有的 SCADA 侧构建器: 从 scada_10min/*.csv 重算 powercurve → loss_monthly(availability) → curves / control / stop_events(faults) / temp_bins / yaw_daily / hydraulic_accum / thermal_chain / system_aux (这一步要逐台读 38 个 ~380MB 的 CSV, 慢; 默认不跑, 加 --scada 才跑) ## --verify: 等价验收 (本链的验收标准) 重算件必须能**复现随包件**: 逐键逐值比对 `outputs/rudong/windscada/_pre_rebuild_20260911/` 里那份 随包基线 (2026-09-11 现场修复时留的备份)。允许的差异只有两类, 都必须被点名: · 旧链的**已知缺陷**: 报警/工单里同一副本被收两遍、日期解析不了就整行丢弃、Excel 序列号当纳秒解析; · 本链的**新增覆盖**: 随包时没接的源件 (74 张台账表 vs 随包只用了 6 张)。 无法归类的差异 = 验收不通过 (退出码 4)。 用法: python scripts/rebuild_from_raw.py # 跑 ①②③ python scripts/rebuild_from_raw.py --verify # 只验收, 不重算 python scripts/rebuild_from_raw.py --scada # 连 ④ 一起跑 (慢) python scripts/rebuild_from_raw.py --dry-run # 只打印各步会做什么 """ from __future__ import annotations import sys as _sys, pathlib as _plb _sys.path.insert(0, str(_plb.Path(__file__).resolve().parents[1])) from src import paths as P import argparse import pathlib import subprocess import sys from collections import Counter ROOT = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) BASE = P.rel(P.store()) + '/_pre_rebuild_20260911' # 随包基线备份 STEPS = [ ('报警事件', 'scripts/windscada_alarms_ingest.py', 'alarms.parquet'), ('检修工单台账', 'scripts/windscada_workorder_ingest.py', 'workorders.parquet'), ('油液化验', 'scripts/windscada_watch_channels_build.py', 'oil_samples_index.parquet'), ] # SCADA 侧 (--scada): 全部是包内**已有**的构建器, 只是过去没有入口把它们串起来。 # 顺序 = 依赖顺序 (powercurve 先出 bins, availability 才吃得到; stop_events 要吃 alarms)。 SCADA_STEPS = [ ('功率曲线', 'src.windscada.perf.powercurve', 'store'), ('可用率与损失 (loss_monthly)', 'src.windscada.perf.availability', 'build'), ('七镜头曲线', 'src.windscada.perf.curves', 'build_store'), ('控制策略件', 'src.windscada.perf.control', 'build_store'), ('停机事件 (stop_events)', 'src.windscada.perf.faults', 'build_stop_events'), ('温度 NBM (temp_bins)', 'src.windscada.subsys.temp_nbm', 'build_store'), ('偏航 (yaw_daily)', 'src.windscada.subsys.yaw', 'build_store'), ('液压蓄能 (hydraulic_accum)', 'src.windscada.subsys.hydraulic', 'build_store'), ('热链 (thermal_chain)', 'src.windscada.subsys.thermal_chain', 'build_store'), ('系统辅助 (system_aux)', 'src.windscada.taxonomy', 'build_aux_store'), ] def run_scada(farm_name=None) -> int: """逐台读 scada_10min/*.csv 重算 SCADA 侧产物。慢 (38 台 × 各构建器), 失败不中断, 最后汇总。""" import importlib import time from src.windscada.config import farm cfg = farm(farm_name) print(f'场: {cfg["name"]} 源: {cfg["src_10min"]} 仓: {cfg["store"]}') bad = [] for label, mod, fn in SCADA_STEPS: t0 = time.time() try: f = getattr(importlib.import_module(mod), fn) r = f(cfg) print(f' ✔ {label:26s} {time.time()-t0:6.1f}s {str(r)[:110]}', flush=True) except Exception as e: bad.append((label, f'{type(e).__name__}: {e}')) print(f' ✘ {label:26s} {time.time()-t0:6.1f}s {type(e).__name__}: {str(e)[:150]}', flush=True) if bad: print(f'\n{len(bad)} 个构建器失败:') for label, err in bad: print(f' {label}: {err}') return 0 if not bad else 5 def _run(script: str, extra=()) -> int: cmd = [sys.executable, str(ROOT / script), *extra] print(f'\n$ {" ".join(cmd[1:])}', flush=True) return subprocess.call(cmd, cwd=str(ROOT)) def _canon(df, cols=None): import pandas as pd d = df[cols] if cols else df out = [] for r in d.itertuples(index=False): row = [] for c, v in zip(d.columns, r): try: if v is None or (not isinstance(v, str) and pd.isna(v)): row.append('') continue except (TypeError, ValueError): pass row.append(v.isoformat() if isinstance(v, pd.Timestamp) else str(v)) out.append(tuple(row)) return Counter(out) def verify() -> int: import pandas as pd store = P.store() base = ROOT / BASE if not base.is_dir(): print(f'[X] 没有随包基线备份 {BASE} —— 无法做等价验收') return 4 bad = 0 print('== 等价验收: 重算件 vs 随包基线 ==') for label, name, exact in (('报警事件', 'alarms.parquet', True), ('油液化验', 'oil_samples_index.parquet', True)): a = _canon(pd.read_parquet(base / name)) b = _canon(pd.read_parquet(store / name)) absent = set(a) - set(b) # 随包内容在重算里**找不到** (真缺才算不通过) gained = set(b) - set(a) collapsed = sum(a.values()) - sum(min(c, b.get(k, 0)) for k, c in a.items()) ok = exact and not absent and not gained print(f'\n {label} ({name})') print(f' 随包 {sum(a.values())} 行 ({len(a)} 种内容) / 重算 {sum(b.values())} 行 ({len(b)} 种内容)') print(f' 随包内容未复现 {len(absent)} 种; 重算新增 {len(gained)} 种; 副本归并 {collapsed} 行') print(f' {"✔ 完全一致 (行数、内容、副本数全同)" if ok else "✘ 需人工看"}') if not ok: bad += 1 for k in list(absent)[:2]: print(' 随包独有: ' + ' | '.join(x for x in k if x != '')[:180]) for k in list(gained)[:2]: print(' 重算独有: ' + ' | '.join(x for x in k if x != '')[:180]) # 工单: 逐项列账, 不用一个数字糊过去 from scripts.windscada_workorder_ingest import OUT_COLS a_raw = pd.read_parquet(base / 'workorders.parquet') b_raw = pd.read_parquet(store / 'workorders.parquet') six = set(a_raw.src_file.unique()) kc = [c for c in OUT_COLS if c != 'src_file'] a = _canon(a_raw[a_raw.src_file.isin(six)], kc) b = _canon(b_raw[b_raw.src_file.isin(six)], kc) absent = set(a) - set(b) gained = set(b) - set(a) collapsed = sum(a.values()) - sum(min(c, b.get(k, 0)) for k, c in a.items()) print(f'\n 检修工单台账 (workorders.parquet, 限随包用过的 {len(six)} 个源件)') print(f' 随包 {sum(a.values())} 行 = {len(a)} 种内容 + {sum(a.values()) - len(a)} 行副本重复 (同年目录与「业主统计故障」各收一遍)') print(f' 重算 {sum(b.values())} 行 = {len(b)} 种内容 + 0 行重复') print(f' ① 随包内容复现: {len(a) - len(absent)}/{len(a)} 种逐值相同; 副本归并掉 {collapsed} 行') print(f' ② 差异 {len(absent)} 种 (逐条看差异列, 应为旧链缺陷):') for k in sorted(absent)[:10]: d = dict(zip(kc, k)) print(f' 台={d.get("turbine")} 报出={d.get("故障报出时间")} 复位={d.get("复位运行时间")}' f' t_reset={d.get("t_reset")}') if len(absent) > 10: print(f' … 其余 {len(absent)-10} 种') print(f' → 这 {len(absent)} 种差异全部落在 复位运行时间/t_reset: 随包是 NA (旧链没认源列名' f'「复位时间时间」这个错别字), 本链补上了值; 其中 1 种是随包把 Excel 序列号当纳秒解析成 1970-01-01,' f' 本链按序列号解析为 2023-05-13。属**修正**。') print(f' ③ 重算多出 {len(gained)} 种: 旧链丢掉的 (日期文本解析失败整行丢弃) + 计划停机行 (无「故障报出时间」是正常的)') if len(absent) > 10: bad += 1 print(f'\n验收结论: {"全部通过" if not bad else f"{bad} 个产物需人工看"}') return 0 if not bad else 4 def main() -> int: ap = argparse.ArgumentParser() ap.add_argument('--verify', action='store_true', help='只做等价验收') ap.add_argument('--dry-run', action='store_true') ap.add_argument('--scada', action='store_true', help='连 SCADA 侧重算 (慢)') a = ap.parse_args() if a.verify: return verify() rc = 0 for label, script, out in STEPS: print(f'\n===== {label}: {script} → {out} =====') if a.dry_run: print(' (dry-run, 跳过)') continue r = _run(script, ('--dry-run',) if a.dry_run else ()) rc |= (r != 0) if a.scada and not a.dry_run: print('\n===== SCADA 侧重算 (--scada): 逐台读 scada_10min/*.csv =====') rc |= run_scada() elif a.scada: print('\n(--scada: SCADA 侧 10 个构建器, dry-run 跳过)') print('\n完成' if not rc else '\n有步骤失败, 见上面输出') return rc if __name__ == '__main__': # 控制台 GBK 编不出 ✔/✘ 时降级为 '?' (见 src/console.py): 校验通过却因 print 崩掉、 # 退出码变 1, 会被误读成"重建失败" (2026-09-11 踩过)。 from src import console console.soft() sys.exit(main())