guanlan-src/scripts/tcm_mask_thresholds_build.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""TCM 掩码阈值件（`tcm_compatible_replay/model/mask_thresholds.tsv`）的**观澜自算**生成端（源代码化清单第 ③ 件）。

## 消费者与格式（不能改的契约）

`src/windcms/data.py::load_masks(cfg)` 读 `cfg['mask_thresholds']` 这个 TSV，列固定：

    state · measurement_key · condition_key · mask_time · value_median · yellow_scalar · red_scalar
    （只收 state=='Active'，按 (measurement_key, condition_key) 取 mask_time 最新一条）

`measurement_key` 由 `data.mask_key(farm_prefix, turbine, sensor, meas)` 拼成
（`<CMS SiteName>/<台>/<测点>/<测量>`, 与 `src/windcms/tcm.py::status_table(..., 'CGN Rudong')` 同前缀）。
消费它的页面：逐台页的阈值线、`status_table` 的 mask 状态列。

## 口径（★必读：这是"观澜自算"，不是厂商阈值）

现场导出件里**没有**掩码定义 —— 实测 25,679 个 `*_decode.json` 里只有 `NumberOfMasks: 175`，
掩码本体（黄/红标量）在 CMS 库内、未随导出 ⇒ **厂商阈值无法从给定原始件复原**。
用户令「所有的计算均要形成观澜的源代码」⇒ 本器按**观澜自己的自适应基线**给阈值：

    value_median / yellow / red = src/sop/discriminators.py::adaptive_baseline_threshold(...)
    （per (台, 测点, 测量, 功率工况)：窗内中位 + k×MAD，k_yellow=3 / k_red=6，基线先净化；样本门 min_n=30）

来历写进同目录 `mask_thresholds.说明.json`（TSV 不能带注释行，否则 `read_csv(sep='\t')` 会被带偏），
页面据此把线标成"观澜自算阈值"而不是"厂商黄/红线"（见 src/windcms/workbench.py 的措辞分支）。

用法: python scripts/tcm_mask_thresholds_build.py [--status] [--window w0316]
"""
from __future__ import annotations

import argparse
import json
import pathlib
import sys
import time

import pandas as pd

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / 'scripts'))
from src import paths as P                                              # noqa: E402
from src.sop.discriminators import adaptive_baseline_threshold as _ABT   # noqa: E402  ← 单一实现

SITE_PREFIX = 'CGN Rudong'      # 与 tcm.status_table(s, masks, 'CGN Rudong') 同一前缀
SETUP = 'mask_thresholds.说明.json'


def _target(farm: str | None = None) -> pathlib.Path:
    return P.tcm_replay(farm) / 'model' / 'mask_thresholds.tsv'


def status(farm: str | None = None) -> int:
    p = _target(farm)
    note = p.parent / SETUP
    print(json.dumps(dict(path=P.rel(p), exists=p.is_file(),
                          rows=(sum(1 for _ in p.open(encoding='utf-8')) - 1) if p.is_file() else 0,
                          caliber=('观澜自算(k×MAD 自适应基线)' if note.is_file() else
                                   ('厂商阈值(随包件)' if p.is_file() else None)),
                          note='不在位 ⇒ 逐台页不画阈值线(文案"厂商未设阈值"); 跑本器可由窗索引自算'),
                     ensure_ascii=False))
    return 0


def build(farm: str | None = None, window: str | None = None) -> int:
    m5 = P.m5(farm)
    wins = [window] if window else [d.name for d in sorted((m5 / 'windows').glob('w[0-9][0-9][0-9][0-9]'))
                                    if (d / 'index.parquet').is_file()]
    if not wins:
        print('[X] 没有可用的窗索引（m5_cms_tcm/windows/*/index.parquet）—— 先跑振动摄入')
        return 2
    cols = ['turbine', 'sensor_name', 'meas_name', 'condition_key', 'scalar_value', 'ds_size', 'trigger_time']
    rows, n_skip = [], 0
    for w in wins:
        ix = m5 / 'windows' / w / 'index.parquet'
        d = pd.read_parquet(ix, columns=[c for c in cols if c in pd.read_parquet(ix).columns])
        if 'ds_size' in d.columns:                       # 只收标量(ds_size==1); 谱记录没有标量
            d = d[pd.to_numeric(d['ds_size'], errors='coerce') == 1]
        d = d[pd.to_numeric(d['scalar_value'], errors='coerce').notna()]
        for (t, sen, meas, ck), g in d.groupby(['turbine', 'sensor_name', 'meas_name', 'condition_key']):
            v = pd.to_numeric(g['scalar_value'], errors='coerce').dropna()
            if len(v) < 30:                              # 样本门: 与 discriminators 的 min_n 一致, 不足则不给阈值
                n_skip += 1
                continue
            r = _ABT(list(v))
            med = r.get('baseline_med')
            yel, red = r.get('yellow'), r.get('red')
            if med is None or yel is None or red is None:
                n_skip += 1
                continue
            rows.append(dict(state='Active',
                             measurement_key=f'{SITE_PREFIX}/{t}/{sen}/{meas}',
                             condition_key=str(ck), mask_time=time.strftime('%Y-%m-%d'),
                             value_median=round(float(med), 6), yellow_scalar=round(float(yel), 6),
                             red_scalar=round(float(red), 6)))
    if not rows:
        print(f'[X] 一条阈值也没算出来（可行组 {n_skip} 组都卡在样本门 n≥30）—— 不写空文件')
        return 5
    out = _target(farm)
    out.parent.mkdir(parents=True, exist_ok=True)
    pd.DataFrame(rows).to_csv(out, sep='\t', index=False, encoding='utf-8')
    note = dict(
        schema='guanlan-mask-thresholds/v1', built=time.strftime('%Y-%m-%d %H:%M:%S'),
        generator='scripts/tcm_mask_thresholds_build.py',
        caliber='观澜自算（不是厂商阈值）',
        why='现场导出件里没有掩码定义（25,679 个 *_decode.json 里只有 NumberOfMasks: 175, 掩码本体在 CMS 库内未导出）'
            '⇒ 厂商黄/红线无法从原始件复原; 本件按观澜自适应基线出阈值。',
        formula='src/sop/discriminators.py::adaptive_baseline_threshold（per (台,测点,测量,工况) 窗内中位 + '
                'k×MAD, k_yellow=3 / k_red=6, 基线先净化, 样本门 n≥30）',
        rows=len(rows), skipped_groups=n_skip, site_prefix=SITE_PREFIX,
        window=','.join(wins),
        usage='src/windcms/data.py::load_masks 读本 TSV; 页面按"观澜自算阈值"标注（见 workbench.py）')
    (out.parent / SETUP).write_text(json.dumps(note, ensure_ascii=False, indent=1), encoding='utf-8')
    print(f'已写 {P.rel(out)}: {len(rows)} 条阈值（组样本门不足跳过 {n_skip} 组）· 口径见 {SETUP}')
    try:
        from src import derived_manifest as DM
        DM.record(P.out_root(farm), {
            f'tcm_compatible_replay/model/mask_thresholds.tsv':
                'scripts/tcm_mask_thresholds_build.py (观澜自算: 中位 + 3/6×MAD 自适应基线)',
            f'tcm_compatible_replay/model/{SETUP}':
                'scripts/tcm_mask_thresholds_build.py (口径说明: 非厂商阈值)'}, by='tcm_mask_thresholds_build')
        print('   已自登记 → _derived_manifest.json')
    except Exception as e:
        print(f'   [i] 自登记跳过: {type(e).__name__}: {e}')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description='TCM 掩码阈值件（观澜自算: 自适应基线）')
    ap.add_argument('--farm', default=None)
    ap.add_argument('--window', default=None)
    ap.add_argument('--status', action='store_true')
    a = ap.parse_args()
    return status(a.farm) if a.status else build(a.farm, a.window)


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try:
            _s.reconfigure(errors='replace')
        except Exception:
            pass
    sys.exit(main())