guanlan-src/app_ontology/app_ontology_guanlan/ontology/populate.py

# -*- coding: utf-8 -*-
"""P1 本体铺开: 38台×九系统判级 + 振动handoff + 工单 + 油样 + 机制库 批量入库.

转录纪律: 判级只转录 L1 正本 (taxonomy.system_matrix / fusion_table), 本层零判级权;
每条 Verdict 带审级与 claim_window (C2), 其 Evidence 指回 L1 产物 (C1);
同一部件不同源判级并存不合并 (如 29# 主轴承: L1温度轴'优秀'lite vs TCM-16'定论'双模审 —
NBM变化型对稳定高位失明的活例, 冲突显式可见, 审级高者主导但都保留).
跑法: .venv/bin/python -m src.ontology.populate"""
from src import paths as P
import json, pathlib, sys
import pandas as pd
from .store import Store
from .schema import ConstraintError

ST = P.store()
ONT = P.ont()

# 九系统 → Evidence 观测面 (粗映射, 转录用; 机制级证据另有精确面)
SYS_FACET = {'变桨': '因侧执行链', '偏航': '因侧执行链', '主轴承': '果侧温度', '齿轮箱': '果侧温度',
             '发电机': '果侧电气', '变流器': '果侧电气', '主控与传感网': '测量链面',
             '叶片与叶根': '结构面', '塔架与基础': '结构面'}
L1_WINDOW = '近90日固定窗 (至2026-07-07; 曲线轴=2025H2)'


def _tid(t):
    return f'turbine/{t}'


# ── 部件对象 (惰性建; 工单 resolves / 油样 about 的落点) ──
COMP_NAMES = {
    'mb_lub_pump': '主轴承集中润滑泵', 'gen_lub_pump': '发电机润滑泵', 'yaw_lub_pump': '偏航润滑泵',
    'main_bearing': '主轴承(未分端)', 'main_bearing_fore': '前主轴承', 'main_bearing_aft': '后主轴承(齿轮箱侧)',
    'gearbox': '主齿轮箱', 'generator': '发电机',
    'gen_bearing_fore': '发电机前轴承', 'gen_bearing_aft': '发电机后轴承',
    'pitch_system': '变桨系统', 'pitch_bearing_a': '变桨轴承A', 'pitch_bearing_b': '变桨轴承B',
    'pitch_bearing_c': '变桨轴承C',
    'yaw_system': '偏航系统', 'yaw_gearbox': '偏航减速器', 'yaw_bearing': '偏航轴承', 'yaw_hydraulic': '偏航液压站',
    'hydraulic_station': '液压站',
}


def _ensure_component(s, t, key):
    """component/{t}/{key} 不存在则建 (链回台); 已存在 (如 seed 的 WTG29 两件) 直接复用不覆盖."""
    cid = f'component/{t}/{key}'
    if cid not in s.objects:
        s.put(dict(id=cid, type='Component', props=dict(name=COMP_NAMES[key]),
                   links=dict(of=[_tid(t)])))
    return cid


def _wo_component_key(txt):
    """工单文本 → 部件 key. 只做确定性强的关键词映射, 含糊不建 (宁缺勿滥).
    顺序 = 具体优先: 润滑泵必须先归属主体 (偏航/发电机/主轴承各有润滑泵, 单写'润滑泵'不知道哪个)."""
    if ('润滑' in txt) or ('油脂' in txt):
        if '主轴' in txt:
            return 'mb_lub_pump'
        if '发电机' in txt:
            return 'gen_lub_pump'
        if '偏航' in txt:
            return 'yaw_lub_pump'
        return None                      # 归属不明的润滑工单: 不建 (宁缺勿滥)
    if '主轴承' in txt:
        return 'main_bearing'
    if '齿轮箱' in txt:
        return 'gearbox'
    if '变桨' in txt:
        return 'pitch_system'
    if '偏航' in txt:
        return 'yaw_system'
    if '发电机' in txt:                   # '偏航电机/变桨电机' 不含'发电机'三连字, 已被上两支接走
        return 'generator'
    if '液压' in txt:
        return 'hydraulic_station'
    return None


# 油样 comp 字段 (结构化, 18 种取值全部确定) → 部件 key; 同义归一
OIL_COMP_KEY = {
    '主轴前': 'main_bearing_fore', '主轴前轴承': 'main_bearing_fore',
    '主轴后': 'main_bearing_aft', '主轴后轴承': 'main_bearing_aft',
    '主齿轮箱': 'gearbox',
    '发电机前': 'gen_bearing_fore', '发电机前轴承': 'gen_bearing_fore', '发电机轴承前': 'gen_bearing_fore',
    '发电机后': 'gen_bearing_aft', '发电机后轴承': 'gen_bearing_aft', '发电机轴承后': 'gen_bearing_aft',
    '液压站': 'hydraulic_station',
    '偏航减速器': 'yaw_gearbox', '偏航液压站': 'yaw_hydraulic', '偏航轴承': 'yaw_bearing',
    '变桨轴承A': 'pitch_bearing_a', '变桨轴承B': 'pitch_bearing_b', '变桨轴承C': 'pitch_bearing_c',
}


def populate():
    sys.path.insert(0, '.')
    from src.windscada import taxonomy
    from src.windscada.subsys import fusion

    s = Store(ONT / 'objects.json')
    n0 = len(s.objects)

    # ── Turbine 38 台 ──
    from src.windscada.config import farm
    cfg = farm()
    for t in cfg['turbines']:
        if _tid(t) not in s.objects:
            s.put(dict(id=_tid(t), type='Turbine', props=dict(name=t.replace('WTG', '') + '#',
                                                              model='SWT-4.0-130', farm='rudong')))

    # F7: Decision 显式分型在 store.put (写入闸层, 所有模块统一) — 见 store._decision_kind
# ── 九系统判级 (L1 正本 = taxonomy.system_matrix; 全状态含优秀/良好 = 全景转录) ──
    mx = taxonomy.system_matrix()
    for t, row in mx.items():
        for sysn, v in row.items():
            eid = f'evidence/L1/{t}/{sysn}'
            # C4 三分判读: L1 依据是多轴拼接串 (";" 分隔), 交叉性 = 有几个独立轴同时非常态。
            # 单轴 → 不可判 (单源不构成交叉印证, C3 同理); 多轴同向 → 支持;
            # ★"各轴无非常态" 也是"不可判"而非"排除" — 判据面没测到 ≠ 该部件被证明无缺陷
            #   (监测面覆盖有限, 见 detectability_prior 盲区)。
            _basis = str(v.get('依据', ''))
            _axes = [x for x in _basis.split(';') if x.strip() and '无非常态' not in x]
            if len(_axes) >= 2:
                _cc, _ccn = '支持', f'{len(_axes)} 个轴同时非常态 (轴: {"; ".join(a.strip()[:26] for a in _axes[:3])})'
            elif len(_axes) == 1:
                _cc, _ccn = '不可判', '仅单轴非常态 — 单源不构成交叉印证 (需第二个独立观测面)'
            else:
                _cc, _ccn = '不可判', '本层各轴无非常态 — 判据面未测到 ≠ 已排除 (监测面覆盖有限, 见盲区先验)'
            s.put(dict(id=eid, type='Evidence',
                       props=dict(facet=SYS_FACET.get(sysn, '外部记录'),
                                  source=f'src/windscada taxonomy.system_matrix[{sysn}] (L1 判据层)',
                                  claim_window=L1_WINDOW, summary=_basis[:200],
                                  crosscheck=_cc, crosscheck_note=_ccn,
                                  crosscheck_facet='L1 多轴合成 (同层多轴, 非第三方独立源)'),
                       links=dict(about=[_tid(t)])))
            s.put(dict(id=f'verdict/L1/{t}/{sysn}', type='Verdict',
                       props=dict(level=v['状态'], system=sysn, review_tier='lite',
                                  review_note='L1 自动判级 (筛查级/候选, 非定谳; SOP 审级=lite)',
                                  claim_window=L1_WINDOW),
                       links=dict(evidencedBy=[eid], about=[_tid(t)])))

    # ── 振动 handoff 判级 (28 台次, fusion_table 已带 component/claim_window/verdict 原文) ──
    ftab, _ = fusion.fusion_table()
    # F6 修 (Critic 逮): 原 id 用 DataFrame 位置索引 i — handoff 增删一行, 所有后续行的 id 整体错位,
    # 已落库的 decision/links 会静默指向别的台次。改内容派生 id (台+部件槽), 与行序无关。
    def _slot(comp):
        import re as _re
        return _re.sub(r'[^0-9A-Za-z\u4e00-\u9fa5]+', '', str(comp))[:16] or 'na'
    for i, r in ftab.iterrows():
        eid = f'evidence/vib/{r.turbine}/{_slot(r.部件)}'
        # C4 三分判读 (审计 ④ 修): 交叉面 = 温度轴 (fusion 表已给该部件的 dev 与判).
        # ★"无信号" 必须写"不可判"不能写"排除" — 温度对早期轴承缺陷本就不敏感 (fleet σ 2.7~3.4K
        #   只能检出 >3K 的严重热负荷), 写"排除"会让下游误以为假设已被否定而放弃现场复查
        #   (memory: 不支持≠无信号)。
        _vlv = fusion.verdict_class(str(r.振动结论))[1]
        _tdev = r.温度dev if '温度dev' in ftab.columns else None
        _tvd = str(r.温度判 or '') if '温度判' in ftab.columns else ''
        if _vlv in ('bad', 'warn'):
            if _tdev is not None and _tdev == _tdev and abs(float(_tdev)) >= 3.0 and '候选' in _tvd:
                _cc, _ccn = '支持', f'温度轴同向: {r.温度通道} {_tdev:+.1f}K [{_tvd}] — 双面同向'
            elif _tdev is not None and _tdev == _tdev and float(_tdev) <= -3.0:
                _cc, _ccn = '不可判', f'温度轴反向 ({r.温度通道} {_tdev:+.1f}K) 但温度对早期缺陷分辨力不足 (fleet σ 2.7~3.4K) — 不构成排除'
            else:
                _cc, _ccn = '不可判', ('温度轴无信号' + (f' ({r.温度通道} {_tdev:+.1f}K [{_tvd or "—"}])' if _tdev == _tdev else '')
                                    + ' — 分辨力不足非阴性; 早期轴承缺陷在温度面本就不显')
        else:
            _cc, _ccn = '不可判', '振动结论非报警级, 交叉面不适用 (单源不构成交叉印证)'
        s.put(dict(id=eid, type='Evidence',
                   props=dict(facet='果侧振动稳态',
                              source='振动线 handoff_vibration_v2.json (正本=m5_cms_tcm findings; 转录粗分面, 精确面见机制对象)',
                              claim_window=str(r.claim_window), summary=str(r.振动信号)[:200],
                              crosscheck=_cc, crosscheck_note=_ccn,
                              crosscheck_facet='果侧温度 (fusion 表同部件通道)'),
                   links=dict(about=[_tid(r.turbine)])))
        # F3 修 (独立 Critic 逮): 振动线 v2.1 显式标注"引用同族/谱线级/测量链问题/未映射"的行
        # **不作部件级结论消费** — 原实现一律提成 Verdict, 等于把"这条不能当部件结论"的行
        # 变成了 28 条里 15 条的部件级判级, 下游(排程/风险/矩阵)会把它们当真判级用。
        # 改: 这类行仍转录为 Evidence(证据在, 不丢), 但降为 props.consumable=False 的 Observation 级 Verdict,
        # 并显式记 not_component_level 原因; 判级类查询(ont_maint_plan/ont_risk)只取 consumable 行。
        _comp = str(r.部件)
        _nc = next((k for k in ('引用同族', '谱线级', '测量链', '未映射') if k in _comp), None)
        s.put(dict(id=f'verdict/vib/{r.turbine}/{_slot(r.部件)}', type='Verdict',
                   props=dict(level=str(r.振动结论), system='振动', component=_comp,
                              review_tier='双模审',
                              review_note='振动线历审 (Gemini+DeepSeek + 双线互复现), 本层转录',
                              consumable=(_nc is None),
                              not_component_level=_nc,
                              consume_note=(None if _nc is None else
                                            f'振动线标注"{_nc}" ⇒ 不作部件级结论消费 (证据保留, 判级不下游)'),
                              claim_window=str(r.claim_window)),
                   links=dict(evidencedBy=[eid], about=[_tid(r.turbine)])))

    # ── 工单 1651 条 → WorkOrder (检修事件语义; 原始行留 L0) ──
    # resolves 链 (审计缺链①): 工单文本含确定性部件关键词 → links.resolves→component/;
    # 匹配面 = 故障名称+位置二级+元器件+维修对象, 含糊不建 (宁缺勿滥)
    w = pd.read_parquet(ST / 'workorders.parquet')
    w = w[w.t_report > '2019']
    n_dirty_loss, n_resolves = 0, 0
    for i, r in w.iterrows():
        raw_loss = r['发电量损失（kWh）']
        loss_kwh, loss_note = None, None
        if pd.notna(raw_loss):
            try:
                loss_kwh = float(raw_loss)
            except (ValueError, TypeError):        # 脏文本混数值列 (如"损失电量统计在不间断电源故障内") — 原文保留不静默丢
                loss_note = str(raw_loss)[:60]; n_dirty_loss += 1
        t = str(r.turbine)
        wo_txt = '|'.join(str(r[c]) for c in ('故障名称', '故障位置二级', '元器件名称', '维修对象')
                          if pd.notna(r[c]))
        links = dict(about=[_tid(t)])
        ckey = _wo_component_key(wo_txt)
        if ckey:
            links['resolves'] = [_ensure_component(s, t, ckey)]
            n_resolves += 1
        s.put(dict(id=f'workorder/{i}', type='WorkOrder',
                   props=dict(turbine=t, date=str(r.t_report)[:10],
                              name=str(r['故障名称'])[:60], loc2=str(r['故障位置二级']),
                              action=str(r['维修动作']) if pd.notna(r['维修动作']) else None,
                              part=str(r['元器件名称']) if pd.notna(r['元器件名称']) else None,
                              loss_kwh=loss_kwh, loss_note=loss_note),
                   links=links))
    if n_dirty_loss:
        print(f'  ⚠ 工单损失列脏文本 {n_dirty_loss} 条 → loss_note 保留原文')
    print(f'  工单 resolves 链 {n_resolves}/{len(w)} 条 (确定性关键词映射, 含糊不建)')

    # ── 油样 506 份索引 → OilSample (审计缺链②: comp 字段结构化确定 → about 补 component/ 链) ──
    oil = pd.read_parquet(ST / 'oil_samples_index.parquet')
    n_oil_comp = 0
    for _, r in oil.iterrows():
        t = str(r.turbine)
        about = [_tid(t)]
        okey = OIL_COMP_KEY.get(str(r.comp))
        if okey:
            about.append(_ensure_component(s, t, okey))
            n_oil_comp += 1
        s.put(dict(id=f'oilsample/{r.sample_id}', type='OilSample',
                   props=dict(turbine=t, comp=str(r.comp), report_date=str(r.date),
                              note='文件名=报告日期, 取样日≈报告日−1~2周; 数值按需逐份解析'),
                   links=dict(about=about)))
    print(f'  油样→部件链 {n_oil_comp}/{len(oil)} 份')

    # ── 机制库 (findings/注记提炼; 29# 润滑断供已入 P0, 此处补 8 族) ──
    # F9: fleet 级机制的 affects 从数据实算台清单 (mblub_alarm_cross 的 "2026窗78条/21台" 口径),
    # 不留空链接 (空 affects = 查询游走不可达); 台号 data-driven 非硬编码
    a = pd.read_parquet(ST / 'alarms.parquet')
    fleet_64001 = sorted(a[(a.code == '64001') & (a.t_on >= '2026-01-01')].turbine.unique())
    MECHS = [
        dict(id='mechanism/rudong/dew_heater_stuck_on', name='防凝露加热器卡开 (温控执行器卡死家族)',
             instances=['WTG31'], status='候选 (跨场家族 n=3+1)',
             facets=['果侧温度', '因侧执行链'],
             source='memory thermal-regulation-stuck-family; 夏duty高×低湿窗×整窗运行三证'),
        dict(id='mechanism/rudong/gende_bpfi_135', name='发电机DE内圈缺陷族 (BPFI 135Hz, OEM表锚定)',
             instances=['WTG16', 'WTG33'], status='预警 (16#处置规划/33#预警; 谱线过六闸)',
             facets=['果侧振动稳态'],
             source='振动线 handoff v1_generator; 包络域显著+原始谱不显著=早期轴承缺陷预期签名'),
        dict(id='mechanism/rudong/accum_fail_recharge_heat', name='变桨蓄能器失效→频繁充压→液压油温高',
             instances=['WTG20', 'WTG08', 'WTG03'], status='候选 (三向重合: 油温高簇×蓄能报警×带宽)',
             facets=['因侧执行链', '果侧温度'],
             source='temp_overview 互证注记 2026-08-25 订正版; 19#油温高但蓄能三轴净=单列待查不并入'),
        dict(id='mechanism/rudong/gen_de_hot_24', name='24# 发电机DE轴承高温 (NDE−DE=−20K)',
             instances=['WTG24'], status='候选 (与2025-12换双轴承史并读)',
             facets=['果侧温度'],
             source='temp_nbm pairs_table; 端别按西门子IO表定案 R=DE'),
        dict(id='mechanism/rudong/gbx_bypass_filter_dp', name='齿轮箱旁路过滤器压力高 (码4312)',
             instances=['WTG27'], status='候选 (27#累计1082h 全场集中; 机制未定: 滤芯堵/油温粘度/传感)',
             facets=['因侧执行链'],
             source='alarms.parquet 码4312; 油路压力必须油温匹配 (memory oil-circuit-pressure)'),
        dict(id='mechanism/rudong/mb_grease_low_systemic', name='主轴承油脂水平低系统性背景 (码64001)',
             instances=fleet_64001, status='系统性背景 (非个体; 报警条数排序无意义)',
             facets=['因侧执行链'], scope_note=f'2026窗报警台清单 (alarms.parquet 实算, {len(fleet_64001)}台)',
             source='mblub_alarm_cross.json: 2026窗78条/21台, 与柱塞无显著关联(19/45 p=0.41), 个体升级须走自身纵向机制链'),
        dict(id='mechanism/rudong/ims_broadband_31', name='31# 齿轮箱中速轴宽带高位 (不命名滚道)',
             instances=['WTG31'], status='预警 (宽带持续高位, 离散谱线无独立选择性 → 只判区域不命名部件)',
             facets=['果侧振动稳态'],
             source='振动线 v1_gearbox; 拒绝命名=限制错误传播半径的防污染闸'),
        dict(id='mechanism/rudong/yaw_drive_alarm_family', name='偏航驱动报警族',
             instances=['WTG29'], status='报警 (L1偏航轴; 29#偏航判报警的驱动维)',
             facets=['因侧执行链'],
             source='taxonomy.system_matrix 偏航轴; 与主轴承润滑机制独立'),
    ]
    for m in MECHS:
        affects = [_tid(t) for t in m['instances'] if t.startswith('WTG')]
        if not affects:
            raise ValueError(f"F9: 机制 {m['id']} affects 为空 — 空链接=游走不可达, 转录处必须给台/系统级落点")
        props = dict(name=m['name'], status=m['status'], facets=m['facets'],
                     independent_source_count=len(set(m['facets'])),   # C3: 源数=facet 去重数
                     source=m['source'])
        if 'scope_note' in m:
            props['scope_note'] = m['scope_note']
        s.put(dict(id=m['id'], type='FailureMechanism', props=props,
                   links=dict(affects=affects)))

    p = s.save()
    return p, n0, len(s.objects)


if __name__ == '__main__':
    # 控制台可能是 GBK(中文 Windows 代码页 936): 正文里的 ✔ ✗ ✅ ⚠ 这类字符编不出来会抛
    # UnicodeEncodeError, 脚本干成了事却以退出码 1 结束(同类坑见 src/console.py)。降级为 '?' 而不是崩;
    # 不用 import 是为了兼顾 python -m 与直接当脚本跑两种启动方式。
    import sys as _sys
    for _s in (_sys.stdout, _sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    p, n0, n1 = populate()
    print(f'✔ P1 铺开: {n0} → {n1} 对象 ({p})')