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

# -*- coding: utf-8 -*-
"""goem.py — §1.2 G-OEM 字段语义实证 工具箱 + goem.json builder + 保护码↔通道覆盖矩阵 (P2-1).

Batch2 债 2026-07-02 骨架级抽库。诚实边界: **语义判读 (单位/坐标系/名实/限电方向) 是人工域** —
本模块只出可复核的证据数字 (range/采样率/哨兵/枚举cross-tab), 判读结论由人写进 per 字段 entry。
产物契约以实跑件为准 (outputs/liangshuiquan/sop/goem.json): {'fields': {字段名: {证据...}}}。
coverage_matrix = 盲区前置盘点 (SOP §1.2; 方法论SOP"模态可用性"的机器化): 每保护码→SCADA通道
有/死/无, 盲区按已发损失定价喂"加点建议" (郭家店 80014 入口油温死列=齿箱跳闸#1 恰在线盲,
A24 一台 179MWh 损失落于此盲区 = 加点最强论据)。纯函数无文件 I/O。
"""
import numpy as np
import pandas as pd

DEFAULT_SENTINELS = (-99.0, -999.0, 9999.0, -9999.0, 32767.0, -11000.0)


def field_unit_range(s, *, sentinels=DEFAULT_SENTINELS):
    """字段 range 实证 (§1.2 单位/哨兵): min/p1/median/p99/max + 各哨兵占比。单位判读留人工
    (1000×错=瓦特当千瓦, 阜山戒; 证据=数字, 结论=人)。"""
    v = pd.to_numeric(s, errors='coerce')
    n = int(v.size)
    sent = {str(x): round(float((v == x).mean()), 4) for x in sentinels if bool((v == x).any())}
    v = v.dropna()
    if v.empty:
        return {'n': n, 'n_valid': 0, 'sentinel_shares': sent, 'note': '全 NaN/无有效数'}
    return {'n': n, 'n_valid': int(v.size),
            'min': round(float(v.min()), 3), 'p1': round(float(v.quantile(0.01)), 3),
            'median': round(float(v.median()), 3), 'p99': round(float(v.quantile(0.99)), 3),
            'max': round(float(v.max()), 3), 'sentinel_shares': sent}


def field_sampling_per_month(df, *, time_col):
    """真采样率 per 月实测 (§1.2; 禁全局 median — 阜山 98% 行来自单月1秒, 全局假报"1秒级全年")。"""
    t = pd.to_datetime(df[time_col], errors='coerce').dropna().sort_values()
    if t.empty:
        return {}
    g = t.diff().dt.total_seconds().groupby(t.dt.to_period('M').astype(str))
    return {m: round(float(x.median()), 1) for m, x in g if x.notna().any()}


def rated_plateau(p, *, q=0.995):
    """rated 从高风 plateau 实测 (§1.2; 单 chunk peek 不代表全场, 新艾里 1500→2000 戒)。"""
    v = pd.to_numeric(p, errors='coerce').dropna()
    return round(float(v.quantile(q)), 1) if not v.empty else None


def enum_crosstab(df, *, enum_col, agg_cols):
    """状态/枚举列 cross-tab (§1.2 语义实证主工具): 每枚举值的 n + 各物理量中位 —
    '待机时 rpm=346' 这类矛盾一眼可见 (实跑件即此形)。"""
    out = {}
    for val, g in df.groupby(enum_col, dropna=False):
        row = {'n': int(len(g))}
        for c in agg_cols:
            if c in g.columns:
                row[c] = round(float(pd.to_numeric(g[c], errors='coerce').median()), 1)
        out[str(val)] = row
    return out


def build_goem(fields, *, note=None):
    """组装 goem dict + 非空校验。返回 (goem, issues)。fields = {字段名: {实证方法/证据数字/结论}}。"""
    issues = []
    if not fields or not isinstance(fields, dict):
        issues.append('goem.fields 空 — §1.2 对将用到的每个字段做实证')
    goem = {'fields': fields or {}}
    if note:
        goem['note'] = note
    return goem, issues


def coverage_matrix(codes, *, channel_map, fault_counts=None, loss_mwh=None):
    """保护码↔SCADA通道覆盖矩阵 + 盲区按已发损失定价 (P2-1, SOP §1.2; 2026-07-02)。

    入: codes = {code: 描述}; channel_map = {code: {'channel': 通道名|None, 'state': 'alive'|'dead'|None}}
        (state 由 reference_channel_liveness 逐台/逐列判后汇总给出; 无通道 = channel None);
        fault_counts = {code: 窗内跳闸次数}; loss_mwh = {code: 窗内已发损失 MWh} (EL-5 口径)。
    出: {'rows': [...], 'blind_ranked': [...按 loss→fault_n 降序], 'summary': {...}}。
    盲区 = state != 'alive' (死列与无通道同为在线盲); blind_ranked 直接喂报告"加点/读保护系统日志"建议。
    DP 锚 (郭家店): 80014 入口油温 SCADA 死列 (跳闸#1 276次) + 80001/02 中间轴无通道 (#3 163次)
    = 齿箱跳闸前两位恰在在线盲区; A24 一台 179MWh 在此盲区。
    """
    fault_counts = fault_counts or {}
    loss_mwh = loss_mwh or {}
    rows = []
    for code, desc in codes.items():
        ch = (channel_map or {}).get(code) or {}
        state = ch.get('state') if ch.get('channel') else None
        state_s = {None: '无通道', 'alive': '活', 'dead': '死列'}.get(state, str(state))
        blind = state != 'alive'
        rows.append({'code': str(code), 'desc': desc, 'channel': ch.get('channel'),
                     'state': state_s, 'blind': blind,
                     'fault_n': int(fault_counts.get(code, 0)),
                     'loss_mwh': round(float(loss_mwh.get(code, 0.0)), 1)})
    blind_ranked = sorted((r for r in rows if r['blind']),
                          key=lambda r: (-r['loss_mwh'], -r['fault_n']))
    return {'rows': rows, 'blind_ranked': blind_ranked,
            'summary': {'n_codes': len(rows), 'n_blind': sum(r['blind'] for r in rows),
                        'blind_loss_mwh': round(sum(r['loss_mwh'] for r in rows if r['blind']), 1),
                        'blind_fault_n': sum(r['fault_n'] for r in rows if r['blind'])},
            'caveat': ('盲区加点建议按 loss→fault_n 排序; state 须先经 liveness 判 (死列≠无通道但同盲); '
                       '损失=EL-5 物理估算口径, 禁报表损失列')}