guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/service/fleet_views.py

# -*- coding: utf-8 -*-
r"""整场取数视图（P12）：`fleet_view` 及其**传递闭包**内的全部本地函数与模块级全局，机械搬到算法服务。

★来源：`app_backEnd/app_backEnd_guanlan/serve.py`（原样搬运，未改写逻辑）。
搬运范围由 AST 计算（种子 `fleet_view` → 递归展开被调本地函数 → 带上被引用的模块级全局），避免漏项。
"""
from __future__ import annotations

import collections
import datetime as _dt
import json
import pathlib
import re
import threading
import time
from typing import Any

from app_common.app_common_guanlan.api import paths as _P
from src.windscada.config import farm

from src import paths as _P                                              # 路径唯一真源 (与 cwd 无关)
from src.windscada import i18n
from src.windscada import taxonomy
from src.windscada.subsys import temp_nbm, hydraulic, yaw as yawmod, pitch as pitchmod
import hashlib, time
import numpy as np, pandas as pd
import sys, json, pathlib, re, threading, urllib.parse, collections
import threading as _thr

# ── 原 serve.py 的模块级全局（逐字搬） ──
CFG = farm(); ST = pathlib.Path(CFG['store'])

LOCK = threading.Lock()

CFG = farm(); ST = pathlib.Path(CFG['store'])

TS = {'报警': 0, '危险': 0, '良好': 1, '不可判': 2, '优秀': 3}

_CACHE = {}

_HEAVY_LOCK = _thr.Lock()

_STAMP = {'val': None, 'at': 0.0}

_STAMP_TTL = 2.0          # 秒: 指纹有效期 (stat ~150 个文件 ≈ 1~3 ms, 不值得每请求都做)

_WIN_BUSY: dict = {'sysmx': set(), 'curves': set(), 'm9': set()}

_WIN_CACHE: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}

_WIN_ERR: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}

_WIN_LOCK = _thr.Lock()


def _product_files():
    """页面取数依赖的产物文件清单 (顺序稳定: 供指纹; 只列**产物**, 不含 reference/ 随包契约)。

    覆盖 _load() 直读的件 + 它经 taxonomy/temp_nbm/hydraulic/fusion 间接读的件:
      windscada/*.parquet|csv|json  · ontology/*.json  · pitch/*.parquet
      m5_cms_tcm/{handoff_vibration_v2,component_history,baseline_38}.json  · windcms/报告_CMS*.md
    """
    try:
        from src.windcms.config import cms_out as _cms_out     # 与写侧同一解析口 (WINDCMS_OUT)
        _cms_dir = _cms_out()
    except Exception:
        _cms_dir = _P.cms()
    pats = ((ST, ('*.parquet', '*.csv', '*.json')),
            (_P.ont(), ('*.json',)),
            (_P.pitch(), ('*.parquet',)),
            (_P.m5(), ('handoff_vibration_v2.json', 'component_history.json', 'baseline_38.json')),
            (_cms_dir, ('报告_CMS振动状态评估报告_*.md',)))
    files = []
    for d, ps in pats:
        for pat in ps:
            files.extend(sorted(d.glob(pat)))
    return files


def products_stamp(force=False):
    """产物指纹 (sha1 前 16 位)。force=True 时忽略 TTL 立即重算。"""
    now = time.time()
    if not force and _STAMP['val'] is not None and (now - _STAMP['at']) < _STAMP_TTL:
        return _STAMP['val']
    h = hashlib.sha1()
    for p in _product_files():
        try:
            st = p.stat()
            h.update(f'{_P.rel(p)}|{st.st_mtime_ns}|{st.st_size}\n'.encode('utf-8'))
        except OSError:
            h.update(f'{_P.rel(p)}|MISSING\n'.encode('utf-8'))
    _STAMP.update(val=h.hexdigest()[:16], at=now)
    return _STAMP['val']


def _load():
    with LOCK:
        stamp = products_stamp()
        if _CACHE.get('__loaded'):
            if _CACHE.get('__stamp') == stamp:
                return
            # 产物变了 (典型: 刚跑完重算) → 重载。**先建后换**: 下面任何一步抛异常都不会破坏旧缓存,
            # 请求照旧能用旧数 (降级但不空白), 同时日志留痕。
            print(f'[reload] 产物指纹变化 {_CACHE.get("__stamp")} → {stamp}, 重载', flush=True)
        tmp = {}
        try:
            for key, name in (('tm', 'temp_monthly.parquet'), ('al', 'alarms.parquet'),
                              ('lm', 'loss_monthly.parquet'), ('bins', 'powercurve_bins.parquet'),
                              ('pcd', 'powercurve_dev.parquet')):
                f = ST / name
                if not f.exists():
                    raise ProductsMissing(name, f)
                tmp[key] = pd.read_parquet(f)
            tmp['al']['month'] = tmp['al']['t_on'].dt.to_period('M').astype(str)
            tmp['pcd'] = tmp['pcd'].set_index('turbine')
            for key, name in (('duty', 'duty_monthly.parquet'),):
                f = ST / name
                tmp[key] = pd.read_parquet(f) if f.exists() else None
            try:
                tmp['sysmx'] = taxonomy.system_matrix()
                tmp['treg'] = temp_nbm.registry()
                tmp['treg'] = tmp['treg'][0] if isinstance(tmp['treg'], tuple) else tmp['treg']
                tmp['hyd'], _ = hydraulic.registry()
                tmp['hyd'] = tmp['hyd'].set_index('turbine')
            except FileNotFoundError as e:            # 这些派生件同样在产物仓里; 缺了就按"无产物"处理
                raise ProductsMissing(getattr(e, 'filename', '派生产物'), getattr(e, 'filename', ST))
            zp = _P.pitch() / 'pitch_zero_monthly.parquet'
            tmp['zero'] = pd.read_parquet(zp) if zp.exists() else None
            tmp['__stamp'] = stamp
            tmp['__loaded'] = True
            _CACHE.update(tmp)                        # 原子提交: 失败时不留下半截缓存
        except Exception:
            if _CACHE.get('__loaded'):
                print('[reload] 重载失败 → 继续用上一份缓存 (页面不会空白, 但数是旧的; 看上面的异常)', flush=True)
            raise


def _month_end(m):
    """'YYYY-MM' → 该月最后一天 'YYYY-MM-DD'。"""
    import calendar
    y, mo = int(m[:4]), int(m[-2:])
    return f'{m}-{calendar.monthrange(y, mo)[1]:02d}'


def months_of(win):
    """窗 → 覆盖到的月份列表（月度类数据按此过滤）。"""
    _load()
    all_m = sorted(_CACHE['tm'].month.unique())
    if win == '全程': return all_m
    if win == '2025H2': return [m for m in all_m if '2025-07' <= m <= '2025-12']
    if win == '2026H1': return [m for m in all_m if '2026-01' <= m <= '2026-06']
    if win == '2026年': return [m for m in all_m if m >= '2026-01']
    import re as _re
    if _re.fullmatch(r'\d{4}-\d{2}', win):        # 单月窗 (2026-08-27 用户令: 按月份过滤和选择)
        return [m for m in all_m if m == win]
    m2 = _re.fullmatch(r'(\d{4}-\d{2})~(\d{4}-\d{2})', win)
    if m2:                                          # 月区间窗 "2025-04~2025-10"
        return [m for m in all_m if m2.group(1) <= m <= m2.group(2)]
    m3 = _re.fullmatch(r'(\d{4}-\d{2}-\d{2})~(\d{4}-\d{2}-\d{2})', win)
    if m3:                                          # ★自定义起止（含）→ 相交的月 (用户令 2026-09-21)
        a, b = sorted((m3.group(1), m3.group(2)))
        return [m for m in all_m if _month_end(m) >= a and f'{m}-01' <= b]
    n = 1 if win == '近30日' else 3
    return all_m[-n:]


def win_range(win):
    """窗 → (起, 止) 日期串，**含两端**（用户令 2026-09-21：时间窗支持自定义起止日期）。

    预设/逐月/月区间窗一律落到"该时间窗覆盖月的月首/月末"，于是与 `months_of()` 同口径；
    日区间窗原样返回。日粒度件（停机事件/报警/日粒度判据）用它做**含端**过滤。
    """
    import re as _re
    if win:
        if _re.fullmatch(r'\d{4}-\d{2}-\d{2}~\d{4}-\d{2}-\d{2}', win):
            a, b = win.split('~')
            return (a, b) if a <= b else (b, a)
        if _re.fullmatch(r'\d{4}-\d{2}', win):
            return (f'{win}-01', _month_end(win))
        m2 = _re.fullmatch(r'(\d{4}-\d{2})~(\d{4}-\d{2})', win)
        if m2:
            return (f'{m2.group(1)}-01', _month_end(m2.group(2)))
        if win == '2025H2':
            return ('2025-07-01', '2025-12-31')
        if win == '2026H1':
            return ('2026-01-01', '2026-06-30')
    ms = months_of(win)
    if not ms:
        return ('0001-01-01', '9999-12-31')
    return (f'{ms[0]}-01', _month_end(ms[-1]))


def span_of(win):
    """窗 → 计算层用的 (起, 止) 元组（含）；判级/曲线等按窗重算的口子都吃这个。"""
    return win_range(win)


def _mem_mb():
    """本进程可用内存（MB）；拿不到就返回 None（不因为这些诊断信息把重算搞挂）。"""
    try:
        import ctypes

        class _MS(ctypes.Structure):
            _fields_ = [('dwLength', ctypes.c_ulong), ('dwMemoryLoad', ctypes.c_ulong),
                        ('ullTotalPhys', ctypes.c_ulonglong), ('ullAvailPhys', ctypes.c_ulonglong),
                        ('ullTotalPageFile', ctypes.c_ulonglong), ('ullAvailPageFile', ctypes.c_ulonglong),
                        ('ullTotalVirtual', ctypes.c_ulonglong), ('ullAvailVirtual', ctypes.c_ulonglong),
                        ('ullAvailExtendedVirtual', ctypes.c_ulonglong)]
        st = _MS()
        st.dwLength = ctypes.sizeof(_MS)
        ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(st))
        return int(st.ullAvailPhys // (1024 * 1024))
    except Exception:
        return None


def _win_key(win):
    a, b = span_of(win)
    return f'{a}~{b}'


def _win_get(kind, win, fn):
    """→ 值 或 None（None = 正在算/刚起算）。失败**记名**（`_WIN_ERR`）而不是静默当"没数据"。"""
    key = _win_key(win)
    with _WIN_LOCK:
        if key in _WIN_CACHE[kind]:
            return _WIN_CACHE[kind][key]
        if key not in _WIN_BUSY[kind]:
            _WIN_BUSY[kind].add(key)

            def _run():
                try:
                    with _HEAVY_LOCK:                       # 重活串行: 同一时刻只算一份
                        m0 = _mem_mb()
                        print(f'[win] {kind} 按窗重算开算 {key}（可用内存 {m0} MB）', flush=True)
                        v = fn()
                        print(f'[win] {kind} 按窗重算完成 {key}（可用内存 {_mem_mb()} MB）', flush=True)
                except Exception as e:                      # 守护失败必响亮
                    _WIN_ERR[kind][key] = f'{type(e).__name__}: {e}'[:200]
                    print(f'[win] {kind} 按窗重算失败 {key}: {_WIN_ERR[kind][key]}', flush=True)
                    v = None
                with _WIN_LOCK:
                    if v is not None:
                        _WIN_CACHE[kind][key] = v
                    _WIN_BUSY[kind].discard(key)
            _thr.Thread(target=_run, name=f'win-{kind}-{key}', daemon=True).start()
            print(f'[win] {kind} 按窗重算启动 {key}（首次约数十秒，页面会先出 pending）', flush=True)
        return None


def sysmx_of(win):
    """→ (判级矩阵, pending)。`pending=True` 时返回的是"另一口径"的旧矩阵，页面必须如实标注。"""
    from src.windscada import taxonomy
    m = _win_get('sysmx', win, lambda: taxonomy.system_matrix(CFG, span=span_of(win)))
    if m is None:
        return _CACHE['sysmx'], True
    return m, False


def m9_of(win):
    """→ (控制参数一致性, pending)。按窗重算（走窄仓 ≈5 s/窗）；干净窗直接用正式产物。"""
    from src.windscada.perf import control as _cm
    a, b = span_of(win)
    if (a, b) == (_cm.WIN[0], '2025-12-31'):
        return _cm.registry(CFG), False
    v = _win_get('m9', win, lambda: _cm.registry(CFG, span=(a, b)))
    if v is None:
        return _cm.registry(CFG), True          # 未就位: 先给上一口径, 页面按 pending 标注
    return v, False


def fleet_view(win):
    _load()
    ms = months_of(win)
    # ★2026-09-21 用户令「判级也按所选时间窗重算」: 判级矩阵按窗算（后台+缓存，首次 pending）。
    sysmx, sysmx_pending = sysmx_of(win)
    # ① 系统分类问题 (全系统)
    systems, sysdist = {}, {}
    for s in taxonomy.SYSTEMS:
        rows = [dict(t=t, st=sysmx[t][s]['状态'], why=i18n.humanize(sysmx[t][s]['依据']))
                for t in CFG['turbines'] if sysmx[t][s]['状态'] in ('报警', '不可判')]   # 良好不出 (用户令: 只显示有问题的)
        rows.sort(key=lambda r: TS.get(r['st'], 9))
        systems[s] = rows
        # 全场分布 (系统入口卡用: rows 只含问题台, 不能当全场分母 — 会显示"共12台")
        from collections import Counter as _C
        sysdist[s] = dict(_C(sysmx[t][s]['状态'] for t in CFG['turbines']))
        # 良好台清单单列 (系统详情页第三档)
        systems[s + '·良好'] = [dict(t=t, st='良好', why=i18n.humanize(sysmx[t][s]['依据']))
                                for t in CFG['turbines'] if sysmx[t][s]['状态'] == '良好']
    # ①b 关注清单: ≥2 系统报警 (逐台独立台页)
    watch = []
    for t in CFG['turbines']:
        al_sys = [x for x in taxonomy.SYSTEMS if sysmx[t][x]['状态'] == '报警']
        if len(al_sys) >= 2:
            watch.append(dict(t=t, n=len(al_sys), syss=al_sys,
                              # ★不截断: 截到 70 字会把依据切成半句, 而模板反解要完整形态
                              #   ⇒ 英文侧整条落回中文 (2026-09-03 实逮, 总览 7 卡 4 张中文)。
                              #   显示长度归前端 CSS 管。
                              why='；'.join(f"{x}: {i18n.humanize(sysmx[t][x]['依据'])}" for x in al_sys)))
    watch.sort(key=lambda r: -r['n'])
    n_alarm_t = sum(1 for t in CFG['turbines'] if any(sysmx[t][x]['状态'] == '报警' for x in taxonomy.SYSTEMS))
    # ② 故障统计梳理 (真时间窗)
    al = _CACHE['al']; a = al[al.month.isin(ms)]
    top_txt = a.groupby(['code', 'text']).agg(n=('code', 'size'), dur_h=('dur_s', lambda x: x.sum() / 3600)).reset_index()
    pareto_n = top_txt.sort_values('n', ascending=False).head(12)
    pareto_d = top_txt.sort_values('dur_h', ascending=False).head(12)
    monthly = a.groupby('month').size().reindex(ms).fillna(0)
    per_t = a.groupby('turbine').size().sort_values(ascending=False).head(10)
    # 经典故障分析: 条数与时长必须配对看 (2026-08-28)。两个独立排行榜分不出
    # "高频短时(信号抖动或重复触发)" 与 "低频长时(硬故障)" — 处置方向相反, 混在一起会误派工。
    # 第三维=影响台数, 分离"单台刷屏"与"全场批次共性"。
    # 残月识别 (2026-08-28 审核逮): 2026-07 只有 6 天数据(覆盖 19.4%), 却被当整月
    # 并进"2026年"的占比与月度趋势 → 末柱"下降"是假象。占比类分母也因此不完整。
    import calendar as _cal
    _lmc = _CACHE['lm']; _lmc = _lmc[_lmc.month.astype(str).isin(ms)]
    _cov = {}
    for _mm in ms:
        _sub = _lmc[_lmc.month.astype(str) == _mm]
        if not len(_sub):
            _cov[_mm] = 0.0; continue
        _y, _mo = int(_mm[:4]), int(_mm[-2:])
        _cal_h = _cal.monthrange(_y, _mo)[1] * 24
        _cov[_mm] = round(float(_sub['rows_'].sum() / 6 / max(_sub['turbine'].nunique(), 1) / _cal_h), 3)
    _nt = a.groupby(['code', 'text'])['turbine'].nunique().rename('nt')
    _med = a.groupby(['code', 'text'])['dur_s'].median().rename('med_s')
    qd = top_txt.set_index(['code', 'text']).join([_nt, _med]).reset_index()
    qd = qd.sort_values('n', ascending=False).head(30)
    faults = dict(
        pareto_n=[dict(k=f"{r.code} {i18n.alarm_label(r.code, r.text)[:18]}", v=int(r.n)) for _, r in pareto_n.iterrows()],
        pareto_d=[dict(k=f"{r.code} {i18n.alarm_label(r.code, r.text)[:18]}", v=round(float(r.dur_h), 1)) for _, r in pareto_d.iterrows()],
        monthly=dict(months=ms, vals=[int(v) for v in monthly],
                     cov=[_cov.get(x, 0.0) for x in ms],
                     per_day=[round(float(monthly[x]) / max(_cov.get(x, 0.0) * _cal.monthrange(int(x[:4]), int(x[-2:]))[1], 1e-9), 1)
                              if _cov.get(x, 0) > 0.02 else None for x in ms]),
        cov=_cov, cov_min=min(_cov.values()) if _cov else 1.0,
        partial=[x for x in ms if _cov.get(x, 1) < 0.5],
        quad=[dict(code=str(r.code), k=i18n.alarm_label(r.code, r.text)[:20], n=int(r.n),
                   h=round(float(r.dur_h), 1), nt=int(r.nt), med=round(float(r.med_s), 1))
              for _, r in qd.iterrows()],
        n_codes=int(len(top_txt)),
        per_t=[dict(k=k, v=int(v)) for k, v in per_t.items()], total=int(len(a)))
    # ③ SOP 控制策略
    lm = _CACHE['lm']; l = lm[lm.month.astype(str).isin(ms)]
    hrs = l.groupby('state')['rows_'].sum() / 6
    loss = l.groupby('state')['loss'].sum() / 1000
    hrs_tot = max(hrs.sum(), 1)
    stop_h = hrs.get('停机', 0) + hrs.get('停机(调度令)', 0)
    disp_h = hrs.get('停机(调度令)', 0)
    lw_h = hrs.get('低风待机', 0)
    # 口径与 availability.summary 单源一致: 1 − 停机/(总 − 调度令 − 低风待机)
    avail = 1 - (stop_h - disp_h) / max(hrs_tot - disp_h - lw_h, 1)
    pcd = _CACHE['pcd']
    # 能量账闭合 (2026-08-28 经典分析): 四行工况表看不出"电量流向"。
    # 理论可发 = 实际上网 + Σ各态损失; 过物理上限核 (不可超 38台×4.0MW×窗时长)。
    _act = l.groupby('state')['act'].sum() / 1000
    _wf = [dict(k=str(k), loss=round(float(loss.get(k, 0)), 0), act=round(float(_act.get(k, 0)), 0),
                h=round(float(hrs.get(k, 0)), 0)) for k in hrs.index]
    _A = float(_act.sum()); _L = float(loss.sum())
    _cap = len(CFG['turbines']) * 4.0 * float(hrs_tot) / max(len(CFG['turbines']), 1)
    # 逐月能量 (2026-08-28 报表端口): 汇报要环比与趋势, 全时间窗聚合给不出。
    # 残月覆盖率一并带出 — 报表里拿残月和整月比环比会读反 (故障月度图已踩过一次)。
    _em = []
    for _mm in ms:
        _sub = l[l.month.astype(str) == _mm]
        if not len(_sub):
            continue
        _a1 = float(_sub['act'].sum()) / 1000
        _l1 = float(_sub['loss'].sum()) / 1000
        _em.append(dict(m=_mm, act=round(_a1, 0), loss=round(_l1, 0), theo=round(_a1 + _l1, 0),
                        loss_pct=round(_l1 / max(_a1 + _l1, 1) * 100, 1),
                        cov=_cov.get(_mm, 1.0),
                        eflh=round(_a1 / max(len(CFG['turbines']) * 4.0, 1), 0)))
    energy = dict(monthly=_em, act=round(_A, 0), loss=round(_L, 0), theo=round(_A + _L, 0),
                  loss_pct=round(_L / max(_A + _L, 1) * 100, 1),
                  eflh=round(_A / max(len(CFG['turbines']) * 4.0, 1), 0),
                  cap=round(_cap, 0), cap_ok=bool(_A + _L < _cap),
                  items=sorted(_wf, key=lambda r: -r['loss']))
    control = dict(
        energy=energy,
        states=[dict(k=k, h=round(float(v), 0), pct=round(float(v / hrs_tot * 100), 1), loss=round(float(loss.get(k, 0)), 0)) for k, v in hrs.items()],
        avail=round(float(avail * 100), 1),
        curve=dict(sigma=round(float(pcd.dev_w.std() * 100), 2),
                   cands=[dict(t=i, dev=round(float(r.dev_w * 100), 2), 判=r['判别']) for i, r in pcd.iterrows() if r['判别'] != '—'],
                   note='固定判别窗 2025-H2'),
        anchors='限电命令面占比 2025≈0.197 → 2026≈0.769 (专项锚)')
    m8 = None
    try:
        from src.windscada.perf import faults as fmod
        if (ST / 'stop_events.parquet').exists():
            m8 = dict(mtbf=fmod.mtbf_summary(ms), stop_pareto=fmod.stop_pareto(ms), seasonal=fmod.seasonal())
    except Exception as e:
        m8 = dict(err=str(e)[:120])
    m9 = None
    try:
        from src.windscada.perf import control as cmod
        if (ST / 'control_profile.parquet').exists():
            m9, m9_pending = m9_of(win)          # ★按窗重算（用户令 2026-09-21）
    except Exception as e:
        m9 = dict(err=str(e)[:120])
        m9_pending = False
    try:
        from src.windscada.perf import reliability as relmod
        rel = relmod.overview(CFG, span=span_of(win))
    except Exception as e:
        rel = dict(err=str(e)[:120])
    try:
        from src.windscada.subsys import fusion as fusmod
        ftab, fmeta = fusmod.fusion_table(CFG, all_turbines=True)
        mrows, mheads = fusmod.matrix(CFG, ms)   # 整体时间窗: 矩阵现窗/记忆轨迹随全局窗选择器
        cap = fusmod.capability()
        ev, ev_bounds = fusmod.drivetrain_events(ms, CFG)
        lvcnt = {}
        for r in mrows:
            lv = r['振动']['level']; lvcnt[lv] = lvcnt.get(lv, 0) + 1
        fkpi = dict(定论=lvcnt.get('bad', 0), 预警=lvcnt.get('warn', 0), 候选观察=lvcnt.get('note', 0),
                    销案正常=lvcnt.get('ok', 0), 未列=lvcnt.get('unlisted', 0),
                    收录台=len(CFG['turbines']) - lvcnt.get('unlisted', 0), 全场=len(CFG['turbines']),
                    机制链=sum(1 for r in mrows if '机制链' in r['润滑'].get('note', '')),
                    系数=dict(wear=cap['classes'].get('wear_progressive', {}).get('coef'),
                            thermal=cap['classes'].get('thermal_acute', {}).get('coef'),
                            unknown=cap['classes'].get('unknown', {}).get('coef')))
        # 矩阵补 报警/工单 两列 (2026-08-28 用户令) — 报警=时间窗内真窗计数+族分布; 工单=历史台账(窗见列头)
        # 报警/工单两列只统计传动链三部件 (2026-08-28 用户令: 只说发电机/齿轮箱/主轴) —
        # 本矩阵每行是"该台传动链状态", 全场报警计数会把变桨/偏航/主控的量混进来 (316 条里绝大多数与传动链无关)
        _DT_PAT = '齿轮|齿箱|润滑油|油冷|滤芯|发电机|定子|绕组|滑环|主轴承|主轴|轴承'
        _a = _CACHE['al']; _aw = _a[_a.month.isin(ms)]
        _aw = _aw[_aw.text.str.contains(_DT_PAT, na=False)]
        _acnt = _aw.groupby('turbine').size()
        _arank = _acnt.rank(ascending=False, method='min')
        _FAM = [('齿轮箱', '齿轮|齿箱|润滑油|油冷|滤芯'), ('发电机', '发电机|定子|绕组|滑环'),
                ('主轴承', '主轴承|主轴')]
        try:
            from src.windscada.subsys import workorder as _wo
            _wod = _wo.load(CFG)
        except Exception:
            _wod = None
        for _r in mrows:
            _t = _r['turbine']; _g = _aw[_aw.turbine == _t]
            _fam, _taken = [], set()
            _CODEFAM = {'3225': '变桨叶片'}       # 文本为空的码走码号归族 (3225=变桨液压, 现场西门子资料对译表)
            for _nm, _pat in _FAM:
                _byc = _g.code.astype(str).map(_CODEFAM) == _nm
                _sub = _g[~_g.index.isin(_taken) & (_g.text.str.contains(_pat, na=False) | _byc)]
                _taken |= set(_sub.index)
                if len(_sub): _fam.append(dict(k=_nm, n=int(len(_sub))))
            _oth = len(_g) - sum(f['n'] for f in _fam)
            if _oth > 0: _fam.append(dict(k='其他', n=int(_oth)))
            # 刷屏闸: 单码占比过半 → 条数不代表"问题多"而是一个码在抖 (08# 4609/4918=93.7% 码3225 集中2026-01)
            _burst = None
            if len(_g):
                _bc = _g.groupby('code').size().sort_values(ascending=False)
                _share = float(_bc.iloc[0]) / len(_g)
                if _share >= 0.5:
                    _sub2 = _g[_g.code == _bc.index[0]]
                    _burst = dict(code=str(_bc.index[0]), share=round(_share, 3), n=int(_bc.iloc[0]),
                                  months=sorted({str(m)[:7] for m in _sub2.t_on.dt.to_period('M').astype(str)}),
                                  med_s=float(_sub2.dur_s.median()) if 'dur_s' in _sub2 else None)
            # 明细 (用户令"浮窗说明之前发生问题"): top 码 + 首末时间
            _top = []
            if len(_g):
                for (_c2, _tx), _sub3 in _g.groupby(['code', 'text']):
                    _top.append(dict(code=str(_c2), text=str(_tx)[:34], n=int(len(_sub3)),
                                     first=str(_sub3.t_on.min())[:10], last=str(_sub3.t_on.max())[:10]))
                _top.sort(key=lambda x: -x['n'])
            _r['报警'] = dict(n=int(_acnt.get(_t, 0)), rank=int(_arank.get(_t, 0)) if _t in _arank else None,
                             tot=len(CFG['turbines']), fam=_fam, burst=_burst, items=_top[:6])
            if _wod is not None and len(_wod):
                _w = _wod[_wod.turbine == _t]
                _wtxt = (_w.get('故障名称', '').astype(str) + ' ' + _w.get('故障位置二级', '').astype(str)
                         + ' ' + _w.get('维修对象', '').astype(str) + ' ' + _w.get('元器件名称', '').astype(str))
                _w = _w[_wtxt.str.contains(_DT_PAT, na=False)]
                _acts = _w['维修动作'].replace('', pd.NA).dropna().value_counts() if len(_w) else None
                _wi = []
                for _, _wr in _w.sort_values('t_report', ascending=False).head(6).iterrows():
                    _wi.append(dict(date=str(_wr.get('t_report'))[:10], name=str(_wr.get('故障名称') or '')[:30],
                                    act=str(_wr.get('维修动作') or ''), part=str(_wr.get('维修对象') or _wr.get('元器件名称') or '')[:14]))
                _r['工单'] = dict(n=int(len(_w)),
                                  acts=[dict(k=str(k), n=int(v)) for k, v in (_acts.head(3).items() if _acts is not None else [])],
                                  items=_wi,
                                  last=(lambda v: str(v.max())[:10] if len(v) else None)(_w.t_report.dropna()[_w.t_report.dropna().dt.year > 2000]) if len(_w) else None)
            else:
                _r['工单'] = dict(n=0, acts=[], last=None)
        _wyr = ''
        if _wod is not None and len(_wod):
            _tv = _wod.t_report.dropna()
            _tv = _tv[_tv.dt.year > 2000]        # 空日期落 1970 epoch, 剔除后再报窗 (否则列头写"1970~")
            if len(_tv): _wyr = f"{_tv.min():%Y}~{_tv.max():%Y}"
        mheads = dict(mheads, 报警=f"传动链三部件 · {ms[0]}~{ms[-1]}" if ms else '传动链三部件',
                      工单=f"传动链三部件 · 台账 {_wyr}" if _wyr else '台账不可用')
        # 决策链进度盘 (2026-08-28 用户问"这个到底能怎么用"): 原来只有 29# 一台的静态六格,
        # 是展示牌不是工作面 — 打开它做不了任何决定。改为对全部需跟踪台报"走到第几步、卡在哪、下一步做什么"。
        # 判定规则单源在此, 前端只渲染不判断。
        _CLOOP = {}
        try:
            _ch = json.loads((ST.parent / 'm5_cms_tcm' / 'component_history.json').read_text(encoding='utf-8'))
            for _k, _v in _ch.get('summary', {}).items():
                if _k.startswith('★换件闭环') and isinstance(_v, list):
                    for _r in _v:
                        _CLOOP[_r['turbine']] = _r
                if _k.startswith('★当前在升') and isinstance(_v, list):
                    for _r in _v:
                        _CLOOP.setdefault(_r['turbine'], {}).update(rising=_r)
        except Exception:
            pass
        _HIT = {'bad', 'warn', 'note'}
        # ★一台可能有多行 (29#/17# 各 2 行: 齿轮箱 + 主轴承)。用 {turbine: row} 直接建字典会被后一行覆盖,
        # 实测 29# 因此取到"齿轮箱二级行星内齿圈·监视"那行, 而它的决策链讲的是"主轴承+集中润滑泵·定论"
        # ⇒ 浮窗里部件、温度通道、油样全是另一个部件的 (memory: 画图须确认各维度来自同一个体)。
        _SEV = {'bad': 0, 'warn': 1, 'note': 2, 'stale': 3, 'ok': 4, 'unlisted': 5}
        _tbi = {}
        for _row in ftab.to_dict('records'):
            _k = _row['turbine']
            if _k not in _tbi or _SEV.get(_row.get('级'), 9) < _SEV.get(_tbi[_k].get('级'), 9):
                _tbi[_k] = _row
        board = []
        for _r in mrows:
            t = _r['turbine']
            srcs = [k for k in ('振动', '温度', '润滑', '油液') if (_r.get(k) or {}).get('level') in _HIT]
            lvl = (_r.get('振动') or {}).get('level')
            tb = _tbi.get(t, {})
            mech = ('机制链' in ((_r.get('润滑') or {}).get('note') or '')
                    or '机制链' in (tb.get('振动结论') or ''))
            n_wo = (_r.get('工单') or {}).get('n') or 0
            cl = _CLOOP.get(t) or {}
            closed = bool(cl.get('verdict') == '恢复')
            rising = cl.get('rising')
            if not (lvl in _HIT or (srcs and lvl != 'ok')):
                continue
            # 六步: done=已达 / open=未达 / unknown=数据不足不可判
            steps = [
                dict(k='证据', st='done' if srcs else 'open', v=f"{len(srcs)} 源", d='/'.join(srcs)),
                dict(k='机制', st='done' if mech else 'open',
                     v='已定性' if mech else '未定性', d='根因定到部件与失效模式'),
                dict(k='判级', st='done' if lvl in _HIT else 'open',
                     v=tb.get('证据状态') or lvl or '—', d=f"设备状态 {tb.get('设备状态') or '—'}"),
                dict(k='排期', st='done' if t == 'WTG29' else 'open',
                     v='情景A 2026-09' if t == 'WTG29' else '未排',
                     d='进检修排程沙盘 (风险×损失双轴)'),
                # ★这一列一律"不可判", 不是偷懒: 工单台账止 2024-11 而判级时间窗在 2026,
                # 台账里的历史工单不可能是针对本次问题的 ⇒ 用 n>0 判"已派工"会造出假的完成态。
                dict(k='动作', st='unknown',
                     v=(f"台账 {n_wo} 单" if n_wo else '台账无'),
                     d='工单台账 2020~2024-11, 25-26 现场未提供 — 历史单不对应本次问题, 须向现场核实'),
                # ★历史闭环不占本格: 05#/20# 2024~2025 换过齿轮箱且已验证恢复, 但本轮又有新的候选级证据,
                # 把历史闭环填进第六格会出现"验收已达而机制未定"的自相矛盾行。历史闭环走 hist_loop 单独标记。
                dict(k='验收', st='open', v='未验',
                     d='本轮动作执行后复测同一判据是否回落' + (
                        f" (该台 {cl.get('replace_date')} 有过一次闭环: {cl.get('component')} 降至 {cl.get('ratio')}×)" if closed else '')),
            ]
            _first = next((x for x in steps if x['st'] != 'done'), None)
            stuck = _first['k'] if _first else None
            stuck_kind = _first['st'] if _first else None
            NEXT = {'证据': '补测: 四源全为常态, 先确认监测面有效',
                    '机制': '定性: 现场/取样把根因定到部件与失效模式 — 机制不清则排期与动作都无依据',
                    '判级': '送审: 按 RV-1 触发相应审级',
                    '排期': '排程: 进沙盘比情景, 出建议窗口',
                    '动作': '核实: 向现场调取 2025~2026 工单 — 台账止于 2024-11, 系统无法判定是否已派工',
                    '验收': '复测: 动作已执行, 复测同一判据是否回落'}
            board.append(dict(t=t, lvl=lvl, ostate=tb.get('设备状态'), estate=tb.get('证据状态'),
                              part=tb.get('部件'), steps=steps, stuck=stuck, stuck_kind=stuck_kind,
                              next=NEXT.get(stuck, ''), done=sum(1 for x in steps if x['st'] == 'done'),
                              rising=rising, hist_loop=(cl if closed else None), srcs=srcs))
        _ORD = {'bad': 0, 'warn': 1, 'note': 2}
        board.sort(key=lambda r: (_ORD.get(r['lvl'], 3), -r['done']))
        _cl_rows = [v for v in _CLOOP.values() if v.get('verdict') == '恢复']
        fus = dict(链盘=dict(rows=board, closed=_cl_rows,
                            stuck=dict(collections.Counter(r['stuck'] for r in board if r['stuck'])),
                            wo_bound='工单台账 2020~2024-11 (现场未提供 25-26)'),
                   表=ftab.to_dict('records'),
                   矩阵=mrows, 列窗=mheads, kpi=fkpi,
                   事件=ev, 事件边界=ev_bounds,
                   能力=dict(classes=cap['classes'], blind=[str(b) for b in cap['blind']], loop=cap.get('loop', {})),
                   窗=dict(振动=fmeta.get('振动窗', '—'), SCADA=fmeta.get('SCADA窗', ''), 油样=fmeta.get('油样窗', '')),
                   色标=fmeta.get('色标', {}), 数据时点=fmeta.get('数据时点', ''), handoff日期=fmeta.get('handoff日期', ''),
                   证据窗末=fmeta.get('证据窗末', ''),
                   纪律=str(fmeta.get('纪律', '')), 盲区=[str(b) for b in fmeta.get('盲区', [])],
                   open_items=[str(o) for o in fmeta.get('open_items', [])], gap=fmeta.get('gap'))
    except Exception as e:
        fus = dict(err=str(e)[:160])
    kpi = dict(报警台=n_alarm_t, 关注台=len(watch), 全场=len(CFG['turbines']))
    return dict(win=win, months=ms, all_months=sorted(_CACHE['tm'].month.unique()),
                systems=systems, sysdist=sysdist, watch=watch, kpi=kpi, rel=rel, fus=fus, faults=faults, control=control, m8=m8, m9=m9,
                # ★2026-09-21: 判级轴现在**按所选窗重算**；首次是后台算，这一份仍是旧口径 ⇒ 如实标 pending
                sysmx_pending=bool(sysmx_pending),
                m9_pending=bool(m9_pending),
                win_pending=bool(sysmx_pending or m9_pending),
                sysmx_span=list(span_of(win)),
                note=('判级轴(变桨/偏航/蓄能/温度)按所选时间窗重算；曲线按所选时间窗重算；'
                      '故障统计/五态/温度月轨迹/停机台账=所选窗真窗'
                      + ('（⚠ 判级矩阵正在按所选时间窗重算，下面系统卡暂为上一份口径，稍后自动刷新）' if sysmx_pending else '')))