reverse_osmosis_model/logging_system.py

# -*- coding: utf-8 -*-
"""
RO膜污染监控与CIP预测系统 - 日志记录模块

功能：
1. 记录每次分析的输入参数
2. 保存预测数据和分析结果
3. 生成可视化图表
4. 导出HTML格式分析报告
"""

import os
import json
import logging
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

class CIPAnalysisLogger:
    """
    CIP分析日志记录器
    
    功能：
    - 记录分析会话信息
    - 保存输入输出数据
    - 生成分析图表和报告
    """
    
    def __init__(self, log_dir="analysis_logs", unit_filter=None):
        """
        初始化日志记录器
        
        Args:
            log_dir: str，日志目录路径，默认"analysis_logs"
            unit_filter: str，机组过滤器，如'RO1'，用于目录命名
        
        目录结构（优化后）：
            analysis_logs/
                CIP_RO1_20251105_155542/  # 类别_机组_时间
                    CIP_Analysis_20251105_155542.log
                    data/
                    plots/
                    reports/
                CIP_ALL_20251105_160000/  # 全机组分析
        """
        # 生成会话ID和时间戳
        self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.session_id = f"CIP_Analysis_{self.timestamp}"
        
        # 创建按类别和时间分组的会话目录
        base_log_dir = Path(log_dir)
        base_log_dir.mkdir(exist_ok=True)
        
        # 目录命名：CIP_机组_时间
        unit_name = unit_filter if unit_filter else "ALL"
        dir_name = f"CIP_{unit_name}_{self.timestamp}"
        
        self.log_dir = base_log_dir / dir_name
        self.log_dir.mkdir(exist_ok=True)
        
        # 创建子目录结构（放在会话目录下）
        self.data_dir = self.log_dir / "data"
        self.plots_dir = self.log_dir / "plots" 
        self.reports_dir = self.log_dir / "reports"
        
        for dir_path in [self.data_dir, self.plots_dir, self.reports_dir]:
            dir_path.mkdir(exist_ok=True)
        
        # 初始化日志
        self.setup_logging()
        
        # 初始化数据存储
        self.analysis_data = {
            "session_info": {
                "session_id": self.session_id,
                "start_time": datetime.now().isoformat(),
                "analysis_type": "CIP时机预测",
                "system_version": "v4_wuhan"
            },
            "input_parameters": {},
            "prediction_data": {},
            "unit_analysis": {},
            "final_results": {},
            "performance_metrics": {}
        }
        
        self.logger.info(f"CIP分析会话: {self.session_id}, 目录: {self.log_dir}")
    
    def setup_logging(self):
        """设置日志配置"""
        log_file = self.log_dir / f"{self.session_id}.log"
        
        # 创建logger
        self.logger = logging.getLogger(self.session_id)
        self.logger.setLevel(logging.INFO)
        
        # 禁止日志传播到父logger，避免重复输出
        self.logger.propagate = False
        
        # 清除已有的所有处理器（防止重复添加）
        if self.logger.handlers:
            for handler in self.logger.handlers[:]:
                handler.close()
                self.logger.removeHandler(handler)
        
        # 创建文件处理器
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_handler.setLevel(logging.INFO)
        
        # 创建控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        
        # 创建格式器
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )
        
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        # 添加处理器
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)
    
    def log_input_parameters(self, strategy, start_date, prediction_start_date=None):
        """
        记录输入参数
        
        Args:
            strategy: int，策略编号
            start_date: str，起始时间
            prediction_start_date: datetime，预测起始时间
        """
        params = {
            "strategy": strategy,
            "start_date": start_date,
            "prediction_start_date": prediction_start_date.isoformat() if prediction_start_date else None,
            "analysis_timestamp": datetime.now().isoformat()
        }
        
        self.analysis_data["input_parameters"] = params
        
        pred_start = f", 预测起始: {prediction_start_date}" if prediction_start_date else ""
        self.logger.info(f"参数 - 策略: {strategy}, 起始: {start_date}{pred_start}")
    
    def log_prediction_data(self, all_data):
        """
        记录预测数据概况
        
        Args:
            all_data: pd.DataFrame，预测数据
        """
        try:
            data_info = {
                "shape": list(all_data.shape),
                "time_range": {
                    "start": all_data.index.min().isoformat(),
                    "end": all_data.index.max().isoformat()
                },
                "columns": list(all_data.columns),
                "data_points": len(all_data),
                "missing_values": all_data.isnull().sum().to_dict()
            }
            
            self.analysis_data["prediction_data"] = data_info
            
            # 保存数据到文件
            data_file = self.data_dir / f"{self.session_id}_prediction_data.csv"
            all_data.to_csv(data_file)
            self.logger.info(f"预测数据 - 形状: {data_info['shape']}, 点数: {data_info['data_points']}, 已保存: {data_file.name}")
            
        except Exception as e:
            self.logger.error(f"记录预测数据失败: {e}")
    
    def log_unit_days(self, unit_days_dict):
        """
        记录各机组预测天数
        
        Args:
            unit_days_dict: dict，机组ID到预测天数的映射
        """
        self.analysis_data["unit_days"] = unit_days_dict
        
        days_str = ", ".join([f"RO{uid}:{days}天" for uid, days in unit_days_dict.items()])
        self.logger.info(f"预测天数 - {days_str}")
    
    def log_unit_analysis_start(self, unit_id, predict_days):
        """
        记录机组分析开始
        
        Args:
            unit_id: int，机组ID
            predict_days: int，预测天数
        """
        self.logger.info(f"[RO{unit_id}] 开始分析, 预测周期: {predict_days}天")
        
        if unit_id not in self.analysis_data["unit_analysis"]:
            self.analysis_data["unit_analysis"][unit_id] = {}
        
        self.analysis_data["unit_analysis"][unit_id]["predict_days"] = predict_days
        self.analysis_data["unit_analysis"][unit_id]["analysis_start"] = datetime.now().isoformat()
    
    def log_unit_pressure_data(self, unit_id, truncated_data, pressure_columns):
        """
        记录机组压差数据
        
        Args:
            unit_id: int，机组ID
            truncated_data: pd.DataFrame，截取后的数据
            pressure_columns: list，压差列名列表
        """
        try:
            unit_data = {
                "pressure_columns": pressure_columns,
                "data_shape": list(truncated_data.shape),
                "data_range": {
                    "start": truncated_data.index.min().isoformat(),
                    "end": truncated_data.index.max().isoformat()
                }
            }
            
            self.analysis_data["unit_analysis"][unit_id]["pressure_data"] = unit_data
            
            cols_str = ", ".join(pressure_columns)
            self.logger.info(f"[RO{unit_id}] 压差列: {len(pressure_columns)}个 ({cols_str})")
            
            # 保存数据文件
            unit_data_file = self.data_dir / f"{self.session_id}_RO{unit_id}_pressure_data.csv"
            truncated_data[pressure_columns].to_csv(unit_data_file)
            
        except Exception as e:
            self.logger.error(f"记录RO{unit_id}压差数据失败: {e}")
    
    def log_cip_analysis_result(self, unit_id, column, optimal_time, analysis):
        """
        记录CIP分析结果（增强版，包含诊断信息）
        
        Args:
            unit_id: int，机组ID
            column: str，压差列名
            optimal_time: pd.Timestamp，最优CIP时间
            analysis: dict，分析结果（可能包含诊断信息）
        """
        try:
            if "cip_results" not in self.analysis_data["unit_analysis"][unit_id]:
                self.analysis_data["unit_analysis"][unit_id]["cip_results"] = []
            
            result = {
                "column": column,
                "optimal_time": optimal_time.isoformat() if optimal_time else None,
                "delay_days": analysis.get("delay_days"),
                "k_value": analysis.get("best_k"),
                "analysis_details": analysis
            }
            
            self.analysis_data["unit_analysis"][unit_id]["cip_results"].append(result)
            
            if optimal_time:
                self.logger.info(f"  {column}: {optimal_time} (第{analysis['delay_days']}天, k={analysis['best_k']:.6f})")
            else:
                error_msg = analysis.get('error', '未知原因')
                self.logger.info(f"  {column}: 未找到CIP时机 - {error_msg}")
                
                # 记录详细诊断信息
                if 'hint' in analysis:
                    self.logger.info(f"    提示: {analysis['hint']}")
                if 'valid_k_count' in analysis:
                    self.logger.info(f"    有效k值数量: {analysis['valid_k_count']}")
                if 'rising_periods_count' in analysis:
                    self.logger.info(f"    上升趋势段数: {analysis['rising_periods_count']}")
                if 'data_days' in analysis:
                    self.logger.info(f"    数据覆盖天数: {analysis['data_days']}天")
                
        except Exception as e:
            self.logger.error(f"记录CIP分析结果失败: {e}")
    
    def log_unit_strategy_result(self, unit_id, optimal_time, strategy_desc):
        """
        记录机组策略选择结果
        
        Args:
            unit_id: int，机组ID
            optimal_time: pd.Timestamp，最优CIP时间
            strategy_desc: str，策略描述
        """
        try:
            strategy_result = {
                "optimal_time": optimal_time.isoformat() if optimal_time else None,
                "strategy_description": strategy_desc,
                "selection_timestamp": datetime.now().isoformat()
            }
            
            self.analysis_data["unit_analysis"][unit_id]["strategy_result"] = strategy_result
            
            self.logger.info(f"[RO{unit_id}] 最优CIP时机: {optimal_time}, 策略: {strategy_desc}")
            
        except Exception as e:
            self.logger.error(f"记录RO{unit_id}策略结果失败: {e}")
    
    def log_final_results(self, result_df):
        """
        记录最终分析结果
        
        Args:
            result_df: pd.DataFrame，最终结果表
        """
        try:
            # 转换为可序列化格式
            final_results = []
            for _, row in result_df.iterrows():
                result = {
                    "unit": row["机组类型"],
                    "cip_time": row["CIP时机"].isoformat() if pd.notna(row["CIP时机"]) else None,
                    "strategy_description": row["策略说明"]
                }
                final_results.append(result)
            
            self.analysis_data["final_results"] = final_results
            
            # 保存结果文件
            result_file = self.data_dir / f"{self.session_id}_final_results.csv"
            result_df.to_csv(result_file, index=False, encoding='utf-8')
            
            results_summary = ", ".join([f"{r['unit']}: {r['cip_time'] or 'N/A'}" for r in final_results])
            self.logger.info(f"最终结果 - {results_summary}, 已保存: {result_file.name}")
            
        except Exception as e:
            self.logger.error(f"记录最终结果失败: {e}")
    
    def create_analysis_plots(self, all_data, unit_days_dict):
        """
        创建分析图表
        
        功能：生成压差趋势图和机组对比图
        
        Args:
            all_data: pd.DataFrame，完整预测数据
            unit_days_dict: dict，各机组预测天数
        """
        try:
            # 创建压差趋势总览图
            fig, axes = plt.subplots(2, 2, figsize=(20, 12))
            fig.suptitle(f'RO膜污染分析总览 - {self.session_id}', fontsize=16, fontweight='bold')
            
            # 为每个机组绘制压差趋势
            for idx, unit_id in enumerate([1, 2, 3, 4]):
                ax = axes[idx//2, idx%2]
                
                # 筛选该机组的压差列
                unit_columns = [col for col in all_data.columns if f'RO{unit_id}' in col and 'DPT' in col]
                
                if unit_columns:
                    # 截取预测天数范围内的数据
                    predict_days = unit_days_dict.get(unit_id, 90)
                    end_time = all_data.index[0] + timedelta(days=predict_days)
                    truncated_data = all_data.loc[all_data.index <= end_time]
                    
                    # 绘制各段压差曲线
                    for col in unit_columns:
                        if col in truncated_data.columns:
                            stage = "一段" if "DPT_1" in col else "二段"
                            ax.plot(truncated_data.index, truncated_data[col], 
                                   label=f'{stage}压差', linewidth=2, alpha=0.8)
                    
                    ax.set_title(f'RO{unit_id} 压差趋势 (预测{predict_days}天)', fontweight='bold')
                    ax.set_xlabel('时间')
                    ax.set_ylabel('压差 (MPa)')
                    ax.legend()
                    ax.grid(True, alpha=0.3)
                    ax.tick_params(axis='x', rotation=45)
                else:
                    # 无数据时显示提示
                    ax.text(0.5, 0.5, f'RO{unit_id}\n无压差数据', 
                           ha='center', va='center', transform=ax.transAxes,
                           fontsize=14, bbox=dict(boxstyle="round,pad=0.3", facecolor="lightgray"))
                    ax.set_title(f'RO{unit_id} - 无数据')
            
            plt.tight_layout()
            
            # 保存图表文件
            plot_file = self.plots_dir / f"{self.session_id}_pressure_trends.png"
            plt.savefig(plot_file, dpi=300, bbox_inches='tight')
            plt.close()
            
            # 创建机组对比图
            comparison_plot = self._create_unit_comparison_plot(unit_days_dict)
            
            self.logger.info(f"分析图表 - 已保存: {plot_file.name}, {comparison_plot}")
            
        except Exception as e:
            self.logger.error(f"创建分析图表失败: {e}")
    
    def _create_unit_comparison_plot(self, unit_days_dict):
        """创建机组对比图"""
        try:
            fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 6))
            
            # 预测天数对比
            units = list(unit_days_dict.keys())
            days = list(unit_days_dict.values())
            
            bars = ax1.bar([f'RO{u}' for u in units], days, 
                          color=['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728'])
            ax1.set_title('各机组预测周期对比', fontweight='bold')
            ax1.set_ylabel('预测天数')
            ax1.grid(True, alpha=0.3)
            
            # 在柱子上显示数值
            for bar, day in zip(bars, days):
                height = bar.get_height()
                ax1.text(bar.get_x() + bar.get_width()/2., height + 1,
                        f'{day}天', ha='center', va='bottom', fontweight='bold')
            
            # CIP时机分布（如果有结果的话）
            if self.analysis_data.get("final_results"):
                cip_times = []
                units_with_cip = []
                
                for result in self.analysis_data["final_results"]:
                    if result["cip_time"]:
                        cip_times.append(pd.to_datetime(result["cip_time"]))
                        units_with_cip.append(result["unit"])
                
                if cip_times:
                    # 计算从预测开始的天数
                    start_time = min(cip_times)
                    days_from_start = [(t - start_time).days for t in cip_times]
                    
                    bars2 = ax2.bar(units_with_cip, days_from_start, 
                                   color=['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728'][:len(units_with_cip)])
                    ax2.set_title('各机组CIP建议时机', fontweight='bold')
                    ax2.set_ylabel('距离预测开始天数')
                    ax2.grid(True, alpha=0.3)
                    
                    # 在柱子上显示数值
                    for bar, day in zip(bars2, days_from_start):
                        height = bar.get_height()
                        ax2.text(bar.get_x() + bar.get_width()/2., height + 0.5,
                                f'{day}天', ha='center', va='bottom', fontweight='bold')
                else:
                    ax2.text(0.5, 0.5, '无有效CIP时机', 
                            ha='center', va='center', transform=ax2.transAxes,
                            fontsize=14, bbox=dict(boxstyle="round,pad=0.3", facecolor="lightcoral"))
            else:
                ax2.text(0.5, 0.5, '分析未完成', 
                        ha='center', va='center', transform=ax2.transAxes,
                        fontsize=14, bbox=dict(boxstyle="round,pad=0.3", facecolor="lightgray"))
            
            plt.tight_layout()
            
            # 保存图表文件
            comparison_file = self.plots_dir / f"{self.session_id}_unit_comparison.png"
            plt.savefig(comparison_file, dpi=300, bbox_inches='tight')
            plt.close()
            
            return comparison_file.name
            
        except Exception as e:
            self.logger.error(f"创建机组对比图失败: {e}")
            return None
    
    def generate_analysis_report(self):
        """
        生成分析报告
        
        功能：
        1. 保存完整分析数据（JSON格式）
        2. 生成HTML格式报告
        3. 输出会话总结
        """
        try:
            # 记录结束时间
            self.analysis_data["session_info"]["end_time"] = datetime.now().isoformat()
            
            # 保存JSON数据
            json_file = self.reports_dir / f"{self.session_id}_analysis_data.json"
            with open(json_file, 'w', encoding='utf-8') as f:
                json.dump(self.analysis_data, f, ensure_ascii=False, indent=2)
            
            # 生成HTML报告
            html_report = self._generate_html_report()
            html_file = self.reports_dir / f"{self.session_id}_report.html"
            with open(html_file, 'w', encoding='utf-8') as f:
                f.write(html_report)
            
            self.logger.info(f"分析报告 - 已保存: {json_file.name}, {html_file.name}")
            
            # 生成会话总结
            self._log_session_summary()
            
        except Exception as e:
            self.logger.error(f"生成分析报告失败: {e}")
    
    def _generate_html_report(self):
        """生成HTML格式的分析报告"""
        session_info = self.analysis_data["session_info"]
        input_params = self.analysis_data["input_parameters"]
        final_results = self.analysis_data.get("final_results", [])
        
        html = f"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>CIP分析报告 - {session_info['session_id']}</title>
    <style>
        body {{ font-family: Arial, sans-serif; margin: 20px; background-color: #f5f5f5; }}
        .container {{ max-width: 1200px; margin: 0 auto; background-color: white; padding: 20px; border-radius: 10px; box-shadow: 0 0 10px rgba(0,0,0,0.1); }}
        h1 {{ color: #2c3e50; text-align: center; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
        h2 {{ color: #34495e; border-left: 4px solid #3498db; padding-left: 10px; }}
        .info-grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(300px, 1fr)); gap: 20px; margin: 20px 0; }}
        .info-card {{ background-color: #ecf0f1; padding: 15px; border-radius: 8px; border-left: 4px solid #3498db; }}
        .result-table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
        .result-table th, .result-table td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
        .result-table th {{ background-color: #3498db; color: white; }}
        .result-table tr:nth-child(even) {{ background-color: #f2f2f2; }}
        .status-success {{ color: #27ae60; font-weight: bold; }}
        .status-warning {{ color: #f39c12; font-weight: bold; }}
        .status-error {{ color: #e74c3c; font-weight: bold; }}
        .timestamp {{ color: #7f8c8d; font-size: 0.9em; }}
    </style>
</head>
<body>
    <div class="container">
        <h1>RO膜污染监控与CIP预测分析报告</h1>
        
        <div class="info-grid">
            <div class="info-card">
                <h3>📋 会话信息</h3>
                <p><strong>会话ID:</strong> {session_info['session_id']}</p>
                <p><strong>开始时间:</strong> <span class="timestamp">{session_info['start_time']}</span></p>
                <p><strong>结束时间:</strong> <span class="timestamp">{session_info.get('end_time', '进行中')}</span></p>
                <p><strong>系统版本:</strong> {session_info['system_version']}</p>
            </div>
            
            <div class="info-card">
                <h3>⚙️ 输入参数</h3>
                <p><strong>策略:</strong> {input_params.get('strategy', 'N/A')}</p>
                <p><strong>起始时间:</strong> {input_params.get('start_date', 'N/A')}</p>
                <p><strong>预测起始:</strong> {input_params.get('prediction_start_date', 'N/A')}</p>
            </div>
        </div>
        
        <h2>🎯 分析结果</h2>
        <table class="result-table">
            <thead>
                <tr>
                    <th>机组</th>
                    <th>CIP建议时机</th>
                    <th>策略说明</th>
                    <th>状态</th>
                </tr>
            </thead>
            <tbody>
        """
        
        for result in final_results:
            status_class = "status-success" if result["cip_time"] else "status-warning"
            status_text = "有建议" if result["cip_time"] else "无建议"
            cip_time_display = result["cip_time"] or "N/A"
            
            html += f"""
                <tr>
                    <td><strong>{result['unit']}</strong></td>
                    <td>{cip_time_display}</td>
                    <td>{result['strategy_description']}</td>
                    <td class="{status_class}">{status_text}</td>
                </tr>
            """
        
        html += """
            </tbody>
        </table>
        
        <h2>📊 数据文件</h2>
        <div class="info-card">
            <p>本次分析生成的所有数据文件和图表已保存在以下目录:</p>
            <ul>
                <li><strong>数据文件:</strong> analysis_logs/data/</li>
                <li><strong>图表文件:</strong> analysis_logs/plots/</li>
                <li><strong>报告文件:</strong> analysis_logs/reports/</li>
            </ul>
        </div>
        
        <h2>ℹ️ 说明</h2>
        <div class="info-card">
            <p>本报告基于RO膜污染监控与CIP预测系统生成，包含了完整的分析过程记录。工艺人员可以根据以下内容进行合理性排查:</p>
            <ul>
                <li>查看各机组的压差趋势图，判断污染发展情况</li>
                <li>对比预测周期与历史经验，评估合理性</li>
                <li>分析k值变化趋势，了解膜污染速率</li>
                <li>检查CIP建议时机是否符合实际运行需求</li>
            </ul>
        </div>
        
        <div class="timestamp" style="text-align: center; margin-top: 30px; border-top: 1px solid #ddd; padding-top: 15px;">
            报告生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
        </div>
    </div>
</body>
</html>
        """
        
        return html
    
    def _log_session_summary(self):
        """
        记录会话总结
        
        功能：输出分析耗时、成功率、文件统计等信息
        """
        session_info = self.analysis_data["session_info"]
        start_time = datetime.fromisoformat(session_info["start_time"])
        end_time = datetime.fromisoformat(session_info["end_time"])
        duration = end_time - start_time
        
        # 统计分析结果
        final_results = self.analysis_data.get("final_results", [])
        success_count = sum(1 for r in final_results if r["cip_time"])
        total_count = len(final_results)
        success_rate = f"{success_count/total_count*100:.1f}%" if total_count > 0 else "N/A"
        
        # 统计生成的文件
        data_files = len(list(self.data_dir.glob(f"{self.session_id}_*.csv")))
        plot_files = len(list(self.plots_dir.glob(f"{self.session_id}_*.png")))
        
        self.logger.info(f"会话结束 - ID: {session_info['session_id']}, 耗时: {duration.total_seconds():.2f}秒, 机组: {total_count}个, 成功: {success_count}个({success_rate}), 文件: {data_files}数据+{plot_files}图表")
    
    def close(self):
        """关闭日志记录器"""
        self.generate_analysis_report()
        
        # 关闭所有处理器
        for handler in self.logger.handlers[:]:
            handler.close()
            self.logger.removeHandler(handler)