USER
User: 模拟过程出错: No boundary conditions were applied to any surfaces! ---------------
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
User: 模拟过程出错: No boundary conditions were applied to any surfaces! ---------------
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 从错误消息 "No boundary conditions were applied to any surfaces!"（未将边界条件应用于任何表面）来看，您的OpenMC模拟可能缺少对几何表面的边界条件设置。在OpenMC中，定义几何边界条件是必需的，以确定粒子在与几何表面相互作用时的行为。
以下是一些可以尝试的步骤来解决这个问题：
1. **定义几何边界条件**：
在定义几何体时，如果几何包含无限空间或您希望设置周边的边界行为，您需要为表面指定边界条件。例如，您可以用 `boundary_type` 参数来设置表面的边界条件。
增加边界条件的设定到您的几何体定义中。例如，如果想定义反射边界，可以这样做：
```python
cylinder = openmc.ZCylinder(r=100, boundary_type='reflective')
```
2. **检查几何体完整性**：
确保几何体是闭合的，并且材料分配是在定义的区域内的，几何体需要是完整且封闭的。
3. **增加外部“盒子”**：
在反应堆或感兴趣的区域外增加一个大尺寸的盒子，并应用“外”边界条件（如真空或反射）：
```python
outer_box = openmc.model.rectangular_prism(
width=200, height=200, boundary_type='reflective'
)
```
并在创建geometry的时候包括这个外部“盒子”：
```python
outer_region = +outer_box & -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=outer_region)
```
通过以上设定，可以确保在OpenMC中的几何体中应用了正确的边界条件，从而消除这个错误。如果问题仍然存在，请确保代码中的几何和材料设置代码都已包含在内，并且没有遗漏其他所需的几何体或设置。
User: 将修改的全部代码写出来
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
User: 将刚才没输出完成的接上
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 为了修复"模拟过程出错: No boundary conditions were applied to any surfaces!" 的问题，需要确保在OpenMC的几何中正确定义了边界条件。这里是修改后的完整代码，包括边界条件的设置：
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100, boundary_type='reflective') # 增加了边界条件
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
outside_cell = openmc.Cell(region=+cylinder, fill=self.water) # 创建外部包围物
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell, outside_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
在这个版本中，我在创建圆柱体的时候增加了边界条件 `boundary_type='reflective'`。这样可以防止错误 "No boundary conditions were applied to any surfaces!" 的发生。这样为几何体定义了合适的边界条件以确保模拟的粒子行为是完整的。
User: ERROR: Nuclear data library does not contain cross sections for U235 at or near
900.000000 K.
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 错误消息 "Nuclear data library does not contain cross sections for U235 at or near 900.000000 K" 指出您的核数据库中没有在900 K附近的铀-235的截面数据。这可能是核数据库的温度范围不包括900 K，或者该温度附近没有进行插值。
以下是一些解决方案：
1. **更改温度设置**：
检查并更改材料温度设置为您的核数据库提供的温度。您可以通过查阅核数据库文档来找到可用的温度。如果库例如提供的是600 K，800 K，和1000 K，您可以使用其中的一个：
```python
self.fuel.temperature = 1000 # 使用可用的温度
```
2. **使用温度插值**：
如果库支持温度插值，可以设置 `temperature['method']` 为 'interpolation'，不过您已经设置了此选项。确保数据目录中的 `cross_sections.xml` 文件确实支持插值。否则，在所有使用的核材料上检查并使用确切可用的温度。
3. **核数据库问题**：
确保您的核数据目录配置路径正确，并且数据完整无误。某些数据包可能需要手动添加支持所需温度的截面。
如所有这些方法均不有效，请确保核数据库中确实包含所使用的材料和温度的截面数据，并考虑调整您的设计和模拟到支持的范围中。
User: 断点的作用
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 在程序调试中，**断点**的作用是暂停程序的执行进程，以便开发人员可以观察和检查程序在该特定点时的状态。这包括查看变量的值、程序流、内存状态和其他实时信息，可以帮助开发人员理解哪里出了问题以及其原因。
在Python中，通过集成开发环境（IDE）如PyCharm、Visual Studio Code和其他调试工具，我们可以很方便地设置断点。以下是一些使用断点的主要目的和优点：
1. **逐步执行**：通过在关键位置设置断点，可以一步一步地手动移动程序执行（即单步调试），从而观察程序状态及其变化。
2. **检查状态**：在断点处，可以检查变量的值并更了解其上下文，这对于找出逻辑错误、验证算法正确性非常重要。
3. **条件断点**：可以设置某些条件下才会触发的断点，使得调试过程更为高效，这样不会为每次单步都需要手动移到特定条件。
4. **动态分析**：可以运行到特定点时暂停，进行内存调试和性能分析。
在您的情况里，如果遇到了温度相关的截面数据错误，可以通过设置断点暂停到温度设置部分，检查在这个阶段的变量值和文件路径。在使用调试工具中的断点，可以更精确地发现问题的根本原因。
如果您希望进一步了解如何在特定IDE中使用断点，我可以提供针对性建议。
User: with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 在您的代码中，以下这段代码在获取核反应堆模拟结果中的有效增殖因子（k_eff）：
```python
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
```
这段代码的作用解读如下：
1. **`openmc.StatePoint(result)`**:
- `result` 是 openmc 运行函数 `openmc.run()` 的输出，这通常是包含在您的当前工作目录或指定路径中的“statepoint”文件。该文件包含模拟完成后的粒子计算统计结果。
- `openmc.StatePoint` 用于加载并访问 statepoint 文件中的模拟结果。
2. **使用 `with` 语句**:
- `with` 语句是 Python 中用于进行上下文管理，确保在离开 `with` 块时，资源（这里是打开的 statepoint 文件）自动正确关闭。
- 这样做不仅是良好的资源管理实践，还可以防止打开文件后忘记关闭导致的资源泄露。
3. **`k_effective.nominal_value`**:
- `sp.k_effective` 返回一个包含有效倍增因数组件的对象，包括其名义值、标准偏差等。
- `.nominal_value` 获取该有效倍增因子的均值（nominal value），这是有效度量反应堆临界性方程的标准方法。
通过这个过程，您将模拟的主要目标输出之一——有效倍增因子（k_eff）加载到您的模拟环境中，用于后续状态更新或策略评估。
如果在这个过程中出现了错误，建议确认：
- 运行的模拟产生了有效的 statepoint 文件。
- `result` 变量指向了正确的 statepoint 文件路径。
- openmc 的日志中是否有关于模拟失败的其他原因信息，尤其在核材料设置中。
User: result = openmc.run()返回为空
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
#plt.style.use(plt.style.available)
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 900 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 600 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100)
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
# 创建材料集合并生成 XML
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# [代码创建几何]
# 创建universe和geometry并生成 XML
root_universe = openmc.Universe(cells=[fuel_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
# os.remove('/home/unknowxxs/python/statepoint.50.h5')
result = openmc.run()
# 获取k_eff
with openmc.StatePoint(result) as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=10)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 如果 `result = openmc.run()` 返回为空，可能是因为模拟未能成功运行或者错误发生在运行过程中。以下是一些可能的原因和检查步骤：
1. **检查 OpenMC 输出和日志**：
- 检查 OpenMC 生成的输出文件（例如 `statepoint` 文件）和日志，以确定模拟是否成功完成。
- OpenMC 的日志通常会指出模拟失败的原因，包括几何配置问题、材料定义错误、核数据路径不正确等。
2. **确认文件路径**：
- 确保 `cross_sections.xml` 路径正确，OpenMC 需要它来寻找核数据。如果路径不正确或者文件损坏，OpenMC 将无法加载必要的截面数据。
- 确保工作目录有写入权限，以便 OpenMC 可以生成输出文件。
3. **模拟参数**：
- 检查设置的模拟参数，例如批次数、粒子数等。确认这些参数在资源允许的范围内。此外，如果粒子数过少，可能导致统计误差大，从而影响截面计算。
- 确保 `materials.xml`, `geometry.xml`, 和 `settings.xml` 这些必要的文件已正确生成，并包含适当的数据。
4. **错误处理**：
- 使用 `try-except` 捕获模拟运行中发生的错误，并输出错误消息以帮助定位问题。
- 您当前代码中已包含错误处理块，确保从中获取错误日志以分析具体失败原因。
5. **逐步调试**：
- 使用断点或者增加调试输出，以检查在 `openmc.run()` 调用前后的变量状态。
- 验证所有的模拟步骤是否都已完成，例如材料属性是否已正确应用、几何是否完全定义等。
6. **可能的替代方案**：
- 如果获取不到有效的 `statepoint` 文件，可以在终端直接运行模拟命令，以手动查看控制台输出中出现的具体错误。
如果这些检查不能解决问题，请提供更多日志或错误信息，以便更具体地帮助诊断问题。通过这些信息您将能够获得更清晰的关于模拟中断或失败的原因分析。
User: 模拟过程出错: 'StatePoint' object has no attribute 'k_effective'
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
# 设置核数据路径
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 293.6 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 293.6 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100, boundary_type='reflective') # 增加了边界条件
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
outside_cell = openmc.Cell(region=+cylinder, fill=self.water) # 创建外部包围物
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell, outside_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
openmc.run()
print("\033[91m模拟结果: ")
# 获取k_eff
with openmc.StatePoint("/home/unknowxxs/python/statepoint.100.h5") as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=1)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: `StatePoint` 对象没有 `k_effective` 属性的错误通常与 `OpenMC` 的接口变化有关。在你的代码中，可能正在访问 `StatePoint` 的属性不正确。请检查以下修复建议：
OpenMC 的 `StatePoint` 对象具有 `keff` 属性而不是 `k_effective`。请更新您的代码以正确访问此属性：
```python
with openmc.StatePoint("/home/unknowxxs/python/statepoint.100.h5") as sp:
self.k_eff = sp.keff[0] # 获取 k_eff 的名义值
```
在上面的代码中，我们通过 `sp.keff` 获取一个数组，该数组包含 `k_eff` 的名义值和标准偏差，因此使用 `[0]` 来访问名义值。
确保引用了正确的 `statepoint` 文件，命名和路径要与 `openmc.run()` 创建的文件一致。可以通过以下方式来验证：
1. 模拟结束后，查看生成了哪些 statepoint 文件，以及它们的命名及路径。
2. 确保 `openmc.run()` 没有返回错误，并成功生成期望的输出文件。
如果这些调整和检查仍然不能解决问题，请参考 OpenMC 文档，确认当前使用的 OpenMC 版本对 `StatePoint` 的解析是否有其他API变更。也可以从日志或其他输出查找进一步的错误提示以诊断问题。
User: 模拟结果:
模拟过程出错: can't convert an affine function (<class 'uncertainties.core.AffineScalarFunc'>) to float; use x.nominal_value
Traceback (most recent call last):
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
# 设置核数据路径
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 293.6 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 293.6 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100, boundary_type='reflective') # 增加了边界条件
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
outside_cell = openmc.Cell(region=+cylinder, fill=self.water) # 创建外部包围物
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell, outside_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
openmc.run()
print("\033[91m模拟结果: ")
# 获取k_eff
with openmc.StatePoint("/home/unknowxxs/python/statepoint.100.h5") as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=1)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 在您的 OpenMC 模拟中收到错误消息，说明在尝试获取 k_eff 时遇到了 `AffineScalarFunc` 而不是简单的浮点数。在 OpenMC 中，`k_eff` 是一个带有不确定度的标量，因此它是通过 `uncertainties` 包中的 `AffineScalarFunc` 类型表示的。这意味着它既有一个名义值（nominal value）也有一个标准误差。
要解决此问题，需要确保从 `AffineScalarFunc` 获取名义值（即基本的浮点值），而不是尝试直接将其转换为浮点数。你可以通过以下方式修改代码：
替换这行代码：
```python
self.k_eff = sp.k_effective.nominal_value
```
为：
```python
self.k_eff = sp.keff.nominal_value # 获取 k_eff 的名义值
```
您还可以添加一行代码注释以解释为什么要这样做：
```python
# 提取名义值以用于后续的数值计算，忽略不确定性
```
此外，请确保在模拟开始前已正确指定所有路径，并确保 `statepoint.100.h5` 是 OpenMC 生成的文件。运行 OpenMC 后，确认无异常并输出 `statepoint` 文件含有期望的结果。
如仍遇到问题，确保输出文件确实在指定路径中生成，并且没有被异常中止或写入失败的情况。通过检查 OpenMC 的日志输出以确认模拟过程的顺利进行。
User: /mnt/wslg/distro/home/unknowxxs/miniconda3/lib/python3.12/tkinter/__init__.py:861: UserWarning: Glyph 27493 (\N{CJK UNIFIED IDEOGRAPH-6B65}) missing from font(s) DejaVu Sans.
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
# 设置核数据路径
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 293.6 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 293.6 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100, boundary_type='reflective') # 增加了边界条件
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
outside_cell = openmc.Cell(region=+cylinder, fill=self.water) # 创建外部包围物
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell, outside_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
openmc.run()
print("\033[91m模拟结果: ")
# 获取k_eff
with openmc.StatePoint("/home/unknowxxs/python/statepoint.100.h5") as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=1)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: 从警告信息来看，您遇到的问题是由于缺少某些字体来正确显示中文字符。具体来说，程序试图使用 `DejaVu Sans` 字体来显示字符，但发现字体缺少特定的中文字符（在您的情况下是**步**这个字符）。
在使用 Matplotlib 进行可视化时，默认字体可能不支持所有语言的字符，所以需要为缺失的字符提供合适的字体方案。
以下是常见的解决方案：
1. **安装支持中文的字体**：
您可以在系统中安装一种支持中文的字体，例如 `Noto Sans CJK` 或 `SimSun` (宋体)。在一些Linux发行版上，可以用包管理器安装这些字体。
2. **在Matplotlib中设置字体**：
您可以在代码中指定字体，以确保Matplotlib使用支持中文字符的字体：
```python
import matplotlib.pyplot as plt
from matplotlib import font_manager
# 设置字体
font_path = '/path/to/chinese/font.ttf' # 替换为您的字体文件路径，如 SimSun.ttf
prop = font_manager.FontProperties(fname=font_path)
plt.rcParams['font.family'] = prop.get_name()
```
注：在Windows上，您可以将路径指向 Windows 自带的中文字体路径，如 `C:/Windows/Fonts/simsun.ttc`。
3. **配置 matplotlibrc 文件**：
通过编辑 `matplotlibrc` 文件，您可以为所有Matplotlib绘图设置全局默认字体。查找 `~/.matplotlib/matplotlibrc` 文件，在其中找到 `font.family`，按需设置为支持中文的字体。
4. **动态字体加载**：
如果在某些情形下，您不希望更改全局设置，也可以在创建特定文本的地方动态应用字体。例如用以下方法设置特定文本的字体：
```python
plt.title('标题', fontproperties=prop) # 这里 prop 是使用 FontProperties 创建的
```
按这些步骤设置好后，相应的字符应该可以正常显示而不再出现缺失警告。如有进一步问题，请确认字体路径和字体库已正确加载，并且字体确实包含您所需的字符集。
User: font_path = '/home/unknowxxs/ttf/DejaVuSans.ttf' # 替换为您的字体文件路径，如 SimSun.ttf
prop = font_manager.FontProperties(fname=font_path)
plt.rcParams['font.family'] = prop.get_name()
```python
import os
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import openmc
import matplotlib.pyplot as plt
# 设置核数据路径
openmc.config['cross_sections'] = '/home/unknowxxs/python/data/cross_sections.xml'
class NuclearReactorEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(11) # 控制棒位置，0-10
# 观察空间：温度、功率、k_eff、控制棒位置、燃料燃耗
self.observation_space = gym.spaces.Box(
low=np.array([500, 0, 0.5, 0, 0]),
high=np.array([1500, 3000, 1.5, 1, 100]),
dtype=np.float32
)
self.max_steps = 1000
self.current_step = 0
self._setup_model()
def _setup_model(self):
# 设置材料
# UO2燃料
self.fuel = openmc.Material(name='fuel')
self.fuel.add_nuclide('U235', 0.05)
self.fuel.add_nuclide('U238', 0.95)
self.fuel.add_nuclide('O16', 2.0)
self.fuel.set_density('g/cm3', 10.0)
self.fuel.temperature = 293.6 # K
# 水冷却剂
self.water = openmc.Material(name='water')
self.water.add_nuclide('H1', 2)
self.water.add_nuclide('O16', 1)
self.water.set_density('g/cm3', 1.0)
self.water.temperature = 293.6 # K
# 控制棒 (B4C)
self.control_rod = openmc.Material(name='control_rod')
self.control_rod.add_nuclide('B10', 0.8)
self.control_rod.add_nuclide('B11', 0.2)
self.control_rod.add_nuclide('C12', 1.0)
self.control_rod.set_density('g/cm3', 2.52)
# 创建材料集合
self.materials = openmc.Materials([self.fuel, self.water, self.control_rod])
self.materials.export_to_xml() # 生成 materials.xml
# 创建几何体
# 简化的圆柱形反应堆堆芯
cylinder = openmc.ZCylinder(r=100, boundary_type='reflective') # 增加了边界条件
fuel_region = -cylinder
fuel_cell = openmc.Cell(fill=self.fuel, region=fuel_region)
outside_cell = openmc.Cell(region=+cylinder, fill=self.water) # 创建外部包围物
# 创建universe和geometry
root_universe = openmc.Universe(cells=[fuel_cell, outside_cell])
self.geometry = openmc.Geometry(root_universe)
self.geometry.export_to_xml() # 生成 geometry.xml
# 设置模拟参数并生成 XML
self.settings = openmc.Settings()
self.settings.batches = 100
self.settings.inactive = 10
self.settings.particles = 1000
self.settings.temperature = {'method': 'interpolation'}
self.settings.export_to_xml() # 生成 settings.xml
def init_visualization(self):
"""初始化可视化图表"""
plt.ion() # 开启交互模式
self.fig, self.axs = plt.subplots(2, 2, figsize=(15, 10))
self.fig.suptitle('核反应堆运行状态监测', fontsize=16)
# 初始化数据存储
self.history = {
'temperature': [],
'power': [],
'k_eff': [],
'burnup': [],
'control_rod': [],
'steps': []
}
# 配置子图
self.lines = {}
# 温度和功率图
self.axs[0, 0].set_title('温度和功率变化')
self.axs[0, 0].set_xlabel('步数')
self.lines['temp'], = self.axs[0, 0].plot([], [], 'r-', label='温度 (K)')
self.lines['power'], = self.axs[0, 0].plot([], [], 'b-', label='功率 (MW)')
self.axs[0, 0].legend()
# k_eff图
self.axs[0, 1].set_title('k_eff变化')
self.axs[0, 1].set_xlabel('步数')
self.lines['k_eff'], = self.axs[0, 1].plot([], [], 'g-', label='k_eff')
self.axs[0, 1].axhline(y=1.0, color='r', linestyle='--', label='临界值')
self.axs[0, 1].legend()
# 控制棒位置图
self.axs[1, 0].set_title('控制棒位置')
self.axs[1, 0].set_xlabel('步数')
self.lines['control'], = self.axs[1, 0].plot([], [], 'k-', label='控制棒位置')
self.axs[1, 0].legend()
# 燃耗图
self.axs[1, 1].set_title('燃料燃耗')
self.axs[1, 1].set_xlabel('步数')
self.lines['burnup'], = self.axs[1, 1].plot([], [], 'm-', label='燃耗 (MWd/kgU)')
self.axs[1, 1].legend()
plt.tight_layout()
def update_visualization(self):
"""更新可视化图表"""
# 更新历史数据
self.history['temperature'].append(self.temperature)
self.history['power'].append(self.power)
self.history['k_eff'].append(self.k_eff)
self.history['burnup'].append(self.burnup)
self.history['control_rod'].append(self.control_rod_pos)
self.history['steps'].append(self.current_step)
# 更新图表
steps = self.history['steps']
# 更新温度和功率
self.lines['temp'].set_data(steps, self.history['temperature'])
self.lines['power'].set_data(steps, self.history['power'])
self.axs[0, 0].relim()
self.axs[0, 0].autoscale_view()
# 更新k_eff
self.lines['k_eff'].set_data(steps, self.history['k_eff'])
self.axs[0, 1].relim()
self.axs[0, 1].autoscale_view()
# 更新控制棒位置
self.lines['control'].set_data(steps, self.history['control_rod'])
self.axs[1, 0].relim()
self.axs[1, 0].autoscale_view()
# 更新燃耗
self.lines['burnup'].set_data(steps, self.history['burnup'])
self.axs[1, 1].relim()
self.axs[1, 1].autoscale_view()
plt.draw()
plt.pause(0.01)
def reset(self, seed=None):
super().reset(seed=seed)
self.current_step = 0
self.temperature = 600 # K
self.power = 1000 # MW
self.k_eff = 1.0
self.burnup = 0 # MWd/kgU
self.control_rod_pos = 0.5
# 初始化可视化
self.init_visualization()
return self._get_observation(), {}
def _get_observation(self):
return np.array([
self.temperature,
self.power,
self.k_eff,
self.control_rod_pos,
self.burnup
], dtype=np.float32)
def step(self, action):
self.current_step += 1
# 更新控制棒位置
self.control_rod_pos = action / 10.0
# 运行OpenMC模拟
try:
# 更新材料温度
self.fuel.temperature = self.temperature
# 运行临界计算
openmc.run()
print("\033[91m模拟结果: ")
# 获取k_eff
with openmc.StatePoint("/home/unknowxxs/python/statepoint.100.h5") as sp:
self.k_eff = sp.k_effective.nominal_value
# 更新系统状态
self._update_state()
# 更新可视化
self.update_visualization()
# 计算奖励
reward = self._calculate_reward()
# 检查是否结束
done = self.current_step >= self.max_steps or self._is_unsafe()
except Exception as e:
print(f"模拟过程出错: {e}")
reward = -1000
done = True
return self._get_observation(), reward, done, False, {}
def _update_state(self):
# 基于k_eff更新功率
power_factor = (self.k_eff - 1.0) * 5000
self.power = np.clip(self.power + power_factor, 0, 3000)
# 更新温度
delta_temp = (self.power - 1000) * 0.1
self.temperature = np.clip(self.temperature + delta_temp, 500, 1500)
# 更新燃耗
self.burnup += self.power * 0.001
def _calculate_reward(self):
reward = 0
# 奖励稳定的运行状态
reward -= abs(self.power - 1000) * 0.01 # 目标功率1000MW
reward -= abs(self.k_eff - 1.0) * 100 # 目标k_eff=1
reward -= abs(self.temperature - 900) * 0.1 # 目标温度900K
# 惩罚不安全状态
if self._is_unsafe():
reward -= 1000
return reward
def _is_unsafe(self):
return (self.temperature > 1400 or
self.power > 2500 or
self.k_eff > 1.2)
def train_agent():
env = make_vec_env(NuclearReactorEnv, n_envs=4)
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=0.0003,
n_steps=2048,
batch_size=64,
n_epochs=10
)
model.learn(total_timesteps=1)
model.save("ppo_nuclear_reactor")
return model
def evaluate_agent(model, num_episodes=5):
env = NuclearReactorEnv()
for episode in range(num_episodes):
obs, _ = env.reset()
episode_reward = 0
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _, _ = env.step(action)
episode_reward += reward
# 打印当前状态
print(f"步骤 {env.current_step}:")
print(f"温度: {env.temperature:.1f}K")
print(f"功率: {env.power:.1f}MW")
print(f"k_eff: {env.k_eff:.3f}")
print(f"控制棒位置: {env.control_rod_pos:.2f}")
print(f"燃耗: {env.burnup:.2f}MWd/kgU")
print("------------------------")
plt.savefig(f'episode_{episode+1}_results.png')
print(f"Episode {episode + 1}: 总奖励 = {episode_reward:.2f}\n")
plt.close('all')
model = train_agent()
print("\n评估核反应堆控制性能：")
evaluate_agent(model)
```
Assistant: