# -*- coding: utf-8 -*- """ CIP监控系统 - 配置文件监控模式 功能： 1. 监控config.json中actual_time字段的变化 2. 检测到变化后，基于actual_time执行CIP预测 3. 将预测结果保存到predicted_time（仅用于记录） 4. 通过回调接口发送预测结果到决策系统 工作流程： 修改actual_time → 检测变化 → 执行预测 → 保存predicted_time → 发送回调 → 继续监控 """ 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) # 日志输出格式 formatter = logging.Formatter( '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # 文件日志处理器，单个文件最大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: 机组名称列表 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', 'RO2', 'RO3', 'RO4'] # 记录上次看到的actual_time，用于检测变化 # 格式：{'RO1': '2025-12-01 12:00:00', 'RO2': '...'} 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作为监控基准 后续只有当actual_time与此基准不同时才触发预测 说明： - last_seen_actual_times用于防止重复处理相同的时间 - 支持新旧配置格式兼容 """ if not self.config or 'cip_times' not in self.config: return for unit_name in self.unit_names: if unit_name in self.config['cip_times']: unit_data = self.config['cip_times'][unit_name] if isinstance(unit_data, dict) and 'actual_time' in unit_data: # 新格式：{'actual_time': '...', 'predicted_time': '...'} self.last_seen_actual_times[unit_name] = unit_data['actual_time'] elif isinstance(unit_data, str): # 兼容旧格式：直接是时间字符串 self.last_seen_actual_times[unit_name] = unit_data def check_for_changes(self): """ 检查actual_time是否有变化 功能： 比对当前配置文件中的actual_time和上次记录的值 发现不同则判定为有变化，需要触发预测 Returns: list: 有变化的机组列表，格式 [(unit_name, new_actual_time), ...] 例如：[('RO1', '2025-10-12 12:00:00'), ('RO3', '2025-10-15 08:00:00')] 说明： - 只有actual_time变化才会触发 - predicted_time的变化不会触发 """ changes = [] 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) if last_actual_time != current_actual_time: changes.append((unit_name, current_actual_time)) logger.info(f"检测到{unit_name}实际CIP时间变化: 旧值={last_actual_time}, 新值={current_actual_time}") return changes def predict_and_update_unit(self, unit_name, start_date_str): """ 预测单个机组并保存结果到配置 功能： 1. 调用main_simple.py中的预测函数 2. 提取预测结果中的CIP时机 3. 保存predicted_time到配置文件（仅用于记录，不参与任何计算） 4. 发送预测结果到回调接口 Args: unit_name: str，机组名称，如'RO1', 'RO2', 'RO3', 'RO4' start_date_str: str，预测起始时间（即actual_time），格式'YYYY-MM-DD HH:MM:SS' Returns: bool: 预测成功返回True，失败返回False 说明： - predicted_time仅用于记录，不影响下次预测的触发 - 下次预测只由actual_time的变化触发 """ try: logger.info(f"[{unit_name}] 开始预测，基于实际CIP时间: {start_date_str}") # 步骤1：调用预测逻辑 result_df = run_cip_analysis(strategy=3, start_date=start_date_str, unit_filter=unit_name) if result_df.empty: logger.warning(f"{unit_name} 预测失败: 无结果") return False # 步骤2：提取该机组的结果 unit_result = result_df[result_df['机组类型'] == unit_name] if unit_result.empty: logger.warning(f"{unit_name} 预测失败: 结果中无此机组") return False cip_time = unit_result.iloc[0]['CIP时机'] if pd.isna(cip_time): logger.warning(f"{unit_name} 预测失败: CIP时机为空") return False # 步骤3：保存预测时间到配置文件（仅用于记录） predicted_time_str = cip_time.strftime('%Y-%m-%d %H:%M:%S') 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}") # 步骤4：发送预测结果到回调接口 try: if send_decision_to_callback(result_df): logger.info(f"[{unit_name}] 回调发送成功") else: logger.error(f"[{unit_name}] 回调发送失败") except Exception as callback_error: logger.error(f"[{unit_name}] 回调发送异常: {callback_error}") 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: # 执行预测 if self.predict_and_update_unit(unit_name, new_actual_time): # 更新监控基准，防止重复处理 self.last_seen_actual_times[unit_name] = new_actual_time logger.info(f"[{unit_name}] 处理完成") else: 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字段的变化 - predicted_time的变化不会触发任何操作 - 相同的actual_time不会重复触发 """ logger.info("开始监控 config.json 变化（每5秒检查一次）") 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()