import gymnasium as gym from gymnasium import spaces import numpy as np from engine_env.core_model import AeroEngineDLL from engine_env.schedule import ControlSchedule class AeroEngineGymEnv(gym.Env): """ 航空发动机控制环境 (Aero-Engine Control Environment) 特点: 1. 外部通过 set_pla() 控制任务输入。 2. 包含起动阶段和稳态控制阶段的复杂约束判定。 3. 奖励函数包含调节时间、超调量、起动功、平滑性等工程指标。 Observation Space (8维): [NH, NL, T5, P3, Target_NH, Target_NL, Error_NH, Error_NL] Action Space (5维, 范围 [-1, 1]): [FanVane, CompVane, AddPower, Wf, A8] """ metadata = {"render_modes": ["human"]} def __init__(self): # 1. 加载核心组件 self.engine = AeroEngineDLL() self.schedule = ControlSchedule() # 2. 仿真参数 self.current_step = 0 self.dt = 0.02 # 仿真步长 20ms self.max_steps = 1000 # 最大步数 (20秒) # 3. 核心状态变量 self.pla = 0.0 # 油门杆角度 (由外部控制) # 4. 定义动作空间 (5个控制量) # 顺序: [风扇导叶, 高压导叶, 起动功率, 燃油Wf, A8] self.action_space = spaces.Box(low=-1.0, high=1.0, shape=(5,), dtype=np.float32) # 5. 定义观测空间 (8维) self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(8,), dtype=np.float32) # 6. 物理范围映射 (Physical Limits) self.limits = { 'Wf': (50.0, 5000.0), # kg/h (燃油流量) 'A8': (0.2, 0.6), # m^2 (喷管面积) 'FanVane': (-20.0, 20.0), # deg (风扇导叶) 'CompVane': (-20.0, 20.0), # deg (压气机导叶) 'AddPower': (0.0, 100000.0) # W (起动电机功率) } # 内部缓存变量 self.t_nh, self.t_nl = 0.0, 0.0 self.limit_t5, self.limit_p3 = 1200.0, 2000.0 # 计时器 (用于约束判定) self.timer_startup = 0.0 self.timer_settle_5 = 0.0 self.timer_settle_1 = 0.0 # 平滑性权重 (Smoothness Weights) # 燃油(Wf)和喷管(A8)给高权重，AddPower允许突变 self.smooth_weights = np.array([1.0, 1.0, 0.5, 5.0, 2.0], dtype=np.float32) # 初始化上一帧动作缓存 self.last_action = np.zeros(5, dtype=np.float32) def reset(self, seed=None, options=None): """ 重置环境到初始状态 options: {'pla': float} 可指定初始 PLA """ super().reset(seed=seed) self.current_step = 0 # 重置所有计时器 self.timer_startup = 0.0 self.timer_settle_5 = 0.0 self.timer_settle_1 = 0.0 # 重置动作缓存 self.last_action = np.zeros(5, dtype=np.float32) # 设定初始 PLA self.pla = options.get('pla', 0.0) if options else 0.0 # 复位底层模型 out = self.engine.reset() # 更新初始目标 self.t_nh, self.t_nl, self.limit_t5, self.limit_p3 = self.schedule.get_targets(self.pla) return self._get_obs(out), {"PLA": self.pla} def set_pla(self, pla_value): """ 【外部接口】设置当前的 PLA (油门杆角度) """ self.pla = np.clip(pla_value, 0.0, 110.0) def step(self, action): """ 环境交互核心函数 action: AI 输出的归一化动作 [-1, 1] """ self.current_step += 1 # ======================================================= # 1. 查表：更新控制目标和限制 (基于当前的 self.pla) # ======================================================= self.t_nh, self.t_nl, self.limit_t5, self.limit_p3 = self.schedule.get_targets(self.pla) # ======================================================= # 2. 动作反归一化 (AI [-1, 1] -> Physical Value) # ======================================================= phys_action = { 'FanVane': self._denormalize(action[0], 'FanVane'), 'CompVane': self._denormalize(action[1], 'CompVane'), 'AddPower': self._denormalize(action[2], 'AddPower'), 'Wf': self._denormalize(action[3], 'Wf'), 'A8': self._denormalize(action[4], 'A8'), } # ======================================================= # 3. 执行物理仿真 # ======================================================= out = self.engine.step(phys_action) # ======================================================= # 4. 计算奖励 (Reward Function) # ======================================================= # 传入 normalized action (action) 用于计算平滑度 total_reward = self._calculate_official_reward(out, phys_action, self.pla, action) # ======================================================= # 5. 终止条件判定 (Terminated Check) # ======================================================= terminated, term_penalty = self._check_terminated(out, self.pla) total_reward += term_penalty truncated = (self.current_step >= self.max_steps) # ======================================================= # 6. 组装返回信息 # ======================================================= obs = self._get_obs(out) info = { "PLA": self.pla, "Real_NH": out.NH, "Target_NH": self.t_nh, "Real_T5": out.T5t, "Action_Physical": phys_action } return obs, total_reward, terminated, truncated, info def _check_terminated(self, out, current_pla): """ 终止判定 """ # ======================================================= # 全局硬约束 - 只有炸机才重开 # ======================================================= # 1. 严重超温 if out.T5t > 1400.0: # print(f"[Terminated] T5 High: {out.T5t:.1f}") return True, -1000.0 # 2. 严重超转 if out.NH > 1.10: # print(f"[Terminated] NH High: {out.NH:.3f}") return True, -1000.0 if out.NL > 1.10: # print(f"[Terminated] NL High: {out.NL:.3f}") return True, -1000.0 return False, 0.0 def _calculate_official_reward(self, out, phys_action, current_pla, normalized_action): """ 官方奖励计算函数 (Official Reward Function) """ step_reward = 0.0 # 提取目标和当前值 t_nh, t_nl = self.t_nh, self.t_nl nh, nl = out.NH, out.NL # 计算误差 err_nh = t_nh - nh err_nl = t_nl - nl abs_err_nh = abs(err_nh) # ======================================================= # 阶段 A: 起动阶段 - PLA < 15 # ======================================================= if current_pla < 15.0: # 1. 【起动时间】 if nh < 0.60: step_reward -= 0.5 # 时间惩罚基数 # 2. 【起动功】 # 逻辑：使用的电功率越大，扣分越多 power_penalty = (phys_action['AddPower'] / 100000.0) * 0.2 step_reward -= power_penalty # 3. 跟踪引导 (主要看 NH) step_reward -= (err_nh ** 2) * 20.0 # ======================================================= # 阶段 B: 正常工作阶段- PLA >= 15 # ======================================================= else: # 1. 【调节时间】 if abs_err_nh > 0.02: # 2% 误差带 step_reward -= 0.5 # 调节时间惩罚 # 2. 【超调量】 if nh > t_nh: overshoot_amount = nh - t_nh step_reward -= overshoot_amount * 50.0 # 重罚超调 # 3. 基础跟踪 (主要看 NL 推力, 兼顾 NH) step_reward -= (err_nl ** 2) * 20.0 step_reward -= (err_nh ** 2) * 10.0 # ======================================================= # 全局安全约束 # ======================================================= # 1. 【涡轮后温度】 if out.T5t > self.limit_t5: over_temp = out.T5t - self.limit_t5 step_reward -= over_temp * 1 # 每超 1K 扣 1分 # 2. 【压气机后压力】 if out.P3s > self.limit_p3: over_pres = out.P3s - self.limit_p3 step_reward -= over_pres * 0.5 # 每超 1kPa 扣 0.5分 # ======================================================= # 3. 动作平滑性 (Action Smoothness) # ======================================================= # 计算动作变化率: delta = current - last delta_action = normalized_action - self.last_action # 计算加权平方和 (L2 Norm) smoothness_cost = np.sum(self.smooth_weights * np.square(delta_action)) # 乘以系数 (调节平滑性在总分中的占比) step_reward -= smoothness_cost * 0.1 # 更新上一帧动作缓存 self.last_action = normalized_action.copy() return step_reward def _get_obs(self, out): state = np.array([out.NH, out.NL, out.T5t, out.P3s], dtype=np.float32) targets = np.array([self.t_nh, self.t_nl], dtype=np.float32) errors = targets - state[:2] return np.concatenate([state, targets, errors]) def _denormalize(self, val, key): min_v, max_v = self.limits[key] return (val + 1.0) / 2.0 * (max_v - min_v) + min_v def close(self): if self.engine: self.engine.close()