fix: 补充全状态向量构建依赖模块

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

@@ -1,17 +1,15 @@
 from typing import Dict
+from dataclasses import replace
+
+import numpy as np
 import pandas as pd
+from sklearn.metrics import r2_score
 
-# -------------------------------
-# 引入环境状态模板（最终输出）
-# -------------------------------
 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.label import UFEventClassifier, PostBackwashInletMarker
+from uf_data_process.filter import ConstantFlowFilter,EventQualityFilter, InletSegmentFilter,FlowOutlierFilter
 from uf_data_process.calculate import UFResistanceCalculator, UFResistanceAnalyzer
 from uf_data_process.fit import ShortTermCycleFoulingFitter, LongTermFoulingFitter
 
@@ -22,6 +20,7 @@ class DQNStateBuilder:
     相关数据：
         * CSV1 = 上一完整化学周期
         * CSV2 = 新周期初始进水段
+        * CSV3 = 新周期预测进水段
     """
 
     def __init__(self, config_path: str):
@@ -32,19 +31,33 @@ class DQNStateBuilder:
             uf_analyze_config.yaml 路径
         """
         self.cfg = UFConfigLoader(config_path)
-        self.uf_cfg = self.cfg.uf
-        self.params = self.cfg.params
+        uf_cfg = self.cfg.uf
+        params = self.cfg.params
+
+
+        self.units = uf_cfg.get("units", ["UF1", "UF2", "UF3", "UF4"])
+        self.stable_inlet_code = uf_cfg.get("stable_inlet_code", [24.0, 26.0])
 
-        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")
+        column_formats = uf_cfg.get("column_formats", {})
+        self.ctrl_format = column_formats.get("ctrl_col", "C.M.{unit}_DB@word_control")
+        self.flow_format = column_formats.get("flow_col", "C.M.{unit}_FT_JS@out")
+        self.tmp_format = column_formats.get("tmp_col", "C.M.{unit}_DB@press_PV")
+        self.per_format = column_formats.get("per_col", "{unit}Per")
+        self.temp_col = column_formats.get("temp_col", "C.M.RO_TT_ZJS@out")
 
-        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)
+        # 过滤器
+        self.min_points = params.get("min_points", 20)
+        self.initial_points = params.get("initial_points", 10)
+
+        self.quality_filter = EventQualityFilter(min_points=self.min_points)
+        self.flow_filter = FlowOutlierFilter(n_sigma=3)
+        self.initial_label = PostBackwashInletMarker(n_points=self.initial_points)
+
+        # 阻力计算器
+        self.res_calc = UFResistanceCalculator(self.units, area_m2=uf_cfg["area_m2"], scale_factor=params.get("scale_factor", 1e10))
+        self.segment_head_n = params.get("segment_head_n", 10)
+        self.segment_tail_n = params.get("segment_tail_n", 10)
 
     # ======================================================================
     # 对外主接口
@@ -52,22 +65,24 @@ class DQNStateBuilder:
 
     def build_from_csv_pair(
         self,
+        unit_name,
+        uf_state_default,
+        state_bounds,
         prev_cycle_csv: str,
         init_cycle_csv: str,
+        predict_cycle_csv: str,
     ) -> UFState:
         """
-        使用【上一完整化学周期 CSV】+【当前周期初始 CSV】构建 UFState
+        使用【上一完整化学周期 CSV】+【当前周期初始 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)
+        df_predict = pd.read_csv(predict_cycle_csv)
 
         # 分别处理两个 CSV
-        prev_features = self._analyze_previous_cycle_csv(df_prev, unit_id)
-        init_features = self._analyze_init_cycle_csv(df_init, unit_id)
+        prev_features = self._analyze_previous_cycle_csv(df_prev, unit_name, uf_state_default, state_bounds)
+        init_features = self._analyze_init_cycle_csv(df_init, unit_name, uf_state_default, state_bounds)
 
         # 化学清洗去除阻力（上一周期末 - 当前初始）
         ceb_removal = max(
@@ -75,19 +90,31 @@ class DQNStateBuilder:
             0.0
         )
 
+        # 使用df_predict修正 nuk
+        corrected_nuk = self._correct_nuk_with_predict(
+            df_predict=df_predict,
+            unit_name=unit_name,
+            R_start=init_features["R_start"],
+            q_mean=init_features["q_mean"],
+            temp_celsius=init_features["temp_mean"],
+            base_nuk=prev_features["nuK"],
+            uf_state_default=uf_state_default,
+        )
+
         # 构建 UFState
-        state = UFState(
-            q_UF=init_features["q_mean"],
+        current_state = replace(
+            uf_state_default,
             TMP=init_features["tmp_mean"],
+            q_UF=init_features["q_mean"],
             temp=init_features["temp_mean"],
-            R=init_features["R_start"],
-            nuK=prev_features["nuK"],
+            R = init_features["R_start"],
+            nuK = corrected_nuk,
             slope=prev_features["slope"],
             power=prev_features["power"],
             ceb_removal=ceb_removal,
         )
 
-        return state
+        return current_state
 
     # ======================================================================
     # 上一完整化学周期分析
@@ -95,8 +122,10 @@ class DQNStateBuilder:
 
     def _analyze_previous_cycle_csv(
         self,
-        df: pd.DataFrame,
-        unit_id: str,
+        df,
+        unit_name,
+        uf_state_default,
+        state_bounds
     ) -> Dict[str, float]:
         """
         上一完整化学周期分析逻辑
@@ -110,66 +139,80 @@ class DQNStateBuilder:
         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"]
+        ctrl_col = self.ctrl_format.format(unit=unit_name)
+        flow_col = self.flow_format.format(unit=unit_name)
+        tmp_col = self.tmp_format.format(unit=unit_name)
+
+        # 事件标注
+        event_clf = UFEventClassifier(unit_name, self.cfg.uf["inlet_codes"],
+                                      self.cfg.uf["physical_bw_code"], self.cfg.uf["chemical_bw_code"],
+                                      ctrl_col)
+        df_unit = event_clf.classify(df)  # 产生 event_type 列
+        df_unit_mark = self.initial_label.mark(df_unit)  # 标记反冲洗事件后的前 N 个进水点
+        seg_df = event_clf.segment(df_unit_mark)  # 根据 event_type 列编号事件段落
+
+        # 对 seg_df 进行按 segment 分组后逐段过滤：
+        const_flow_filter = ConstantFlowFilter(flow_col=flow_col, repeat_len=20)
+        segments = const_flow_filter.filter(seg_df)  # 去除出现网络错误的进水段
+        segments = self.quality_filter.filter(segments)  # 去除时间过短的进水段
+        # 提取稳定进水段
+        stable_extractor = InletSegmentFilter(ctrl_col, stable_codes=self.stable_inlet_code, min_points=self.min_points)
+        stable_segments = stable_extractor.extract(segments)  # 提取稳定进水数据
+
+        if len(stable_segments) == 0:
+            raise ValueError("上一周期无有效稳定进水段，无法构建状态，请使用run_dqn_decide.py")
+
+        # 膜阻力计算
+        stable_segments = self.res_calc.calculate_for_segments(
+            stable_segments,
+            temp_col=self.temp_col,
+            flow_col=flow_col,
+            tmp_col=tmp_col,
         )
-        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"
+        res_col = f"{unit_name}_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]
+        stable_segments = ura.analyze_segments(stable_segments)
+        df_all = stable_segments[-1]
         R_end = df_all["R_scaled_end"].iloc[0]
 
         # ===== 确保 time 为 datetime =====
-        for i, seg in enumerate(segments):
+        for i, seg in enumerate(stable_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
+                stable_segments[i] = seg
 
         # -------- 5️⃣ 短期污染拟合（nuK）--------
-        st_fitter = ShortTermCycleFoulingFitter(unit_id)
-        nuK, _ = st_fitter.fit_cycle(segments)
+        st_fitter = ShortTermCycleFoulingFitter(unit_name)
+        nuK, st_r2 = st_fitter.fit_cycle(stable_segments)
+        if (
+                pd.isna(nuK)
+                or pd.isna(st_r2)
+                or not np.isfinite(nuK)
+                or st_r2 < 0.4
+        ):
+            nuK = uf_state_default.nuK
 
         # -------- 6️⃣ 长期不可逆污染拟合 --------
-        lt_fitter = LongTermFoulingFitter(unit_id)
-        slope, power, _ = lt_fitter.fit_cycle(segments)
+        lt_fitter = LongTermFoulingFitter(unit_name)
+        slope, power, lt_r2 = lt_fitter.fit_cycle(stable_segments)
+        if (
+                pd.isna(slope)
+                or pd.isna(power)
+                or pd.isna(lt_r2)
+                or not np.isfinite(slope)
+                or not np.isfinite(power)
+                or lt_r2 < 0.4
+        ):
+            slope = uf_state_default.slope
+            power = uf_state_default.power
 
         return {
             "R_end": R_end,
@@ -184,79 +227,106 @@ class DQNStateBuilder:
 
     def _analyze_init_cycle_csv(
         self,
-        df: pd.DataFrame,
-        unit_id: str,
+        df,
+        unit_name,
+        uf_state_default,
+        state_bounds
     ) -> Dict[str, float]:
         """
         当前周期初始进水段分析
-
-        特点：
-        - 不切段
-        - 不过滤
-        - 只计算均值
         """
 
-        # -------- 1️⃣ 事件标注 --------
-        df = self._label_events(df, unit_id)
+        flow_col = self.flow_format.format(unit=unit_name)
+        tmp_col = self.tmp_format.format(unit=unit_name)
+        temp_col = self.temp_col
+        res_col = f"{unit_name}_R_scaled"
 
-        # -------- 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 = self.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),
+            temp_col=self.temp_col,
+            flow_col=flow_col,
+            tmp_col=tmp_col,
         )
         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()),
+            "tmp_mean": float(df[tmp_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:
+    def _correct_nuk_with_predict(
+            self,
+            df_predict,
+            unit_name,
+            R_start,
+            q_mean,
+            temp_celsius,
+            base_nuk,
+            uf_state_default,
+    ):
         """
-        根据列名自动识别 UF 单元编号
+        使用 predict_csv 中预测的渗透率，
+        对短期污染参数 nuK 进行在线修正。
         """
-        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
+        per_col = self.per_format.format(unit=unit_name)
+        if per_col not in df_predict.columns:
+            return base_nuk
+        per = pd.to_numeric(
+            df_predict[per_col],
+            errors="coerce"
+        )
+        mask = (
+                np.isfinite(per)
+                & (per > 0)
+        )
+        per = per[mask]
+
+        if len(per) < 5:
+            return base_nuk
+
+
+        mu = self.res_calc.xishan_viscosity(temp_celsius)
+        R = 3.6e11 / (mu * per)
+        R_scaled = R / self.res_calc.scale_factor
+        R_scaled = (R_scaled- R_scaled.iloc[0]+ R_start)
+        t = np.arange(len(R_scaled)) * 60.0
+        J_const = q_mean / self.res_calc.A / 3600.0
+
+        if not np.isfinite(J_const) or J_const <= 0:
+            return base_nuk
+
+        x = J_const * t
+
+        try:
+            coef = np.polyfit(x, R_scaled, 1)
+            predict_nuk = float(coef[0])
+            pred = np.polyval(coef, x)
+            r2 = r2_score(R_scaled, pred)
+
+        except Exception:
+            return base_nuk
+
+        if (
+                not np.isfinite(r2)
+                or not np.isfinite(predict_nuk)
+                or predict_nuk <= 0
+                or r2 < 0.4
+        ):
+            return base_nuk
+
+        corrected_nuk = (
+                0.7 * base_nuk
+                + 0.3 * predict_nuk
         )
-        df = clf.classify(df)
-        df = clf.segment(df)
-        return df
+
+        if (
+                not np.isfinite(corrected_nuk)
+                or corrected_nuk <= 0
+        ):
+            return uf_state_default.nuK
+
+        return float(corrected_nuk)