deploy_pickup/core/alert_aggregator.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
alert_aggregator.py
-------------------
报警聚合器 - 跨设备聚合抑制 + 分类型冷却

核心功能：
1. 跨设备聚合抑制：同一水厂 N 分钟内 >=M 个设备同时报警 -> 全部抑制（环境噪声）
2. 分类型冷却时间：同类型异常 24 小时冷却，不同类型异常 1 小时冷却
3. 调试模式感知：mode_id=4 时使用短冷却并跳过聚合窗口，方便现场测试
"""

import logging
from datetime import datetime, timedelta
from collections import defaultdict

logger = logging.getLogger(__name__)


class AlertAggregator:

    def __init__(self,
                 push_callback,
                 aggregate_enabled=True,
                 window_seconds=300,
                 min_devices=2,
                 cooldown_same_type_hours=24,
                 cooldown_diff_type_hours=1,
                 mode_provider=None,
                 debug_cooldown_minutes=5):
        # 聚合窗口到期后，符合条件的报警通过此回调实际推送
        self.push_callback = push_callback

        # 跨设备聚合配置
        self.aggregate_enabled = aggregate_enabled
        # 聚合窗口时长（秒），该时间段内收集同一水厂的所有报警
        self.window_seconds = window_seconds
        # 触发抑制的最小设备数，>=此数量则判定为环境噪声
        self.min_devices = min_devices

        # 分类型冷却配置
        # 同一设备同类型异常的冷却时间（小时），防止重复报警
        self.cooldown_same_type_hours = cooldown_same_type_hours
        # 同一设备不同类型异常的冷却时间（小时），允许较快报警新类型
        self.cooldown_diff_type_hours = cooldown_diff_type_hours

        # 项目模式感知：回调返回当前 mode_id（1=日常, 2=参观, 3=检修, 4=调试）
        # mode_id=4 时使用短冷却并跳过聚合窗口
        self._mode_provider = mode_provider
        # 调试模式下同类型异常的冷却时间（分钟），远小于正常的24h
        self._debug_cooldown_minutes = debug_cooldown_minutes

        # 聚合窗口状态
        # key: plant_name
        # value: {"start_time": datetime, "alerts": [alert_info_dict, ...]}
        self.pending_windows = {}

        # 冷却记录
        # key: device_code
        # value: {anomaly_type_code: last_alert_datetime}
        self.cooldown_records = defaultdict(dict)

        logger.info(
            f"报警聚合器已初始化 | "
            f"聚合={'启用' if aggregate_enabled else '禁用'} "
            f"窗口={window_seconds}秒 最小设备数={min_devices} | "
            f"冷却: 同类型={cooldown_same_type_hours}h 不同类型={cooldown_diff_type_hours}h | "
            f"调试冷却={debug_cooldown_minutes}分钟"
        )

    def _is_debug_mode(self):
        # 判断当前是否为调试模式（mode_id=4）
        if self._mode_provider is None:
            return False
        try:
            return self._mode_provider() == 4
        except Exception:
            return False

    def check_cooldown(self, device_code, anomaly_type_code):
        # 检查该设备是否在冷却期内
        # 返回 True 表示冷却中（应抑制），False 表示可以报警
        records = self.cooldown_records.get(device_code, {})
        now = datetime.now()

        if not records:
            # 该设备从未报过警，不在冷却期
            return False

        # 调试模式使用分钟级短冷却，加速现场测试迭代
        is_debug = self._is_debug_mode()

        if anomaly_type_code in records:
            last_time = records[anomaly_type_code]
            elapsed_seconds = (now - last_time).total_seconds()

            if is_debug:
                # 调试模式：同类型异常使用分钟级冷却
                cooldown_seconds = self._debug_cooldown_minutes * 60
                if elapsed_seconds < cooldown_seconds:
                    remaining_min = (cooldown_seconds - elapsed_seconds) / 60
                    logger.info(
                        f"[调试模式] 同类型冷却中: {device_code} | "
                        f"类型={anomaly_type_code} | "
                        f"剩余 {remaining_min:.1f} 分钟"
                    )
                    return True
            else:
                # 正常模式：同类型异常 24 小时冷却
                elapsed_hours = elapsed_seconds / 3600
                if elapsed_hours < self.cooldown_same_type_hours:
                    remaining = self.cooldown_same_type_hours - elapsed_hours
                    logger.info(
                        f"同类型冷却中: {device_code} | "
                        f"类型={anomaly_type_code} | "
                        f"剩余 {remaining:.1f} 小时"
                    )
                    return True
        else:
            if is_debug:
                # 调试模式：不同类型异常零冷却，切换场景时立即可报
                pass
            else:
                # 正常模式：不同类型异常 1 小时冷却
                last_any = max(records.values()) if records else None
                if last_any:
                    elapsed_hours = (now - last_any).total_seconds() / 3600
                    if elapsed_hours < self.cooldown_diff_type_hours:
                        remaining = self.cooldown_diff_type_hours - elapsed_hours
                        logger.info(
                            f"不同类型冷却中: {device_code} | "
                            f"新类型={anomaly_type_code} | "
                            f"剩余 {remaining:.1f} 小时"
                        )
                        return True

        return False

    def record_cooldown(self, device_code, anomaly_type_code):
        # 记录报警时间，用于后续冷却判断
        self.cooldown_records[device_code][anomaly_type_code] = datetime.now()

    def submit_alert(self, plant_name, device_code, anomaly_type_code, push_kwargs):
        # 提交一条报警到聚合队列
        #
        # 参数:
        #     plant_name: 水厂名称（聚合维度）
        #     device_code: 设备编号
        #     anomaly_type_code: 异常类型编码（level_two）
        #     push_kwargs: 原 _push_detection_result 的全部关键字参数
        now = datetime.now()

        # 第一步：检查分类型冷却
        if self.check_cooldown(device_code, anomaly_type_code):
            logger.info(f"报警被冷却抑制: {device_code} | 类型={anomaly_type_code}")
            return

        # 调试模式（mode_id=4）：跳过聚合窗口，单次异常立即推送
        # 调试目的是验证报警链路，不需要等待多设备比对
        if self._is_debug_mode():
            logger.info(f"[调试模式] 直接推送: {device_code} | 类型={anomaly_type_code}")
            self._do_push(device_code, anomaly_type_code, push_kwargs)
            return

        if not self.aggregate_enabled:
            # 聚合禁用时直接推送
            self._do_push(device_code, anomaly_type_code, push_kwargs)
            return

        # 第二步：加入聚合窗口
        if plant_name not in self.pending_windows:
            # 该水厂没有活跃窗口，创建新窗口
            self.pending_windows[plant_name] = {
                "start_time": now,
                "alerts": []
            }
            logger.info(f"聚合窗口已开启: {plant_name} | 窗口={self.window_seconds}秒")

        # 加入队列（去重：同一设备在同一窗口内只保留一次）
        window = self.pending_windows[plant_name]
        existing_devices = {a["device_code"] for a in window["alerts"]}
        if device_code not in existing_devices:
            window["alerts"].append({
                "device_code": device_code,
                "anomaly_type_code": anomaly_type_code,
                "push_kwargs": push_kwargs,
                "submit_time": now
            })
            logger.info(
                f"报警已加入聚合窗口: {plant_name}/{device_code} | "
                f"当前窗口内设备数={len(window['alerts'])}"
            )

    def check_and_flush(self):
        # 定期调用，检查所有聚合窗口是否到期，到期后执行聚合判定
        now = datetime.now()
        expired_plants = []

        for plant_name, window in self.pending_windows.items():
            elapsed = (now - window["start_time"]).total_seconds()
            if elapsed >= self.window_seconds:
                expired_plants.append(plant_name)

        for plant_name in expired_plants:
            window = self.pending_windows.pop(plant_name)
            alerts = window["alerts"]
            device_count = len(alerts)

            if device_count >= self.min_devices:
                # 多设备同时报警 -> 判定为环境噪声，全部抑制
                device_names = [a["device_code"] for a in alerts]
                logger.warning(
                    f"聚合抑制: {plant_name} | "
                    f"{device_count}个设备同时报警，判定为环境噪声 | "
                    f"设备: {', '.join(device_names)}"
                )
            else:
                # 设备数不足 min_devices -> 逐个推送（正常报警）
                for alert in alerts:
                    self._do_push(
                        alert["device_code"],
                        alert["anomaly_type_code"],
                        alert["push_kwargs"]
                    )

    def _do_push(self, device_code, anomaly_type_code, push_kwargs):
        # 实际执行推送，并记录冷却时间
        try:
            self.push_callback(**push_kwargs)
            # 推送成功后记录冷却
            self.record_cooldown(device_code, anomaly_type_code)
            logger.info(f"报警已推送: {device_code} | 类型={anomaly_type_code}")
        except Exception as e:
            logger.error(f"报警推送失败: {device_code} | {e}")