|
|
@@ -21,6 +21,67 @@ 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
|
|
|
|
|
|
|
|
|
+
|
|
|
+# ── ★② 真输入装载（2026-10-06）：A 轴 ← L6 过闸谱线表；C 轴 ← 实物锚登记 ───────────────
|
|
|
+# 说明：这两样都是**仓内既有产物**（非新增数据源）：
|
|
|
+# A: outputs/rudong/m5_cms_tcm/model_run_l6.parquet（逐台逐线"过闸"结果, 列含 _闸/_依据/定级/绝对锚/_窗）
|
|
|
+# 口径映射（保守, 已在 notes 写明）：某台有 ≥1 条 PASS 线 ⇒ 视为"该台走通了全闸"⇒ A 轴按 5 闸计；
|
|
|
+# 绝对锚：该台任一行"解封判据"命中 physical_in_window 或"绝对锚"列非「—」⇒ anchor_ok=True。
|
|
|
+# C: reference/rudong/positive_anchors.json 的 anchors（tier ∈ physical_in_window/closed_loop/physical_historical）
|
|
|
+# 同一台取**最高档**（20 > 12 > 8）。
|
|
|
+
|
|
|
+
|
|
|
+def load_axis_inputs(root=None) -> dict:
|
|
|
+ """返回 {turbine: {"a":{"gates_passed":int,"anchor_ok":bool,"pass_lines":int,"level":str},
|
|
|
+ "c":{"tier":str,"evidence":str}}}（容错: 缺件则空 dict）。"""
|
|
|
+ import json
|
|
|
+ import pathlib as _p
|
|
|
+ base = _p.Path(root) if root else _p.Path(__file__).resolve().parents[3]
|
|
|
+ out: dict = {}
|
|
|
+ # A: L6 过闸谱线表
|
|
|
+ try:
|
|
|
+ import pandas as _pd
|
|
|
+ pq = base / "outputs/rudong/m5_cms_tcm/model_run_l6.parquet"
|
|
|
+ if pq.exists():
|
|
|
+ df = _pd.read_parquet(pq)
|
|
|
+ for tid, g in df.groupby("台"):
|
|
|
+ key = _norm_tid(tid)
|
|
|
+ passes = int((g["_闸"].astype(str).str.upper() == "PASS").sum())
|
|
|
+ anchor = bool(g["解封判据"].astype(str).str.contains("physical_in_window").any()
|
|
|
+ or (g.get("绝对锚") is not None
|
|
|
+ and g["绝对锚"].astype(str).ne("—").any()))
|
|
|
+ lv = "候选" if (g["定级"].astype(str) == "候选").any() else "参考"
|
|
|
+ out.setdefault(key, {})["a"] = {"gates_passed": 5 if passes else 0,
|
|
|
+ "anchor_ok": anchor, "pass_lines": passes, "level": lv}
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ # C: 实物锚登记（同台取最高档）
|
|
|
+ try:
|
|
|
+ ap = base / "reference/rudong/positive_anchors.json"
|
|
|
+ if ap.exists():
|
|
|
+ doc = json.loads(ap.read_text(encoding="utf-8"))
|
|
|
+ rank = {"physical_in_window": 3, "closed_loop": 2, "physical_historical": 1}
|
|
|
+ for a in (doc.get("anchors") or []):
|
|
|
+ key = _norm_tid(a.get("turbine"))
|
|
|
+ tier = str(a.get("tier") or "")
|
|
|
+ cur = (out.setdefault(key, {}).get("c") or {}).get("tier")
|
|
|
+ if rank.get(tier, 0) > rank.get(cur or "", 0):
|
|
|
+ out[key]["c"] = {"tier": tier, "evidence": str(a.get("evidence") or "")[:80],
|
|
|
+ "component_class": str(a.get("component_class") or "")}
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def _n_win(frow: dict) -> int:
|
|
|
+ """从融合表行的 窗区间 解析窗数（容错；解析不出按 1 窗）。"""
|
|
|
+ wr = str(frow.get('窗区间') or '')
|
|
|
+ try:
|
|
|
+ return max(1, wr.count('[') // 2)
|
|
|
+ except Exception:
|
|
|
+ return 1
|
|
|
+
|
|
|
+
|
|
|
def _pt_map(handoff: dict) -> dict:
|
|
|
"""handoff['per_turbine'] 是列表 ⇒ 归一成 {turbine: row}（容错）。"""
|
|
|
out = {}
|
|
|
@@ -44,26 +105,38 @@ def _norm_tid(x) -> str:
|
|
|
return 'WTG%02d' % int(m.group(1))
|
|
|
|
|
|
|
|
|
-def build_axes(frow: dict, prow: dict | None) -> dict:
|
|
|
- """按可用数据装配五轴；缺数轴标 [INS] 并计 0。返回 {axes, ins, notes}。"""
|
|
|
+def build_axes(frow: dict, prow: dict | None, axin: dict | None = None) -> dict:
|
|
|
+ """按可用数据装配五轴；缺数轴标 [INS] 并计 0。axin = load_axis_inputs() 的逐台真输入。"""
|
|
|
prow = prow or {}
|
|
|
+ axin = axin or {}
|
|
|
notes: list[str] = []
|
|
|
ins: list[str] = []
|
|
|
|
|
|
- # ---- A 内部证据强度：找 handoff 里的"闸"过数 ----
|
|
|
- a = None
|
|
|
+ # ---- A 内部证据强度：优先用 L6 过闸谱线的真输入 ----
|
|
|
+ _a = (axin.get('a') or {})
|
|
|
+ if _a:
|
|
|
+ A = VC.score_internal(int(_a.get('gates_passed') or 0), n_gates=5,
|
|
|
+ anchor_ok=bool(_a.get('anchor_ok')),
|
|
|
+ persist_windows=int(prow.get('persist_windows') or 0),
|
|
|
+ n_windows=int(prow.get('n_windows') or _n_win(frow)))
|
|
|
+ notes.append('A 来源=L6 过闸谱线: PASS 线 %s 条 · 绝对锚 %s · 定级 %s'
|
|
|
+ % (_a.get('pass_lines'), _a.get('anchor_ok'), _a.get('level')))
|
|
|
+ # 直接跳到 B（跳过下面的 handoff 取数）
|
|
|
+ a = None if False else 'FROM_L6'
|
|
|
+ else:
|
|
|
+ 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 无逐台闸过数')
|
|
|
+ ins.append('A:本场无 L6 过闸记录且 handoff 无逐台闸过数')
|
|
|
A = 0.0
|
|
|
- else:
|
|
|
+ elif a != 'FROM_L6':
|
|
|
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))
|
|
|
+ n_windows=int(prow.get('n_windows') or _n_win(frow)))
|
|
|
|
|
|
# ---- B 独立源共时确认（同窗）----
|
|
|
srcs = set()
|
|
|
@@ -74,7 +147,7 @@ def build_axes(frow: dict, prow: dict | None) -> dict:
|
|
|
srcs.add('oil')
|
|
|
B = VC.score_independent(srcs) if srcs else 0.0
|
|
|
if not srcs:
|
|
|
- ins.append('B:温度轴无判定且油样不新鲜(同窗无独立源)')
|
|
|
+ ins.append('B(结构性):同窗独立源不在窗(温度轴判「—」/油样 stale) —— 字段级现实, 非逐台缺陷')
|
|
|
|
|
|
# ---- C 实物锚 ----
|
|
|
C = 0.0
|
|
|
@@ -83,10 +156,15 @@ def build_axes(frow: dict, prow: dict | None) -> dict:
|
|
|
if prow.get(k):
|
|
|
tier = str(prow[k])
|
|
|
break
|
|
|
+ if not tier:
|
|
|
+ _c = (axin.get('c') or {})
|
|
|
+ if _c.get('tier'):
|
|
|
+ tier = str(_c['tier'])
|
|
|
+ notes.append('C 来源=实物锚登记: %s（%s）' % (tier, _c.get('evidence', '')[:40]))
|
|
|
if tier:
|
|
|
C = VC.score_physical(tier)
|
|
|
else:
|
|
|
- ins.append('C:handoff 无实物锚字段')
|
|
|
+ ins.append('C:本台无实物锚登记且 handoff 无该字段')
|
|
|
|
|
|
# ---- D 样本充分性 ----
|
|
|
nw = None
|
|
|
@@ -111,22 +189,23 @@ def build_axes(frow: dict, prow: dict | None) -> dict:
|
|
|
return {'axes': axes, 'ins': ins, 'notes': notes}
|
|
|
|
|
|
|
|
|
-def score_turbine(frow: dict, prow: dict | None) -> dict:
|
|
|
- r = build_axes(frow, prow)
|
|
|
+def score_turbine(frow: dict, prow: dict | None, axin: dict | None = None) -> dict:
|
|
|
+ r = build_axes(frow, prow, axin)
|
|
|
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:')
|
|
|
+ hard = ('A:', 'C:', 'E:') # B 属结构性缺件, 单列
|
|
|
missing = [s for s in r['ins'] if s.startswith(hard)]
|
|
|
+ structural = [s for s in r['ins'] if s.startswith('B(')]
|
|
|
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']}
|
|
|
+ 'usable': usable, 'missing_inputs': missing, 'structural': structural,
|
|
|
+ 'caveat': caveat, 'ins': r['ins']}
|
|
|
|
|
|
|
|
|
def attach_confidence(ftab, handoff: dict | None = None):
|
|
|
@@ -134,11 +213,14 @@ def attach_confidence(ftab, handoff: dict | None = None):
|
|
|
|
|
|
"""
|
|
|
pm = _pt_map(handoff or {})
|
|
|
+ axin = load_axis_inputs()
|
|
|
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'))))
|
|
|
+ _ax = dict(axin.get(tid) or {})
|
|
|
+ _c = (pm.get(tid) or pm.get(str(d.get('turbine'))) or {})
|
|
|
+ s = score_turbine(d, _c, _ax)
|
|
|
d['置信度'] = s
|
|
|
out.append(d)
|
|
|
import pandas as pd
|
zhouyang.xie