vent_agent_2.0/extensions/main_fan/workflow_monitoring.py

"""按冻结阈值判定现成监测数值；模型不能降低已确认的告警等级。"""
from copy import deepcopy
import math

VERSION = 'main-fan-monitoring-rules-v1'
RANK = {'NORMAL': 0, 'WARNING': 1, 'ABNORMAL': 2}


def finite(value):
    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)


def number(value):
    return format(value, '.8g')


def classify(value, normal, severe):
    if finite(severe) and value >= severe:
        return 'ABNORMAL'
    return 'WARNING' if value > normal else 'NORMAL'


def assess(monitoring, rules):
    """仅判定预处理已确认可比的比例和停启间隔，不重选点或重新计算比例。"""
    normal_flow = rules['switch']['flow_deviation_percent']
    normal_interval = rules['switch']['maximum_switch_interval_minutes']
    severe_flow = rules.get('severity', {}).get('abnormal_flow_deviation_percent')
    severe_interval = rules.get('severity', {}).get('abnormal_switch_interval_minutes')
    for threshold in (normal_flow, normal_interval):
        if not finite(threshold) or threshold < 0:
            raise ValueError('监测正常阈值必须为非负有限数值')
    for threshold in (severe_flow, severe_interval):
        if threshold is not None and (not finite(threshold) or threshold < 0):
            raise ValueError('已配置的监测严重阈值必须为非负有限数值')
    events, lines, minimum = [], [], 'NORMAL'
    units = {row['fan']: row.get('unit', '') for row in monitoring.get('flows', [])}
    for candidate in monitoring.get('switch_candidates', []):
        comparison = candidate.get('comparison', {})
        before, after, change = (comparison.get(key) for key in
                                 ('before_value', 'after_value', 'change_percent'))
        event = {key: candidate.get(key) for key in ('id', 'from_fan', 'to_fan', 'time_range')}
        period = '～'.join(candidate.get('time_range', []))
        label = f'{period}（{candidate["from_fan"]}→{candidate["to_fan"]}）'
        flow = {'result_code': None, 'before_value': before, 'after_value': after,
                'change_percent': change, 'reason': comparison.get('reason', '')}
        if (all(finite(value) for value in (before, after, change)) and before > 0
                and after >= 0 and change >= 0 and not comparison.get('reason')):
            flow['result_code'] = classify(change, normal_flow, severe_flow)
            if flow['result_code'] == 'ABNORMAL':
                verdict = f'达到{number(severe_flow)}%严重超限阈值'
            elif flow['result_code'] == 'WARNING':
                verdict = f'超过{number(normal_flow)}%一般超限阈值'
            else:
                verdict = f'未超过{number(normal_flow)}%一般超限阈值'
            lines.append(f'{label}：切换前后参考风量{number(before)}→{number(after)} '
                         f'{units.get(candidate["from_fan"], "")}，变化率{number(change)}%，{verdict}。')
        else:
            lines.append(f'{label}：本次没有可计算变化率的前后风量对照，不据此告警。')
        duration = candidate.get('duration_minutes')
        bounds = candidate.get('duration_bounds_minutes') or {}
        status = candidate.get('duration_status')
        has_gaps = candidate.get('duration_has_gaps', False)
        interval = {'result_code': None, 'duration_minutes': duration,
                    'duration_bounds_minutes': bounds, 'duration_has_gaps': has_gaps,
                    'duration_status': status, 'comparison_value_minutes': None}
        if finite(duration) and duration >= 0:
            # 估算仅以采样下界作为告警依据，缺测/未知及跨阈值范围不能变成告警。
            confirmed = duration if status == 'provided' else bounds.get('lower') if status == 'estimated' else None
            if (not has_gaps and finite(confirmed) and confirmed >= 0
                    and (status != 'estimated' or confirmed <= duration)):
                interval['comparison_value_minutes'] = confirmed
                interval['result_code'] = classify(confirmed, normal_interval, severe_interval)
            text = f'{label}：按开停观测估算停启间隔约{number(duration * 60)}秒'
            overlap = candidate.get('overlap_seconds')
            if finite(overlap) and overlap > 0:
                text += f'，先启后停的观测重叠约{number(overlap)}秒'
            if interval['result_code'] == 'ABNORMAL':
                text += f'，可靠间隔下界达到{number(severe_interval)}分钟严重超限阈值'
            elif interval['result_code'] == 'WARNING':
                text += f'，可靠间隔下界超过{number(normal_interval)}分钟一般超限阈值'
            elif interval['result_code'] == 'NORMAL' and duration <= normal_interval:
                text += f'，该估算值未超过{number(normal_interval)}分钟阈值'
            else:
                text += '，采样范围或缺测限制使其不能作为间隔超限依据'
            lines.append(text + '。')
        else:
            lines.append(f'{label}：未取得可配对的停启间隔，不据此告警。')
        event.update(flow=flow, interval=interval)
        events.append(event)
        for item in (flow, interval):
            if item['result_code'] and RANK[item['result_code']] > RANK[minimum]:
                minimum = item['result_code']
    summary = {'ABNORMAL': '监测数据达到严重超限阈值，研判为异常。',
               'WARNING': '监测数据超过一般阈值，研判为预警。',
               'NORMAL': '现有监测数据未形成明确超限告警。'}[minimum]
    limitations = ('风量为切换前后各1小时附近最近单点对照，零值按原样保留；'
                   '开停间隔为采样观测估算，均不代表已核实连续稳定工况。')
    if any('物理对应尚未确认' in note for note in monitoring.get('notes', [])):
        limitations += '两系统风机按原编号对照，物理对应尚未确认。'
    date_range = '至'.join(monitoring.get('range', []))
    return {'version': VERSION, 'minimum_result_code': minimum,
            'thresholds': {'normal_flow_percent': normal_flow, 'severe_flow_percent': severe_flow,
                           'normal_interval_minutes': normal_interval, 'severe_interval_minutes': severe_interval},
            'events': events, 'summary': summary,
            'reason': '\n'.join([summary, *lines, limitations,
                '性能报告作为辅助核对，不抵消以上已确认的监测超限。'] if minimum != 'NORMAL'
                else [summary, *lines, limitations]),
            'data_detail': f'监测时段：{date_range}。开停和风量来自本轮固定预处理，共{len(events)}个切换候选，'
                           f'{sum(row["flow"]["result_code"] is not None for row in events)}组有效风量变化率对照。'}


def resolve(model_decision, assessment):
    """保留模型原件；明确监测告警的编码、摘要和正文由固定证据生成。"""
    final = deepcopy(model_decision)
    minimum = assessment['minimum_result_code']
    model_code = model_decision['judgment']['resultCode']
    if minimum != 'NORMAL':
        if RANK.get(model_code, -1) <= RANK[minimum]:
            # 同时换掉矛盾摘要/正文，不能仅改编码仍保留“运行正常”的旧解释。
            final['dataDetail'] = assessment['data_detail'] + '性能报告及模型辅助分析随本轮原始输入留档。'
            final['judgment'].update(resultCode=minimum, result=assessment['summary'], reason=assessment['reason'])
        else:
            # 性能有清晰证据时仍可提高等级，不能被较低的监测等级覆盖。
            final['judgment']['result'] = assessment['summary'] + model_decision['judgment']['result']
            final['judgment']['reason'] = assessment['reason'] + '\n\n模型辅助分析：\n' + model_decision['judgment']['reason']
    return final, {'version': VERSION, 'minimum_result_code': minimum,
                   'model_result_code': model_code, 'final_result_code': final['judgment']['resultCode'],
                   'monitoring_enforced': minimum != 'NORMAL',
                   'model_decision': model_decision, 'final_decision': final}