reverse_osmosis_model/main_simple.py

# -*- coding: utf-8 -*-
"""
RO膜污染监控与CIP预测 - 基于预测数据的最优时机分析

核心功能：分析RO膜压差预测数据，计算最优CIP清洗时机

CIP时机选择策略:
1. 最早时机策略：一段或二段任一需要CIP时即触发
2. 最晚时机策略：等待所有段都需要CIP时触发
3. 加权平均策略：综合两段污染程度，污染严重段权重更大
4. 污染严重程度策略：基于k值最大的段决策

使用方法:
    main(strategy=3)  # 使用策略3
    main(strategy=1, start_date='2025-08-26 00:00:00')  # 指定策略和时间
"""

import pandas as pd
import numpy as np
import logging
from logging.handlers import RotatingFileHandler
from sklearn.linear_model import LinearRegression
from fouling_model_0922.predict import Predictor
import warnings
from datetime import datetime, timedelta
from logging_system import CIPAnalysisLogger
import json
import requests
import time
import os

warnings.filterwarnings('ignore', category=FutureWarning)

# 日志系统配置
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('main_simple.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)

# 加载配置文件
def load_config():
    """
    加载配置文件
    
    Returns:
        dict: 配置字典，失败时返回None
    """
    config_path = os.path.join(os.path.dirname(__file__), 'config.json')
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        return config
    except Exception as e:
        logger.error(f"配置文件加载失败: {e}")
        return None

# 加载配置
config = load_config()

def update_cip_history_in_config(result_df):
    """
    保存CIP预测结果到配置文件
    
    功能：将预测的CIP时机写入config.json的cip_times字段的predicted_time
    
    Args:
        result_df: DataFrame，包含机组类型和CIP时机两列
        
    Returns:
        bool: 保存成功返回True，失败返回False
        
    注意：
        此函数已废弃，smart_monitor会自动保存predicted_time
        保留此函数仅为兼容性
    """
    global config
    
    if config is None:
        print("配置文件未加载")
        return False
    
    try:
        config_path = os.path.join(os.path.dirname(__file__), 'config.json')
        
        with open(config_path, 'r', encoding='utf-8') as f:
            current_config = json.load(f)
        
        # 遍历结果，写入配置
        updated_units = []
        for _, row in result_df.iterrows():
            if pd.notna(row["CIP时机"]):
                unit_name = row["机组类型"]
                cip_time = row["CIP时机"].strftime('%Y-%m-%d %H:%M:%S')
                
                if unit_name in current_config.get('cip_times', {}):
                    # 新格式：只更新predicted_time
                    if isinstance(current_config['cip_times'][unit_name], dict):
                        current_config['cip_times'][unit_name]['predicted_time'] = cip_time
                    else:
                        # 兼容旧格式：转换为新格式
                        current_config['cip_times'][unit_name] = {
                            'actual_time': current_config['cip_times'][unit_name],
                            'predicted_time': cip_time
                        }
                    updated_units.append(f"{unit_name}: {cip_time}")
        
        if updated_units:
            with open(config_path, 'w', encoding='utf-8') as f:
                json.dump(current_config, f, ensure_ascii=False, indent=2)
            
            config = current_config
            return True
        else:
            return False
            
    except Exception as e:
        logger.error(f"保存CIP预测时间失败: {e}")
        return False

def validate_data(data, name="数据"):
    """
    验证时间序列数据格式
    
    检查项：
    1. 数据非空
    2. 索引类型为DatetimeIndex
    
    Args:
        data: pd.Series或pd.DataFrame
        name: 数据名称，用于错误提示
        
    Returns:
        bool: 验证通过返回True
        
    Raises:
        ValueError: 验证失败时抛出
    """
    if data is None or data.empty:
        raise ValueError(f"{name}为空或无效")
    if not isinstance(data.index, pd.DatetimeIndex):
        raise ValueError(f"{name}的索引必须是时间格式")
    return True


class OptimalCIPPredictor:
    """
    CIP最优时机预测器
    
    工作原理：
    1. 使用滑动窗口计算k值（膜污染速率）
    2. 识别k值连续上升趋势
    3. 在满足时间约束前提下，选择k值最大的时间点
    
    参数说明：
    - window_days: 滑动窗口大小（天），用于线性回归计算k值
    - min_continuous_rising: 最小连续上升点数，确保趋势稳定
    - min_delay_days: 最小延迟天数，避免过早建议CIP
    """
    
    def __init__(self, window_days=7, min_continuous_rising=3, min_delay_days=30):
        """
        初始化预测器
        
        Args:
            window_days: 滑动窗口天数（默认7天）
            min_continuous_rising: 最小连续上升点数（默认3点）
            min_delay_days: 最小延迟天数（默认30天）
        """
        self.window_days = window_days
        self.window_hours = window_days * 24  # 转换为小时
        self.min_continuous_rising = min_continuous_rising
        self.min_delay_days = min_delay_days
    
    def calculate_sliding_k_values(self, pressure_series):
        """
        计算滑动窗口k值序列
        
        基于机理模型: ΔP(t) = ΔP₀ + k×t
        通过线性回归计算斜率k，表示膜污染速率
        
        Args:
            pressure_series: pd.Series，压差时间序列，索引为时间
            
        Returns:
            pd.Series: k值序列，前window_hours个值为NaN
        """
        # 初始化k值序列，索引与输入数据保持一致
        k_values = pd.Series(index=pressure_series.index, dtype=float)
        
        # 滑动窗口遍历，从第window_hours个点开始
        for i in range(self.window_hours, len(pressure_series)):
            # 取当前窗口内的数据
            window_data = pressure_series.iloc[i-self.window_hours:i]
            
            # 数据质量检查：窗口内至少80%的数据有效
            if len(window_data) < self.window_hours * 0.8:
                continue
                
            # 构造时间点序列 [0, 1, 2, ..., window_hours-1]
            time_points = np.arange(len(window_data)).reshape(-1, 1)
            
            try:
                # 线性回归拟合：y = a + k*x，取斜率k作为污染速率
                model = LinearRegression()
                model.fit(time_points, window_data.values)
                k = model.coef_[0]
                k_values.iloc[i] = k
            except:
                # 回归失败时跳过该点
                continue
        
        return k_values
    
    def find_continuous_rising_periods(self, k_values):
        """
        识别k值连续上升的时间段
        
        遍历k值序列，找出所有连续上升的区间
        只保留持续时间大于等于min_continuous_rising的区间
        
        Args:
            k_values: pd.Series，k值序列
            
        Returns:
            list: 连续上升时间段列表，格式 [(start_idx, end_idx, duration), ...]
        """
        rising_periods = []
        start_idx = None  # 当前上升段的起始索引
        
        # 遍历k值序列，寻找连续上升段
        for i in range(1, len(k_values)):
            # 跳过缺失值
            if pd.isna(k_values.iloc[i]) or pd.isna(k_values.iloc[i-1]):
                start_idx = None
                continue
                
            # 判断k值是否上升
            if k_values.iloc[i] > k_values.iloc[i-1]:
                # 开始新的上升段
                if start_idx is None:
                    start_idx = i-1
            else:
                # k值不再上升，结束当前上升段
                if start_idx is not None:
                    duration = i - start_idx
                    # 只保留持续时间足够长的上升段
                    if duration >= self.min_continuous_rising:
                        rising_periods.append((start_idx, i-1, duration))
                    start_idx = None
        
        # 处理序列末尾的上升趋势
        if start_idx is not None:
            duration = len(k_values) - start_idx # 持续时间 
            if duration >= self.min_continuous_rising: # 持续时间足够长  
                rising_periods.append((start_idx, len(k_values)-1, duration)) # 添加上升时间段
        
        return rising_periods
    
    def find_optimal_cip_time(self, pressure_series):
        """
        最优CIP时机
        
        核心步骤：
        1. 计算滑动窗口k值（膜污染速率）
        2. 识别k值连续上升的时间段
        3. 应用时间约束（距离起点至少min_delay_days天）
        4. 在有效时间段内选择k值最大的时间点
        
        Args:
            pressure_series: pd.Series，压差时间序列
            
        Returns:
            tuple: (optimal_time, analysis_result)
                - optimal_time: pd.Timestamp，最优CIP时间，失败时返回None
                - analysis_result: dict，分析结果详情
        """
        # 步骤1：计算滑动k值
        k_values = self.calculate_sliding_k_values(pressure_series)
        valid_k_count = k_values.dropna().shape[0]
        
        # 检查：k值数量是否足够
        if valid_k_count < 10:
            return None, {"error": "有效k值数量不足"}
        
        # 步骤2：识别连续上升时间段
        rising_periods = self.find_continuous_rising_periods(k_values)
        
        if not rising_periods:
            return None, {"error": "未发现连续上升趋势"}
        
        # 步骤3：应用时间约束，筛选有效时间段
        min_delay_time = pressure_series.index[0] + timedelta(days=self.min_delay_days)
        valid_periods = []
        
        for start_idx, end_idx, duration in rising_periods:
            period_start_time = pressure_series.index[start_idx]
            period_end_time = pressure_series.index[end_idx]
            
            # 检查时间段是否在约束范围内
            if period_end_time >= min_delay_time:
                if period_start_time < min_delay_time:
                    # 时间段部分在约束范围内，截取有效部分
                    delay_idx = pressure_series.index.get_indexer([min_delay_time], method='nearest')[0]
                    if delay_idx <= end_idx:
                        valid_periods.append((delay_idx, end_idx, end_idx - delay_idx + 1))
                else:
                    # 时间段完全在约束范围内
                    valid_periods.append((start_idx, end_idx, duration))
        
        if not valid_periods:
            return None, {"error": f"无满足时间约束的上升趋势(需>={self.min_delay_days}天后)"}
        
        # 步骤4：在有效时间段内寻找k值最大的点
        best_time = None
        best_k = -np.inf
        
        for start_idx, end_idx, duration in valid_periods:
            period_k_values = k_values.iloc[start_idx:end_idx+1]
            max_k_idx = period_k_values.idxmax()  # k值最大点的索引
            max_k_value = period_k_values.max()   # k值最大值
            
            if max_k_value > best_k:
                best_k = max_k_value
                best_time = max_k_idx
        
        # 构建分析结果
        analysis_result = {
            "success": True,
            "delay_days": (best_time - pressure_series.index[0]).days,
            "best_k": float(best_k)
        }
        
        return best_time, analysis_result

def select_optimal_cip_strategy_1(cip_results):
    """
    策略1：最早时机策略
    
    选择逻辑：取一段和二段中较早需要CIP的时机
    适用场景：保守运维，及时维护
    """
    if not cip_results:
        return None, "无有效CIP时机"
    
    earliest_result = min(cip_results, key=lambda x: x['delay_days'])
    return earliest_result['time'], f"最早时机策略 - {earliest_result['column']} (第{earliest_result['delay_days']}天)"

def select_optimal_cip_strategy_2(cip_results):
    """
    策略2：最晚时机策略
    
    选择逻辑：取一段和二段中较晚需要CIP的时机
    适用场景：最大化运行时间
    """
    if not cip_results:
        return None, "无有效CIP时机"
    
    latest_result = max(cip_results, key=lambda x: x['delay_days'])
    return latest_result['time'], f"最晚时机策略 - {latest_result['column']} (第{latest_result['delay_days']}天)"

def select_optimal_cip_strategy_3(cip_results):
    """
    策略3：加权平均策略（推荐）
    
    选择逻辑：根据k值对各段CIP时机加权，污染严重段权重更大
    适用场景：平衡运行时间和维护需求
    """
    if not cip_results:
        return None, "无有效CIP时机"
    
    if len(cip_results) == 1:
        result = cip_results[0]
        return result['time'], f"单段加权策略 - {result['column']} (第{result['delay_days']}天)"
    
    # 计算加权平均天数
    total_weight = sum(result['k_value'] for result in cip_results)
    weighted_days = sum(result['delay_days'] * result['k_value'] for result in cip_results) / total_weight
    
    # 找最接近加权平均天数的时机
    target_days = int(round(weighted_days))
    closest_result = min(cip_results, key=lambda x: abs(x['delay_days'] - target_days))
    
    return closest_result['time'], f"加权平均策略 - {closest_result['column']} (目标第{target_days}天，实际第{closest_result['delay_days']}天)"

def select_optimal_cip_strategy_4(cip_results):
    """
    策略4：污染严重程度策略
    
    选择逻辑：选择k值最大（污染最严重）的段的CIP时机
    适用场景：基于实际污染状况决策
    """
    if not cip_results:
        return None, "无有效CIP时机"
    
    max_k_result = max(cip_results, key=lambda x: x['k_value'])
    return max_k_result['time'], f"污染严重程度策略 - {max_k_result['column']} (k值={max_k_result['k_value']:.6f}, 第{max_k_result['delay_days']}天)"

def select_optimal_cip_time(cip_results, strategy=1):
    """
    根据指定策略选择最优CIP时机
    
    Args:
        cip_results: list，各段CIP分析结果列表，每个元素包含time、delay_days、k_value等字段
        strategy: int，策略编号，1-4分别对应不同策略
    
    Returns:
        tuple: (optimal_time, description)
            - optimal_time: pd.Timestamp，最优CIP时间
            - description: str，策略描述
    
    Raises:
        ValueError: 当策略编号无效时抛出
    """
    strategy_map = {
        1: select_optimal_cip_strategy_1,  # 最早时机
        2: select_optimal_cip_strategy_2,  # 最晚时机
        3: select_optimal_cip_strategy_3,  # 加权平均（推荐）
        4: select_optimal_cip_strategy_4   # 污染严重程度
    }
    
    if strategy not in strategy_map:
        raise ValueError(f"无效策略编号: {strategy}，支持的策略: 1-4")
    
    return strategy_map[strategy](cip_results)

def analyze_ro_unit_cip_timing(strategy=1, start_date=None, unit_filter=None):
    """
    分析RO机组的最优CIP时间
    
    功能：
    1. 获取压差预测数据
    2. 分析各机组各段的CIP时机
    3. 根据策略选择最优CIP时间
    
    Args:
        strategy: int，CIP时机选择策略（1-4）
            1: 最早时机策略
            2: 最晚时机策略  
            3: 加权平均策略（推荐）
            4: 污染严重程度策略
        start_date: str，预测起始时间，格式'YYYY-MM-DD HH:MM:SS'，默认使用当前时间
        unit_filter: str，指定分析的机组，如'RO1'，默认分析所有机组
    
    Returns:
        pd.DataFrame: 包含机组类型、CIP时机、策略说明的结果表
    """
    # 初始化日志记录器
    analysis_logger = CIPAnalysisLogger()
    
    try:
        # 获取预测数据
        try:
            all_data = Predictor().predict(start_date=start_date)  # 获取预测数据
            if all_data.empty: 
                analysis_logger.logger.error("预测数据为空")
                return pd.DataFrame()
        except Exception as e:
            analysis_logger.logger.error(f"获取预测数据失败: {e}")
            return pd.DataFrame()
        
        # 将date列设置为索引
        all_data = all_data.set_index('date')
        
        # 获取预测数据的起始时间
        prediction_start_date = all_data.index[0].to_pydatetime()
        logger.info(f"预测起始: {prediction_start_date.strftime('%Y-%m-%d %H:%M:%S')}")

        # 记录输入参数和预测数据
        analysis_logger.log_input_parameters(strategy, start_date, prediction_start_date)
        analysis_logger.log_prediction_data(all_data)
        
        # 确定要分析的机组
        if unit_filter:
            unit_ids = [int(unit_filter.replace('RO', ''))]
        else:
            unit_ids = [1, 2, 3, 4]
        
        # 获取各机组的预测天数
        from cip.run_this import main as get_unit_days
        unit_days_dict = {}
        for unit_id in unit_ids:
            unit_days_dict[unit_id] = get_unit_days(unit_id, prediction_start_date)
        
        # 记录机组预测天数
        analysis_logger.log_unit_days(unit_days_dict)
        
        # 初始化预测器
        predictor = OptimalCIPPredictor(window_days=7, min_continuous_rising=3, min_delay_days=30)
        
        # 存储分析结果
        results = []
        
        # 遍历分析各机组
        for unit_id in unit_ids:
            # 获取该机组的预测天数
            predict_days = unit_days_dict[unit_id]
            
            # 记录分析开始
            analysis_logger.log_unit_analysis_start(unit_id, predict_days)
            
            # 截取预测天数范围内的数据
            end_time = all_data.index[0] + timedelta(days=predict_days)
            truncated_data = all_data.loc[all_data.index <= end_time]
            
            # 筛选该机组的压差列
            ro_name = f"RO{unit_id}"
            pressure_columns = [col for col in truncated_data.columns if ro_name in col and 'DPT' in col and 'pred' in col]
            
            if not pressure_columns:
                logger.warning(f"[{ro_name}] 未找到压差列")
                continue
            
            # 记录压差数据
            analysis_logger.log_unit_pressure_data(unit_id, truncated_data, pressure_columns)
            
            # 收集各段的CIP分析结果
            cip_results = []
            
            for column in pressure_columns:
                pressure_series = truncated_data[column].dropna()
                pressure_series.name = column
                
                # 数据点数检查：至少需要30天数据
                if len(pressure_series) < 30 * 24:
                    continue
                
                try:
                    # 寻找最优CIP时机
                    optimal_time, analysis = predictor.find_optimal_cip_time(pressure_series)
                    
                    # 记录分析结果
                    analysis_logger.log_cip_analysis_result(unit_id, column, optimal_time, analysis)
                    
                    if optimal_time:
                        cip_results.append({
                            'column': column,
                            'time': optimal_time,
                            'delay_days': analysis['delay_days'],
                            'k_value': analysis['best_k']
                        })
                        
                except Exception as e:
                    analysis_logger.log_cip_analysis_result(unit_id, column, None, {"error": str(e)})
            
            # 根据策略选择最优CIP时机
            if cip_results:
                optimal_time, strategy_desc = select_optimal_cip_time(cip_results, strategy)
                results.append({
                    '机组类型': f"RO{unit_id}",
                    'CIP时机': optimal_time,
                    '策略说明': strategy_desc
                })
                logger.info(f"[RO{unit_id}] CIP时机 → {optimal_time.strftime('%Y-%m-%d %H:%M:%S')}")
                
                analysis_logger.log_unit_strategy_result(unit_id, optimal_time, strategy_desc)
            else:
                results.append({
                    '机组类型': f"RO{unit_id}",
                    'CIP时机': None,
                    '策略说明': "无有效CIP时机"
                })
                logger.warning(f"[RO{unit_id}] 未找到有效CIP时机")
                
                analysis_logger.log_unit_strategy_result(unit_id, None, "无有效CIP时机")
    
        # 生成结果DataFrame
        result_df = pd.DataFrame(results)
        
        # 记录最终结果
        analysis_logger.log_final_results(result_df)
        
        # 生成分析图表
        analysis_logger.create_analysis_plots(all_data, unit_days_dict)
        
        return result_df
        
    except Exception as e:
        analysis_logger.logger.error(f"分析过程中发生错误: {e}")
        raise
    finally:
        # 确保日志记录器正确关闭
        analysis_logger.close()

def main(strategy=3, start_date=None, unit_filter=None):
    """
    主执行函数
    
    功能：执行RO机组CIP时机分析并发送结果到回调接口
    
    Args:
        strategy: int，CIP时机选择策略（1-4），默认3（加权平均策略）
            1: 最早时机策略
            2: 最晚时机策略  
            3: 加权平均策略
            4: 污染严重程度策略
        start_date: str，预测起始时间，格式'YYYY-MM-DD HH:MM:SS'，默认None（使用当前时间）
        unit_filter: str，指定预测的机组，如'RO1'，默认None（预测所有机组）
    
    Returns:
        pd.DataFrame: 分析结果
    
    示例：
        result_df = main()  # 使用默认参数
        result_df = main(start_date='2025-07-01 00:00:00')  # 指定时间
        result_df = main(strategy=1, unit_filter='RO1')  # 指定策略和机组
    """
    # 执行分析 
    result_df = analyze_ro_unit_cip_timing(strategy=strategy, start_date=start_date, unit_filter=unit_filter)
    
    # 发送回调
    if config and not result_df.empty:
        callback_success = send_decision_to_callback(result_df)
        if not callback_success:
            logger.error("决策结果发送失败")

    return result_df

def send_decision_to_callback(decision_data):
    """
    将CIP决策结果发送到回调接口
    
    功能：将分析结果按照API格式封装，通过HTTP POST发送到回调地址
    
    Args:
        decision_data: pd.DataFrame，决策数据，包含机组类型和CIP时机
    
    Returns:
        bool: 发送成功返回True，失败返回False
    """
    if config is None:
        logger.error("配置文件未加载")
        return False
    
    try:
        # 构建回调URL
        callback_url = config['api']['base_url'] + config['api']['callback_endpoint']
        
        # 设置请求头
        headers = {
            "Content-Type": "application/json",
            "JWT-TOKEN": config['api']['jwt_token']
        }
        
        # 获取项目ID
        project_id = config['scada']['project_id']
        
        # 构造回调数据
        callback_list = []
        if isinstance(decision_data, pd.DataFrame):
            for _, row in decision_data.iterrows():
                if pd.notna(row["CIP时机"]):
                    callback_list.append({
                        "type": row["机组类型"],
                        "project_id": project_id,
                        "ctime": row["CIP时机"].strftime("%Y-%m-%d %H:%M:%S")
                    })
        else:
            callback_list = [decision_data]
        
        # 封装为API要求的格式
        payload = {
            "list": callback_list
        }
        
        # 发送HTTP请求（带重试机制）
        max_retries = 3
        retry_interval = 10
        log_type = callback_list[0]["type"] if callback_list else "UNKNOWN"
        
        for attempt in range(1, max_retries + 1):
            try:
                response = requests.post(callback_url, headers=headers, json=payload, timeout=15)
                response.raise_for_status()
                
                logger.info(f"[{log_type}] 回调成功")
                return True
                
            except requests.exceptions.RequestException as e:
                if attempt < max_retries:
                    time.sleep(retry_interval)
                else:
                    logger.error(f"[{log_type}] 回调失败（重试{max_retries}次）: {e}")
        
        return False
        
    except Exception as e:
        logger.error(f"构建回调数据时出错: {e}")
        return False

if __name__ == '__main__':
    # 示例调用
    # main()  # 使用当前时间
    main(start_date='2025-08-26 00:00:00', unit_filter='RO1')  # 使用历史时间