reverse_osmosis_model/smart_monitor.py

# -*- coding: utf-8 -*-
"""
CIP监控系统 - 配置文件监控模式（支持分段监控）

功能：
1. 监控config.json中各机组段（一段、二段）的actual_time字段变化
2. 检测到变化且actual_time < 当前时间时，基于actual_time执行CIP预测
3. 将预测结果保存到对应段的predicted_time（仅用于记录）
4. 通过回调接口发送预测结果到决策系统
5. 继续等待actual_time的下次更新

触发条件（同时满足）：
- actual_time 发生了变化
- actual_time < 当前时间（已过期）

工作流程：
修改actual_time（且 < 当前时间）→ 检测变化 → 执行预测 → 保存predicted_time → 发送回调 → 继续监控actual_time

分段监控说明：
- 支持独立监控每个段（RO1-一段、RO1-二段等）
- 每个段有独立的actual_time和predicted_time
- 可以为不同段设置不同的CIP时间
- 预测时会分别输出每个段的CIP时机
"""

import os
import json
import time
import threading
import pandas as pd
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime
from main_simple import main as run_cip_analysis
from main_simple import send_decision_to_callback

# 日志系统配置
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 禁止传播到root logger，避免重复输出
logger.propagate = False

# 日志输出格式
formatter = logging.Formatter(
    '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# 清除已有的处理器（防止重复添加）
if logger.handlers:
    for handler in logger.handlers[:]:
        handler.close()
        logger.removeHandler(handler)

# 文件日志处理器，单个文件最大5MB，保留3个备份
file_handler = RotatingFileHandler('smart_monitor.log', maxBytes=5 * 1024 * 1024, backupCount=3, encoding='utf-8')
file_handler.setFormatter(formatter)

# 控制台日志处理器
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)

# 添加处理器
logger.addHandler(file_handler)
logger.addHandler(console_handler)

class SmartCIPMonitor:
    """
    CIP监控器（支持分段监控）
    
    核心功能：
    1. 监控config.json中各机组段的actual_time字段
    2. 检测到时间变化后，自动触发CIP预测
    3. 保存predicted_time到配置文件（仅用于记录）
    4. 发送预测结果到回调接口
    
    属性：
        config_path: 配置文件路径
        config: 配置文件内容
        running: 监控运行状态
        monitor_thread: 监控线程
        unit_names: 机组段名称列表（如['RO1-一段', 'RO1-二段', ...]）
        last_seen_actual_times: 记录上次看到的各机组段actual_time，用于检测变化
    """
    
    def __init__(self):
        """
        初始化监控系统
        
        初始化流程：
        1. 设置配置文件路径
        2. 加载配置文件
        3. 初始化监控基准时间
        """
        self.config_path = os.path.join(os.path.dirname(__file__), 'config.json')
        self.config = None
        self.running = False
        self.monitor_thread = None
        # 支持分段监控：每个机组包含一段和二段
        self.unit_names = [
            'RO1-一段', 'RO1-二段',
            'RO2-一段', 'RO2-二段',
            'RO3-一段', 'RO3-二段',
            'RO4-一段', 'RO4-二段'
        ]
        
        # 记录上次看到的actual_time，用于检测变化
        # 格式：{'RO1-一段': '2025-12-01 12:00:00', 'RO1-二段': '...', ...}
        self.last_seen_actual_times = {}
        
        if not self.load_config():
            logger.error("配置文件加载失败")
        else:
            self.initialize_last_seen_times()
    
    def load_config(self):
        """
        加载配置文件
        
        Returns:
            bool: 加载成功返回True
        """
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                self.config = json.load(f)
            return True
        except Exception as e:
            logger.error(f"配置文件加载失败: {e}")
            return False
    
    def save_config(self):
        """
        保存配置文件
        
        Returns:
            bool: 保存成功返回True
        """
        try:
            with open(self.config_path, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, ensure_ascii=False, indent=2)
            return True
        except Exception as e:
            logger.error(f"配置文件保存失败: {e}")
            return False
    
    def initialize_last_seen_times(self):
        """
        初始化监控基准时间（支持分段）
        
        功能：
        读取config.json中各机组段当前的actual_time作为监控基准
        但是：只记录未来的时间，已过期的时间不记录，以便首次检查时能触发预测
        
        说明：
        - last_seen_actual_times用于防止重复处理相同的时间
        - 支持分段配置（如'RO1-一段', 'RO1-二段'）
        - 支持新旧配置格式兼容
        - 已过期的时间(< 当前时间)不记录，确保首次检查时能被检测为"变化"
        """
        if not self.config or 'cip_times' not in self.config:
            return
        
        current_time = datetime.now()
        
        for unit_name in self.unit_names:
            if unit_name in self.config['cip_times']:
                unit_data = self.config['cip_times'][unit_name]
                actual_time_str = None
                
                if isinstance(unit_data, dict) and 'actual_time' in unit_data:
                    # 新格式：{'actual_time': '...', 'predicted_time': '...'}
                    actual_time_str = unit_data['actual_time']
                elif isinstance(unit_data, str):
                    # 兼容旧格式：直接是时间字符串
                    actual_time_str = unit_data
                
                # 只记录未来的时间，已过期的时间不记录
                if actual_time_str:
                    actual_time_dt = self.parse_time_string(actual_time_str)
                    if actual_time_dt and actual_time_dt >= current_time:
                        # 未来时间：记录基准，避免触发预测
                        self.last_seen_actual_times[unit_name] = actual_time_str
                        logger.info(f"{unit_name}: 未来时间，不触发预测")
                    elif actual_time_dt:
                        # 已过期时间：不记录基准，让首次检查时能检测到变化
                        logger.info(f"{unit_name}: 已过期，待触发预测")
                    # 如果为None则不记录
    
    def parse_time_string(self, time_str):
        """
        解析时间字符串为datetime对象
        
        Args:
            time_str: str，时间字符串，格式'YYYY-MM-DD HH:MM:SS'
            
        Returns:
            datetime: 解析后的datetime对象，解析失败返回None
        """
        if not time_str or not isinstance(time_str, str):
            return None
        
        try:
            return datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S')
        except Exception as e:
            logger.warning(f"时间字符串解析失败: {time_str}, 错误: {e}")
            return None
    
    def check_for_changes(self):
        """
        检查actual_time是否有变化且小于当前时间（支持分段）
        
        功能：
        比对当前配置文件中的actual_time和上次记录的值
        发现不同且小于当前时间则判定为有变化，需要触发预测
        
        Returns:
            list: 有变化的机组段列表，格式 [(unit_name, new_actual_time), ...]
                  例如：[('RO1-一段', '2025-10-12 12:00:00'), ('RO1-二段', '2025-10-15 08:00:00')]
        
        说明：
        - 只有actual_time变化才会触发
        - 必须满足 actual_time < 当前时间
        - predicted_time的变化不会触发
        - 支持分段独立监控（每个段可以有不同的actual_time）
        """
        changes = []
        current_time = datetime.now()
        
        if not self.config or 'cip_times' not in self.config:
            return changes
        
        for unit_name in self.unit_names:
            if unit_name in self.config['cip_times']:
                unit_data = self.config['cip_times'][unit_name]
                current_actual_time = None
                
                if isinstance(unit_data, dict) and 'actual_time' in unit_data:
                    # 新格式：读取actual_time字段
                    current_actual_time = unit_data['actual_time']
                elif isinstance(unit_data, str):
                    # 兼容旧格式：整个值就是时间
                    current_actual_time = unit_data
                
                # 检查是否有变化
                if current_actual_time:
                    last_actual_time = self.last_seen_actual_times.get(unit_name)
                    
                    # 条件1：actual_time 发生了变化
                    if last_actual_time != current_actual_time:
                        # 条件2：actual_time 小于当前时间
                        actual_time_dt = self.parse_time_string(current_actual_time)
                        
                        if actual_time_dt is None:
                            logger.warning(f"{unit_name} actual_time 格式无效: {current_actual_time}")
                            continue
                        
                        if actual_time_dt < current_time:
                            changes.append((unit_name, current_actual_time))
                            logger.info(f"{unit_name}: 检测到变化，触发预测 ({current_actual_time})")
                        else:
                            logger.info(f"{unit_name}: 未来时间，不触发 ({current_actual_time})")
                            # 更新监控基准，避免重复提示
                            self.last_seen_actual_times[unit_name] = current_actual_time
        
        return changes
    
    def extract_unit_base_name(self, unit_name):
        """
        从分段机组名称中提取基础机组号
        
        Args:
            unit_name: str，分段机组名称，如'RO1-一段'
        
        Returns:
            str: 基础机组号，如'RO1'
        
        示例：
            'RO1-一段' -> 'RO1'
            'RO2-二段' -> 'RO2'
        """
        if '-' in unit_name:
            return unit_name.split('-')[0]
        return unit_name
    
    def predict_and_update_unit(self, unit_name, start_date_str):
        """
        预测并只处理触发段的结果（严格控制：谁触发只处理谁）
        
        功能：
        1. 调用预测函数（会预测整个机组的所有段，这是模型限制）
        2. 只提取触发段的结果（其他段的结果丢弃）
        3. 只保存触发段的predicted_time到配置文件
        4. 只发送触发段的回调
        5. 其他未触发的段等它们自己触发后再处理
        
        Args:
            unit_name: str，触发的机组段名称，如'RO1-一段', 'RO2-二段'
            start_date_str: str，预测起始时间（即actual_time），格式'YYYY-MM-DD HH:MM:SS'
            
        Returns:
            bool: 预测成功返回True，失败返回False
            
        说明：
        - 严格的分段控制：只处理触发段，其他段不保存、不发送
        - predicted_time仅用于记录，不影响下次预测的触发
        - 下次预测只由actual_time的变化触发
        
        示例流程：
        1. RO1-一段触发 → 预测RO1 → 只处理RO1-一段 → RO1-二段丢弃
        2. 稍后RO1-二段触发 → 预测RO1 → 只处理RO1-二段 → RO1-一段丢弃（已处理过）
        """
        try:
            logger.info(f"[{unit_name}] 开始预测 (基于 {start_date_str})")
            
            # 步骤1：提取基础机组号（用于unit_filter参数）
            base_unit_name = self.extract_unit_base_name(unit_name)
            
            # 步骤2：调用预测逻辑（会预测整个机组的所有段，这是模型限制）
            # 禁用 main() 内部的回调发送，由 smart_monitor 统一管理
            logger.info(f"[{unit_name}] 执行预测 (unit_filter={base_unit_name}，会预测整个机组)")
            result_df = run_cip_analysis(
                strategy=3, 
                start_date=start_date_str, 
                unit_filter=base_unit_name,
                separate_stages=True,  # 明确指定分段输出
                send_callback=False  # 禁用内部回调，避免重复发送
            )
            
            if result_df.empty:
                logger.warning(f"[{unit_name}] 预测失败: 无结果")
                return False
            
            # 显示预测得到的所有段
            all_predicted_stages = result_df['机组类型'].tolist()
            logger.info(f"[{unit_name}] 预测完成，得到 {len(all_predicted_stages)} 个段: {all_predicted_stages}")
            
            # 步骤3：只提取触发段的结果（关键：只处理触发的段！）
            logger.info(f"[{unit_name}] 只提取触发段 '{unit_name}' 的结果，其他段丢弃")
            unit_result = result_df[result_df['机组类型'] == unit_name]
            if unit_result.empty:
                logger.warning(f"[{unit_name}] 预测失败: 结果中无此机组段")
                logger.info(f"[{unit_name}] 可用的机组类型: {result_df['机组类型'].tolist()}")
                return False
            
            cip_time = unit_result.iloc[0]['CIP时机']
            if pd.isna(cip_time):
                logger.warning(f"[{unit_name}] 预测失败: CIP时机为空")
                return False
            
            # 步骤4：只保存触发段的预测时间到配置文件（其他段不保存）
            predicted_time_str = cip_time.strftime('%Y-%m-%d %H:%M:%S')
            logger.info(f"[{unit_name}] 只保存触发段的 predicted_time: {predicted_time_str}")
            
            if unit_name not in self.config['cip_times']:
                # 配置中不存在该机组段，创建新记录
                self.config['cip_times'][unit_name] = {
                    'actual_time': start_date_str,
                    'predicted_time': predicted_time_str
                }
            else:
                # 更新预测时间，保持actual_time不变
                if not isinstance(self.config['cip_times'][unit_name], dict):
                    # 兼容旧格式：转换为新格式
                    self.config['cip_times'][unit_name] = {
                        'actual_time': self.config['cip_times'][unit_name],
                        'predicted_time': predicted_time_str
                    }
                else:
                    # 只更新predicted_time，不修改actual_time
                    self.config['cip_times'][unit_name]['predicted_time'] = predicted_time_str
            
            if not self.save_config():
                logger.error(f"[{unit_name}] 配置保存失败")
                return False
            
            logger.info(f"[{unit_name}] 预测成功 → {predicted_time_str}")
            
            # 步骤5：只发送触发段的回调（其他段不发送）
            logger.info(f"[{unit_name}] 只发送触发段的回调")
            send_decision_to_callback(unit_result)
            
            # 明确说明哪些段被丢弃了
            other_stages = [s for s in all_predicted_stages if s != unit_name]
            if other_stages:
                logger.info(f"[{unit_name}] 其他段的结果已丢弃，等它们自己触发后再处理: {other_stages}")
            
            return True
                
        except Exception as e:
            logger.error(f"[{unit_name}] 预测失败: {e}", exc_info=True)
            return False
    
    def process_changes(self, changes):
        """
        处理检测到的actual_time变化
        
        功能：
        遍历所有检测到的变化，逐个执行预测并更新监控基准
        
        Args:
            changes: list，变化列表，格式 [(unit_name, new_actual_time), ...]
                     例如：[('RO1', '2025-10-12 12:00:00')]
        
        处理流程：
        1. 执行预测：调用predict_and_update_unit
        2. 更新监控基准：将新的actual_time记录到last_seen_actual_times（无论成功失败）
        3. 重新加载配置：确保获取最新数据
        4. 短暂延时：避免频繁操作
        """
        for unit_name, new_actual_time in changes:
            # 执行预测
            success = self.predict_and_update_unit(unit_name, new_actual_time)
            
            # 无论预测成功还是失败，都要更新监控基准，避免无限循环
            self.last_seen_actual_times[unit_name] = new_actual_time
            
            if not success:
                logger.error(f"[{unit_name}] 处理失败，已记录时间避免重复")
            
            # 重新加载配置，确保有最新数据
            self.load_config()
            time.sleep(2)  # 短暂延时，避免频繁操作
    
    def monitor_loop(self):
        """
        监控主循环
        
        功能：
        1. 定期重新加载config.json（每5秒）
        2. 检查actual_time是否有变化且小于当前时间
        3. 发现满足条件的变化后自动触发预测
        4. 发送预测结果到回调接口
        
        监控频率：
        - 每5秒检查一次配置文件
        - 发现变化立即处理
        - 处理失败不影响后续监控
        
        触发条件（同时满足）：
        - actual_time 发生了变化
        - actual_time < 当前时间（已过期）
        
        说明：
        - 只监控actual_time字段的变化
        - predicted_time的变化不会触发任何操作
        - 相同的actual_time不会重复触发
        - 未来时间的actual_time不会触发预测
        """
        logger.info("监控启动 (每5秒检查一次，触发条件: actual_time变化且<当前时间)")
        
        while self.running:
            try:
                # 步骤1：重新加载配置
                if not self.load_config():
                    logger.error("配置加载失败，10秒后重试...")
                    time.sleep(10)
                    continue
                
                # 步骤2：检查是否有变化
                changes = self.check_for_changes()
                
                if changes:
                    # 发现变化，立即处理
                    self.process_changes(changes)
                
                # 步骤3：等待一段时间后再次检查
                time.sleep(5)  # 每5秒检查一次配置文件
                    
            except Exception as e:
                logger.error(f"监控循环出错: {e}", exc_info=True)
                time.sleep(10)  # 出错后等待10秒再继续
    
    def start(self):
        """
        启动监控系统
        """
        if self.running:
            logger.warning("监控系统已在运行")
            return
        
        logger.info("CIP监控系统启动（监控 config.json 中的 actual_time 变化）")
        
        # 启动监控线程
        self.running = True
        self.monitor_thread = threading.Thread(target=self.monitor_loop, daemon=True)
        self.monitor_thread.start()

        try:
            while self.running:
                time.sleep(1)
        except KeyboardInterrupt:
            self.stop()
    
    def stop(self):
        """
        停止监控系统
        """
        logger.info("停止监控系统...")
        self.running = False
        
        if self.monitor_thread and self.monitor_thread.is_alive():
            self.monitor_thread.join(timeout=5)
        
        logger.info("监控系统已停止")

def main():
    """
    主函数
    
    功能：启动监控系统
    """
    monitor = SmartCIPMonitor()
    
    try:
        monitor.start()
    except Exception as e:
        logger.error(f"系统运行出错: {e}", exc_info=True)
    finally:
        if monitor.running:
            monitor.stop()

if __name__ == '__main__':
    main()