|
|
@@ -0,0 +1,63 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+r"""★C02 接线（用户令 2026-10-06）：把**逐窗 level 序列**从多窗 L6 结果里生成，并过原厂迟滞规则。
|
|
|
+
|
|
|
+背景：`sop/fusion_diag.py::apply_hysteresis(window_levels, h=3)` 早已存在（厂商 Hysteresis=3 的通用化：
|
|
|
+连续 h 窗同向才改级，否则保持），但**一直缺入参** —— 全窗裁决不产生"逐窗 level 序列"。
|
|
|
+本模块补上生产端：`outputs/rudong/m5_cms_tcm/model_run_l6.parquet` 现在带 `_窗`（5 个周窗）⇒
|
|
|
+同一 (台, 线) 跨窗的 `定级` 即序列。
|
|
|
+
|
|
|
+口径（如实）：
|
|
|
+- 级别序：`候选·新发`(3) > `候选`(2) > `参考`(1)；未在该窗过闸 ⇒ 该窗**无记录**（不补 0，序列只含实际过闸的窗）。
|
|
|
+- 迟滞由 `fusion_diag.apply_hysteresis` 执行（**不改写它的语义**）；本模块只负责喂序列与呈现。
|
|
|
+- 用途：**诊断/呈现**（显示锁存）；未接入任何判级，除非用户下令。
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import pathlib
|
|
|
+
|
|
|
+RANK = {'候选·新发': 3, '候选': 2, '参考': 1}
|
|
|
+
|
|
|
+
|
|
|
+def window_level_sequences(path=None) -> dict:
|
|
|
+ """返回 {(台, 线): [(窗, 定级, 级别int), ...]}（按窗升序；只含实际过闸的窗）。"""
|
|
|
+ import pandas as pd
|
|
|
+ p = pathlib.Path(path) if path else pathlib.Path(__file__).resolve().parents[3] / \
|
|
|
+ 'outputs/rudong/m5_cms_tcm/model_run_l6.parquet'
|
|
|
+ if not p.exists():
|
|
|
+ return {}
|
|
|
+ df = pd.read_parquet(p)
|
|
|
+ if '定级' not in df.columns or '_窗' not in df.columns:
|
|
|
+ return {}
|
|
|
+ df = df.assign(lv=df['定级'].map(lambda x: RANK.get(str(x), 0)))
|
|
|
+ out = {}
|
|
|
+ for (t, line), g in df.groupby(['台', '线']):
|
|
|
+ s = g.sort_values('_窗')
|
|
|
+ out[(str(t), str(line))] = [(str(w), str(lv), int(i)) for w, lv, i in zip(s['_窗'], s['定级'], s['lv'])]
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def latched(path=None, h: int = 3) -> dict:
|
|
|
+ """对每条序列执行原厂迟滞；返回 {(台, 线): {"raw": [...], "latched": [...], "changed": bool}}。"""
|
|
|
+ from . import fusion_diag as FD
|
|
|
+ out = {}
|
|
|
+ for key, seq in window_level_sequences(path).items():
|
|
|
+ levels = [x[2] for x in seq]
|
|
|
+ try:
|
|
|
+ res = FD.apply_hysteresis(levels, h=h)
|
|
|
+ except Exception as ex: # 实现异常不得静默: 如实带出
|
|
|
+ out[key] = {'err': type(ex).__name__ + ': ' + str(ex)[:80], 'raw': levels}
|
|
|
+ continue
|
|
|
+ lat = res[1] if isinstance(res, tuple) and len(res) > 1 else res
|
|
|
+ out[key] = {'raw': levels, 'latched': list(lat) if hasattr(lat, '__iter__') else lat,
|
|
|
+ 'changed': list(lat) != levels if hasattr(lat, '__iter__') else None}
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def summary(path=None, h: int = 3) -> dict:
|
|
|
+ d = latched(path, h)
|
|
|
+ n = len(d)
|
|
|
+ ch = [k for k, v in d.items() if v.get('changed')]
|
|
|
+ err = [k for k, v in d.items() if v.get('err')]
|
|
|
+ return {'n_seq': n, 'n_changed': len(ch), 'changed': [('%s/%s' % k) for k in ch[:10]],
|
|
|
+ 'n_err': len(err), 'h': h,
|
|
|
+ '口径': '原厂迟滞 Hysteresis=3：连续 3 窗同向才改级；序列只含实际过闸的窗'}
|
zhouyang.xie