guanlan-src/app_ontology/app_ontology_guanlan/sop/vib_confidence_wire.py

# -*- coding: utf-8 -*-
r"""★① 置信度**接线**（用户令 2026-10-06）：把 `vib_confidence` 的五轴评分接到现有产物上。

设计要点（刻意保守，防"凑分"）：
- **只读现有产物**：融合表行（`fusion_table`）+ 振动 handoff（`per_turbine` 列表）。**不新增数据源**。
- **缺数的轴显式标 `[INS]`** 并计 0 分，同时在 `notes` 里写明缺什么 —— 不猜、不补、不做默认满分。
- **只加不改**：产出 `置信度` 字段；**不参与**任何判级/等级/结论的生成（接线不影响现有裁决）。
- 五轴取数（能取到什么就取什么）：
  · A 内部证据强度：handoff 里若有逐台"闸/门"通过数则用之；否则 `[INS]`。
  · B 独立源共时确认：融合表行的 `温度判`（非 —）与 `油新鲜=True` 计为**同窗**独立源；两者都无 ⇒ `[INS]`。
  · C 实物锚：handoff 行里若有 `tier/anchor/实物` 字段则映射；否则 `[INS]`。
  · D 样本充分性：融合表 `窗区间` 的窗数 + handoff 行里的记录数（若有）；无窗数 ⇒ 按 1 窗计并标注。
  · E 反证搜索：**当前无产物承载** ⇒ 一律 `[INS]`（这正是"没做过反证搜索 = 0 分"的口径）。
"""
from __future__ import annotations

import re

from . import vib_confidence as VC

WINDOW_CLAUSE_DEN = lambda n: max(4, -(-2 * int(n) // 3)) if n and n > 0 else 4   # ceil(2N/3) 起步 4


def _pt_map(handoff: dict) -> dict:
    """handoff['per_turbine'] 是列表 ⇒ 归一成 {turbine: row}（容错）。"""
    out = {}
    pt = (handoff or {}).get('per_turbine') or []
    if isinstance(pt, dict):
        return dict(pt)
    for row in pt:
        if not isinstance(row, dict):
            continue
        tid = row.get('turbine') or row.get('id') or row.get('台') or row.get('WTG')
        if tid:
            out[str(tid)] = row
    return out


def _norm_tid(x) -> str:
    s = str(x or '').strip()
    m = re.search(r'(\d+)', s)
    if not m:
        return s
    return 'WTG%02d' % int(m.group(1))


def build_axes(frow: dict, prow: dict | None) -> dict:
    """按可用数据装配五轴；缺数轴标 [INS] 并计 0。返回 {axes, ins, notes}。"""
    prow = prow or {}
    notes: list[str] = []
    ins: list[str] = []

    # ---- A 内部证据强度：找 handoff 里的"闸"过数 ----
    a = None
    for k in ('gates_passed', '闸过', '闸', 'passed_gates', 'n_gates_passed'):
        if k in prow and isinstance(prow[k], (int, float)):
            a = float(prow[k])
            break
    if a is None:
        ins.append('A:handoff 无逐台闸过数')
        A = 0.0
    else:
        A = VC.score_internal(int(a), n_gates=int(prow.get('n_gates') or 5),
                              anchor_ok=bool(prow.get('anchor_ok') or False),
                              persist_windows=int(prow.get('persist_windows') or 0),
                              n_windows=int(prow.get('n_windows') or 1))

    # ---- B 独立源共时确认（同窗）----
    srcs = set()
    tjudge = str(frow.get('温度判') or '').strip()
    if tjudge and tjudge not in ('—', '-', ''):
        srcs.add('temperature')
    if str(frow.get('油新鲜')).lower() == 'true':
        srcs.add('oil')
    B = VC.score_independent(srcs) if srcs else 0.0
    if not srcs:
        ins.append('B:温度轴无判定且油样不新鲜(同窗无独立源)')

    # ---- C 实物锚 ----
    C = 0.0
    tier = None
    for k in ('tier', 'anchor_tier', '实物锚', 'physical_tier'):
        if prow.get(k):
            tier = str(prow[k])
            break
    if tier:
        C = VC.score_physical(tier)
    else:
        ins.append('C:handoff 无实物锚字段')

    # ---- D 样本充分性 ----
    nw = None
    wr = str(frow.get('窗区间') or '')
    if wr:
        nw = max(1, wr.count('[') // 2 or wr.count(',')) if wr else 1
    nrec = 0
    for k in ('n_records', '记录数', 'n'):
        if isinstance(prow.get(k), (int, float)):
            nrec = int(prow[k])
            break
    D = VC.score_sample(nrec, (nw or 1), False, n_windows=(nw or 1))
    if nw is None:
        ins.append('D:无窗区间解析结果')

    # ---- E 反证搜索：当前无产物承载 ----
    E = 0.0
    ins.append('E:反证搜索无产物承载(没人做过=0 分)')

    axes = {'A_internal': round(A, 1), 'B_independent': round(B, 1),
            'C_physical': round(C, 1), 'D_sample': round(D, 1), 'E_falsification': round(E, 1)}
    return {'axes': axes, 'ins': ins, 'notes': notes}


def score_turbine(frow: dict, prow: dict | None) -> dict:
    r = build_axes(frow, prow)
    conf = VC.confidence(r['axes'])
    total = conf.get('total') if isinstance(conf, dict) else conf
    band = (conf or {}).get('band') if isinstance(conf, dict) else None
    # ★诊断门槛：A/B/C/E 只要能取到的轴缺件, 分数就**不可用于判断机组状态**
    #   （低分反映的是"输入缺件"而不是"证据弱" —— 防止把缺数据读成"证据不足"）
    hard = ('A:', 'B:', 'C:', 'E:')
    missing = [s for s in r['ins'] if s.startswith(hard)]
    usable = (len(missing) == 0)
    caveat = ('' if usable else
              '本分**不可用于判断机组状态**：输入端缺件 ' + str(len(missing)) + ' 项（' + '；'.join(missing) + '）。'
              '低分反映"缺少输入"，不等于"证据弱"；补件后重算才有意义。')
    return {'axes': r['axes'], 'total': total, 'band': band,
            'usable': usable, 'missing_inputs': missing, 'caveat': caveat,
            'ins': r['ins']}


def attach_confidence(ftab, handoff: dict | None = None):
    """给融合表加一列 置信度（dict），不改任何既有列。

    """
    pm = _pt_map(handoff or {})
    out = []
    for _, row in ftab.iterrows():
        d = row.to_dict()
        tid = _norm_tid(d.get('turbine'))
        s = score_turbine(d, pm.get(tid) or pm.get(str(d.get('turbine'))))
        d['置信度'] = s
        out.append(d)
    import pandas as pd
    return pd.DataFrame(out)