""" run_dqn_decide.py UF 超滤 DQN 决策主入口（Inference / Online Assist） 职责： 1. 构造物理世界（physics） 2. 实例化决策器（UFDQNDecider） 3. 构造当前工厂状态（observation） 4. 调用模型给出策略建议 5. 生成 PLC 下发指令（限幅 / 限速） 6. 评估该指令在物理模型下的效果（只评估，不下发） """ from pathlib import Path import numpy as np # ========== 参数 / 物理 ========== from uf_train.env.uf_resistance_models_load import load_resistance_models from uf_train.env.uf_physics import UFPhysicsModel from uf_train.env.env_params import UFState, UFPhysicsParams, UFStateBounds, UFRewardParams, UFActionSpec # ========== 决策器 ========== from uf_train.rl_model.DQN.dqn_decider import UFDQNDecider def build_physics(): """ 构造与训练一致的物理模型（只做一次） """ phys_params = UFPhysicsParams() res_fp, res_bw = load_resistance_models(phys_params) physics = UFPhysicsModel( phys_params=phys_params, resistance_model_fp=res_fp, resistance_model_bw=res_bw, ) return physics def generate_plc_instructions(current_L_s, current_t_bw_s, model_prev_L_s, model_prev_t_bw_s, model_L_s, model_t_bw_s): """ 根据工厂当前值、模型上一轮决策值和模型当前轮决策值，生成PLC指令。 新增功能： 1. 处理None值情况：如果模型上一轮值为None，则使用工厂当前值； 如果工厂当前值也为None，则返回None并提示错误。 """ action_spec = UFActionSpec() adjustment_threshold = 1.0 # 处理None值情况 if model_prev_L_s is None: if current_L_s is None: print("错误: 过滤时长的工厂当前值和模型上一轮值均为None") return None, None else: # 使用工厂当前值作为基准 effective_current_L = current_L_s source_L = "工厂当前值(模型上一轮值为None)" else: # 模型上一轮值不为None，继续检查工厂当前值 if current_L_s is None: effective_current_L = model_prev_L_s source_L = "模型上一轮值(工厂当前值为None)" else: effective_current_L = model_prev_L_s source_L = "模型上一轮值" # 对反洗时长进行同样的处理 if model_prev_t_bw_s is None: if current_t_bw_s is None: print("错误: 反洗时长的工厂当前值和模型上一轮值均为None") return None, None else: effective_current_t_bw = current_t_bw_s source_t_bw = "工厂当前值(模型上一轮值为None)" else: if current_t_bw_s is None: effective_current_t_bw = model_prev_t_bw_s source_t_bw = "模型上一轮值(工厂当前值为None)" else: effective_current_t_bw = model_prev_t_bw_s source_t_bw = "模型上一轮值" # 检测所有输入值是否在规定范围内（只对非None值进行检查） # 工厂当前值检查（警告） if current_L_s is not None and not (action_spec.L_min_s <= current_L_s <= action_spec.L_max_s): print(f"警告: 当前过滤时长 {current_L_s} 秒不在允许范围内 [{action_spec.L_min_s}, {action_spec.L_max_s}]") if current_t_bw_s is not None and not (action_spec.t_bw_min_s <= current_t_bw_s <= action_spec.t_bw_max_s): print(f"警告: 当前反洗时长 {current_t_bw_s} 秒不在允许范围内 [{action_spec.t_bw_min_s}, {action_spec.t_bw_max_s}]") # 模型上一轮决策值检查（警告） if model_prev_L_s is not None and not (action_spec.L_min_s <= model_prev_L_s <= action_spec.L_max_s): print(f"警告: 模型上一轮过滤时长 {model_prev_L_s} 秒不在允许范围内 [{action_spec.L_min_s}, {action_spec.L_max_s}]") if model_prev_t_bw_s is not None and not (action_spec.t_bw_min_s <= model_prev_t_bw_s <= action_spec.t_bw_max_s): print(f"警告: 模型上一轮反洗时长 {model_prev_t_bw_s} 秒不在允许范围内 [{action_spec.t_bw_min_s}, {action_spec.t_bw_max_s}]") # 模型当前轮决策值检查（错误） if model_L_s is None: raise ValueError("错误: 决策模型建议的过滤时长不能为None") elif not (action_spec.L_min_s <= model_L_s <= action_spec.L_max_s): raise ValueError(f"错误: 决策模型建议的过滤时长 {model_L_s} 秒不在允许范围内 [{action_spec.L_min_s}, {action_spec.L_max_s}]") if model_t_bw_s is None: raise ValueError("错误: 决策模型建议的反洗时长不能为None") elif not (action_spec.t_bw_min_s <= model_t_bw_s <= action_spec.t_bw_max_s): raise ValueError(f"错误: 决策模型建议的反洗时长 {model_t_bw_s} 秒不在允许范围内 [{action_spec.t_bw_min_s}, {action_spec.t_bw_max_s}]") print(f"过滤时长基准: {source_L}, 值: {effective_current_L}") print(f"反洗时长基准: {source_t_bw}, 值: {effective_current_t_bw}") # 使用选定的基准值进行计算调整 L_diff = model_L_s - effective_current_L L_adjustment = 0 if abs(L_diff) >= adjustment_threshold * action_spec.L_step_s: if L_diff >= 0: L_adjustment = action_spec.L_step_s else: L_adjustment = -action_spec.L_step_s next_L_s = effective_current_L + L_adjustment t_bw_diff = model_t_bw_s - effective_current_t_bw t_bw_adjustment = 0 if abs(t_bw_diff) >= adjustment_threshold * action_spec.t_bw_step_s: if t_bw_diff >= 0: t_bw_adjustment = action_spec.t_bw_step_s else: t_bw_adjustment = -action_spec.t_bw_step_s next_t_bw_s = effective_current_t_bw + t_bw_adjustment return next_L_s, next_t_bw_s def calc_uf_cycle_metrics(current_state, max_tmp_during_filtration, min_tmp_during_filtration, L_s: float, t_bw_s: float): """ 计算 UF 超滤系统的核心性能指标 参数: L_s (float): 单次过滤时间（秒） t_bw_s (float): 单次反洗时间（秒） 返回: dict: { "k_bw_per_ceb": 小周期次数, "ton_water_energy_kWh_per_m3": 吨水电耗, "recovery": 回收率, "net_delivery_rate_m3ph": 净供水率 (m³/h), "daily_prod_time_h": 日均产水时间 (小时/天) "max_permeability": 全周期最高渗透率(lmh/bar) } """ # 模拟该参数下的超级周期 info, next_state = physics.simulate_one_supercycle(current_state, L_s=L_s, t_bw_s=t_bw_s) # 获得模型模拟周期信息 k_bw_per_ceb = info["k_bw_per_ceb"] ton_water_energy_kWh_per_m3 = info["ton_water_energy_kWh_per_m3"] recovery = info["recovery"] daily_prod_time_h = info["daily_prod_time_h"] # 获得模型模拟周期内最高跨膜压差/最低跨膜压差 if max_tmp_during_filtration is None: max_tmp_during_filtration = info["max_TMP_during_filtration"] if min_tmp_during_filtration is None: min_tmp_during_filtration = info["min_TMP_during_filtration"] # 计算最高渗透率 max_permeability = 100 * current_state.q_UF / (128*40) / min_tmp_during_filtration return { "k_bw_per_ceb": k_bw_per_ceb, "ton_water_energy_kWh_per_m3": ton_water_energy_kWh_per_m3, "recovery": recovery, "daily_prod_time_h": daily_prod_time_h, "max_permeability": max_permeability } def run_dqn_decide( model_path: Path, physics, # -------- 工厂当前值 -------- current_state: UFState ): """ 单轮 DQN 决策流程 """ # 构造决策器 decider = UFDQNDecider( physics=physics, model_path=model_path, seed=0, ) # 模型决策（不推进真实环境） decision = decider.decide(current_state) action_id = decision["action_id"] model_L_s = decision["L_s"] model_t_bw_s = decision["t_bw_s"] return action_id, model_L_s, model_t_bw_s # ============================== # 示例调用 # ============================== if __name__ == "__main__": MODEL_PATH = "model/dqn_model.zip" TMP0 = 0.019 # 原始 TMP0 q_UF = 300 # 进水流量 temp = 20.0 #进水温度 current_state = UFState(TMP=TMP0, q_UF=q_UF, temp=temp) physics = build_physics() action_id, model_L_s, model_t_bw_s = run_dqn_decide( model_path=MODEL_PATH, physics=physics, current_state=current_state, ) # 环境实例化，模型加载等功能放在UFDQNDecider类中 current_L_s = 3800 current_t_bw_s = 40 model_prev_L_s = 4040 model_prev_t_bw_s = 60 L_s, t_bw_s = generate_plc_instructions(current_L_s, current_t_bw_s, model_prev_L_s, model_prev_t_bw_s, model_L_s, model_t_bw_s) # 获取模型下发指令 L_s = 4100 t_bw_s = 96 max_tmp_during_filtration = 0.050176 # 新增工厂数据接口：周期最高/最低跨膜压差，无工厂数据接入时传入None，calc_uf_cycle_metrics()自动获取模拟周期中的跨膜压差最值 min_tmp_during_filtration = 0.012496 execution_result = calc_uf_cycle_metrics(current_state, max_tmp_during_filtration, min_tmp_during_filtration, L_s, t_bw_s) print("\n===== 单步决策结果 =====") print(f"模型选择的动作: {action_id}") print(f"模型选择的L_s: {model_L_s} 秒, 模型选择的t_bw_s: {model_t_bw_s} 秒") print(f"指令下发的L_s: {L_s} 秒, 指令下发的t_bw_s: {t_bw_s} 秒") print(f"指令对应的反洗次数: {execution_result['k_bw_per_ceb']}") print(f"指令对应的吨水电耗: {execution_result['ton_water_energy_kWh_per_m3']}") print(f"指令对应的回收率: {execution_result['recovery']}") print(f"指令对应的日均产水时间: {execution_result['daily_prod_time_h']}") print(f"指令对应的最高渗透率: {execution_result['max_permeability']}")