278 lines
10 KiB
Python
278 lines
10 KiB
Python
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()
|