guanlan-src/app_ETL/app_ETL_guanlan/builders/windscada_monthly_build.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""月度派生件构建器 —— 补上 v0.2.0 里"有消费者、却没生成端"的那批产物 (2026-09-12)。

## 为什么有这个脚本

从零重算 (`rebuild_from_raw.py`) 只覆盖 16 件; 随包里另有 15 件 parquet **没有任何生成端**
(全库搜遍: 只有读取方、0 处写入方), 于是工作台主数据 `/api/fleet` 一缺 `temp_monthly.parquet`
就整个回 `err=no_products` —— 页面看着像"没数据", 其实是"缺产物"。

本脚本把这批件**从 data/raw 重新算出来**, 并且**逐值对齐随包件**: 随包件在这里当"标准答案"
(存在 `_products_off/.../windscada/`), `--verify` 会报每件的命中率; 规则是从"标准答案"反推并
用全量比对确认的, 不是猜的。

## 已确认的口径 (2026-09-12, 用随包件反推 + 逐值验证)

    temp_monthly          每台 × 每月 × 每个温度通道, 在 **发电态功率门 p ≥ 500kW** 行上的**中位数**
                          形状 38×19×27 = 19494 行; 实测 WTG01 的 513 格 **逐值 100% 相同**
                          (对照: 全行中位数只有 5.8% 命中 —— 因为停机时绕组是冷的, 会把中位拉低)
                          口径出处: `src/windscada/subsys/temp_nbm.py` 的"发电态×功率档(500kW)", PBINS 步长 500

## 用法

    python scripts/windscada_monthly_build.py --verify        # 只比对(不写盘), 报每件命中率
    python scripts/windscada_monthly_build.py                 # 算并写盘
    python scripts/windscada_monthly_build.py --turbines WTG01,WTG02   # 只算几台(调试用)
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[3]

import argparse
import pathlib
import sys

ROOT = _install_root(__file__)      # 模块化后按标记找安装根（原 parents[1] 已不成立）
sys.path.insert(0, str(ROOT))

import pandas as pd                                          # noqa: E402

from src import paths as P                                   # noqa: E402
from src.windscada.config import farm                        # noqa: E402
from src.windscada.data import load_10min, contracted_cols   # noqa: E402

POWER_GATE = 500.0        # 发电态功率门 (kW); 出处见 temp_nbm 的"发电态×功率档(500kW)"
GROUPS = ['A.功率', 'B.温度NBM']


def baseline_dir():
    """"标准答案"目录: 随包件在哪 (整目录视角; 单件比对用 baseline_for)。"""
    inplace = P.store() / '_pre_rebuild_20260911'
    if inplace.is_dir():
        return inplace
    for p in sorted((ROOT / '_products_off').rglob('windscada')):
        if p.is_dir():
            return p
    return None


def baseline_for(name: str):
    """这一件产物的"标准答案"在哪。

    注意两个暂存处**内容不同**, 必须逐件找:
      · `outputs/<场>/windscada/_pre_rebuild_20260911/` = **台账验收基线**(3 件: alarms/workorders/oil_samples);
      · `_products_off*/**/windscada/`                   = 随包整份产物(31 件 parquet, 含月度派生件)。
    原先只认第一个目录, 于是 temp_monthly 会被报成"没有随包基线可比", 而结论却打印"全部逐值一致"
    —— 2026-09-12 实测逮到(假结论比没结论更糟)。
    """
    cands = []
    bd = baseline_dir()
    if bd:
        cands.append(bd / name)
    for p in sorted((ROOT).glob('_products_off*/**/windscada')):
        cands.append(p / name)
    for c in cands:
        if c.is_file():
            return c
    return None


BASELINE = baseline_dir()


def temp_channels(cfg) -> list:
    """契约里 B.温度NBM 组的活通道 (以 _mean 结尾) —— temp_monthly/watch_channels 的列集。"""
    return [c for c in contracted_cols(cfg, groups=['B.温度NBM']) if c.endswith('_mean')]


def month_of(s: pd.Series) -> pd.Series:
    return s.dt.to_period('M').astype(str)


def build_temp_monthly(cfg, turbines=None, progress=True):
    """→ (长表 DataFrame[turbine, channel, month, med], 明细 dict)

    规则: 每台 × 每月 × 每个温度通道, 在 p ≥ 500kW 行上的中位数。

    ★**不做"样本太少就不出月"的门** (用户令 2026-09-20 裁决: 保持现状、如实反映源件):
      实测 2026-09 只有 1 行(补充件 `WTG01-B2.csv` 的最后一个时间戳 `2026-09-01 00:00`),
      该月照样出一行 —— 页面时间窗按月过滤时它就"一格数据"。理由是**产物=当前 data/raw 的函数**:
      源件里有这个时间戳, 就该看得见; 加"最少样本"门会让"数据到没到"这件事变成猜的。
      要少这一格, 现场把那个时间戳裁掉或让现场别导出月初那一行即可, 不在本器里静默过滤。
    """
    chans = temp_channels(cfg)
    rows, info = [], {}
    for t in (turbines or cfg['turbines']):
        d = load_10min(t, cfg, groups=GROUPS)
        d['month'] = month_of(d['ts'])
        n_all = len(d)
        d = d[d['grd_wtc_ActPower_mean'] >= POWER_GATE]
        if not len(d):
            info[t] = dict(rows_all=n_all, rows_gate=0, months=0)
            continue
        med = d.groupby('month')[chans].median()
        long = med.stack().rename('med').reset_index()
        long.columns = ['month', 'channel', 'med']
        long.insert(0, 'turbine', t)
        rows.append(long)
        info[t] = dict(rows_all=n_all, rows_gate=len(d), months=len(med))
        if progress:
            print(f'    {t}: 全 {n_all} 行 → 发电态 {len(d)} 行, {len(med)} 个月', flush=True)
    return (pd.concat(rows, ignore_index=True) if rows else pd.DataFrame(columns=['turbine', 'channel', 'month', 'med'])), info


PRODUCTS = {
    'temp_monthly.parquet': build_temp_monthly,
}


def verify(cfg) -> int:
    """逐件与随包件比对: 行数、键集合、逐值相等比例。

    三种结果分开报, 不许混:
      ✅ 逐值一致  ❌ 有差异(规则需修正)  ?? 找不到标准答案(**不能**当成一致)
    """
    ok_all, compared, no_base = True, 0, []
    for name in PRODUCTS:
        mine_p = P.store() / name
        if not mine_p.exists():
            print(f'  [ -- ] {name:34s} 本仓还没算 (跳过)')
            continue
        base_p = baseline_for(name)
        if base_p is None:
            no_base.append(name)
            print(f'  [ ?? ] {name:34s} 找不到随包件可对比 (台账基线只有 3 件; 随包整份产物在 _products_off*/ 下)')
            continue
        compared += 1
        a, b = pd.read_parquet(base_p), pd.read_parquet(mine_p)
        keys = [c for c in a.columns if c != 'med']
        m = a.merge(b, on=keys, how='outer', suffixes=('_base', '_mine'), indicator=True)
        val = [c for c in a.columns if c not in keys][0]
        both = m[m._merge == 'both']
        n = len(both)
        eq = int((both[f'{val}_base'].sub(both[f'{val}_mine']).abs() < 1e-9).sum())
        only_b = int((m._merge == 'left_only').sum())
        only_m = int((m._merge == 'right_only').sum())
        good = (n > 0 and eq == n and only_b == 0 and only_m == 0)
        ok_all &= good
        mark = '✅' if good else '❌'
        print(f'  [{mark}] {name:34s} 基线 {len(a)} 行 / 本仓 {len(b)} 行; 逐值相等 {eq}/{n};'
              f' 仅基线 {only_b}; 仅本仓 {only_m}   [标准答案: {P.rel(base_p)}]')
        if not good and n:
            d = both.assign(_d=(both[f'{val}_base'] - both[f'{val}_mine']).abs())
            print('       差异最大的 3 格:')
            print(d.nlargest(3, '_d')[keys + [f'{val}_base', f'{val}_mine']].to_string(index=False))
    if not compared:
        print(f'\n结论: **无法比对**(没有一件找到随包件) —— 这不是"通过"。缺标准答案: {no_base}')
        return 5
    if no_base:
        print(f'\n注意: 另有 {len(no_base)} 件找不到标准答案, 未比对: {no_base}')
    print('\n结论:', '已比对的部分全部逐值一致' if ok_all else '有差异 —— 规则还需修正, 别急着写盘')
    return 0 if ok_all else 4


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--verify', action='store_true', help='只与随包基线比对, 不写盘')
    ap.add_argument('--turbines', default=None, help='只算这几台 (逗号分隔)')
    a = ap.parse_args()
    cfg = farm()
    ts = a.turbines.split(',') if a.turbines else None
    print(f'场站 {cfg["name"] if "name" in cfg else cfg["raw_station"]} · 温度通道 {len(temp_channels(cfg))} 个 · 发电态功率门 {POWER_GATE:.0f} kW\n')
    if not a.verify:
        for name, fn in PRODUCTS.items():
            print(f'== 算 {name}')
            df, info = fn(cfg, turbines=ts)
            out = P.store() / name
            out.parent.mkdir(parents=True, exist_ok=True)
            df.to_parquet(out, index=False)
            print(f'   → {P.rel(out)}  {df.shape[0]} 行 × {df.shape[1]} 列\n')
    print('== 与随包基线逐值比对 ==')
    return verify(cfg)


if __name__ == '__main__':
    # 控制台可能是 GBK(中文 Windows 936): print 里的 ✅ ❌ 编不出来会抛 UnicodeEncodeError,
    # 明明算完了却以退出码 1 结束。降级为 '?' 而不是崩(同类坑见 src/console.py)。
    import sys as _sys
    for _s in (_sys.stdout, _sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    sys.exit(main())