DualFlow/models/uf-rl/rl_model/DQN/dqn_model/dqn_statebuilder.py

from typing import Dict
import pandas as pd

# -------------------------------
# 引入环境状态模板（最终输出）
# -------------------------------
from env.env_params import UFState

# -------------------------------
# 引入分析类
# -------------------------------
from uf_data_process.load import UFConfigLoader
from uf_data_process.label import UFEventClassifier
from uf_data_process.filter import EventQualityFilter, InletSegmentFilter
from uf_data_process.calculate import UFResistanceCalculator, UFResistanceAnalyzer
from uf_data_process.fit import ShortTermCycleFoulingFitter, LongTermFoulingFitter

class DQNStateBuilder:
    """
    在 DQN 决策前构建状态的工具类

    相关数据：
        * CSV1 = 上一完整化学周期
        * CSV2 = 新周期初始进水段
    """

    def __init__(self, config_path: str):
        """
        Parameters
        ----------
        config_path : str
            uf_analyze_config.yaml 路径
        """
        self.cfg = UFConfigLoader(config_path)
        self.uf_cfg = self.cfg.uf
        self.params = self.cfg.params

        self.units = self.uf_cfg["units"]
        self.column_formats = self.uf_cfg.get("column_formats", {})
        self.flow_format = self.column_formats.get("flow_col", "C.M.{unit}_FT_JS@out")
        self.ctrl_format = self.column_formats.get("ctrl_col", "C.M.{unit}_DB@word_control")

        self.area_m2 = self.uf_cfg["area_m2"]

        self.scale_factor = self.params.get("scale_factor", 1e10)
        self.segment_head_n = self.params.get("segment_head_n", 10)
        self.segment_tail_n = self.params.get("segment_tail_n", 10)

    # ======================================================================
    # 对外主接口
    # ======================================================================

    def build_from_csv_pair(
        self,
        prev_cycle_csv: str,
        init_cycle_csv: str,
    ) -> UFState:
        """
        使用【上一完整化学周期 CSV】+【当前周期初始 CSV】构建 UFState
        """

        df_prev = pd.read_csv(prev_cycle_csv)
        df_init = pd.read_csv(init_cycle_csv)

        # 自动识别 UF 单元编号（UF1 / UF2 / ...）
        unit_id = self._infer_unit_id(df_prev)

        # 分别处理两个 CSV
        prev_features = self._analyze_previous_cycle_csv(df_prev, unit_id)
        init_features = self._analyze_init_cycle_csv(df_init, unit_id)

        # 化学清洗去除阻力（上一周期末 - 当前初始）
        ceb_removal = max(
            prev_features["R_end"] - init_features["R_start"],
            0.0
        )

        # 构建 UFState
        state = UFState(
            q_UF=init_features["q_mean"],
            TMP=init_features["tmp_mean"],
            temp=init_features["temp_mean"],
            R=init_features["R_start"],
            nuK=prev_features["nuK"],
            slope=prev_features["slope"],
            power=prev_features["power"],
            ceb_removal=ceb_removal,
        )

        return state

    # ======================================================================
    # 上一完整化学周期分析
    # ======================================================================

    def _analyze_previous_cycle_csv(
        self,
        df: pd.DataFrame,
        unit_id: str,
    ) -> Dict[str, float]:
        """
        上一完整化学周期分析逻辑

        步骤：
        1. 事件标注
        2. 进水段过滤（质量过滤）
        3. 膜阻力计算
        4. 提取周期末稳定阻力
        5. 拟合 nuK
        6. 拟合长期不可逆污染（slope / power）
        """

        # -------- 事件标注 --------
        df = self._label_events(df, unit_id)


        # -------- 保留过滤进水段 --------
        inlet_filter = InletSegmentFilter(
            control_col=f"C.M.{unit_id}_DB@word_control",
            stable_value=self.uf_cfg["stable_inlet_code"],
            min_points=self.params["min_stable_points"],
        )
        segments = inlet_filter.extract(df)

        quality_filter = EventQualityFilter(
            min_points=self.params["min_stable_points"]
        )
        segments = quality_filter.filter(segments)

        if len(segments) == 0:
            raise ValueError("上一周期无有效稳定进水段，无法构建状态")

        # -------- 3️⃣ 膜阻力计算 --------
        res_calc = UFResistanceCalculator(
            units=[unit_id],
            area_m2=self.area_m2,
            scale_factor=self.scale_factor,
        )
        segments = res_calc.calculate_for_segments(
            segments,
            temp_col=self.uf_cfg["temp_col"],
            flow_col=self.uf_cfg["flow_col_template"].format(unit=unit_id),
        )



        # -------- 膜阻力统计 --------
        res_col = f"{unit_id}_R_scaled"
        ura = UFResistanceAnalyzer(
            resistance_col=res_col,
            head_n=self.segment_head_n,
            tail_n=self.segment_tail_n
        )
        segments = ura.analyze_segments(segments)
        df_all = segments[-1]
        R_end = df_all["R_scaled_end"].iloc[0]

        # ===== 确保 time 为 datetime =====
        for i, seg in enumerate(segments):
            if not pd.api.types.is_datetime64_any_dtype(seg["time"]):
                seg = seg.copy()
                seg["time"] = pd.to_datetime(seg["time"], errors="coerce")
                seg = seg.dropna(subset=["time"])
                segments[i] = seg

        # -------- 5️⃣ 短期污染拟合（nuK）--------
        st_fitter = ShortTermCycleFoulingFitter(unit_id)
        nuK, _ = st_fitter.fit_cycle(segments)

        # -------- 6️⃣ 长期不可逆污染拟合 --------
        lt_fitter = LongTermFoulingFitter(unit_id)
        slope, power, _ = lt_fitter.fit_cycle(segments)

        return {
            "R_end": R_end,
            "nuK": float(nuK),
            "slope": float(slope),
            "power": float(power),
        }

    # ======================================================================
    # 当前周期初始进水段分析
    # ======================================================================

    def _analyze_init_cycle_csv(
        self,
        df: pd.DataFrame,
        unit_id: str,
    ) -> Dict[str, float]:
        """
        当前周期初始进水段分析

        特点：
        - 不切段
        - 不过滤
        - 只计算均值
        """

        # -------- 1️⃣ 事件标注 --------
        df = self._label_events(df, unit_id)

        # -------- 2️⃣ 仅保留进水行 --------
        df = df[df["event_type"] == "inlet"].copy()

        if df.empty:
            raise ValueError("初始 CSV 中无进水数据")

        # -------- 3️⃣ 膜阻力计算 --------
        res_calc = UFResistanceCalculator(
            units=[unit_id],
            area_m2=self.area_m2,
            scale_factor=self.scale_factor,
        )
        segments = [df]
        segments = res_calc.calculate_for_segments(
            segments,
            temp_col=self.uf_cfg["temp_col"],
            flow_col=self.uf_cfg["flow_col_template"].format(unit=unit_id),
        )
        df = segments[-1]

        flow_col = self.uf_cfg["flow_col_template"].format(unit=unit_id)
        temp_col = self.uf_cfg["temp_col"]
        press_col = f"C.M.{unit_id}_DB@press_PV"
        res_col = f"{unit_id}_R_scaled"

        return {
            "q_mean": float(df[flow_col].mean()),
            "tmp_mean": float(df[press_col].mean()),
            "temp_mean": float(df[temp_col].mean()),
            "R_start": float(df[res_col].mean()),
        }

    # ======================================================================
    # 工具函数
    # ======================================================================

    def _infer_unit_id(self, df: pd.DataFrame) -> str:
        """
        根据列名自动识别 UF 单元编号
        """
        for unit in self.units:
            key = self.flow_format.format(unit=unit)
            if key in df.columns:
                return unit
        raise ValueError("无法从 CSV 列名识别 UF 单元编号")

    def _label_events(self, df: pd.DataFrame, unit_id: str) -> pd.DataFrame:
        """
        为 DataFrame 标注 event_type
        """
        ctrl_col = self.uf_cfg["flow_col_template"].format(unit=unit_id)
        clf = UFEventClassifier(
            unit_name=unit_id,
            inlet_codes=self.uf_cfg["inlet_codes"],
            physical_code=self.uf_cfg["physical_bw_code"],
            chemical_code=self.uf_cfg["chemical_bw_code"],
            ctrl_col
        )
        df = clf.classify(df)
        df = clf.segment(df)
        return df