import numpy as np
import gymnasium as gym
from gymnasium import spaces
import bpf
import bpf_score
class LCBPFOptimizationEnv(gym.Env):
"""
LCラダーBPFのL,C値を強化学習で最適化するGymnasium環境。
事前に synthesize_lc_bpf() で作ったLC_baseを基準に、
L,Cを exp(x) 倍して探索する。
"""
metadata = {"render_modes": []}
def __init__(
self,
n,
LC_base,
Q_values,
f1,
f2,
f0,
fstart,
fstop,
points,
fq=None,
z0=50.0,
first_element="series",
targets=None,
objective="min_insertion_loss",
x_limit=0.7,
action_step=0.08,
max_steps=40,
success_reward=100.0,
name="RL_LC_BPF",
active_rows=None,
):
super().__init__()
self.n = n
self.LC_base = np.array(LC_base, dtype=float, copy=True)
self.Q_values = np.array(Q_values, dtype=object, copy=True)
self.f1 = f1
self.f2 = f2
self.f0 = f0
self.fstart = fstart
self.fstop = fstop
self.points = points
self.fq = f0 if fq is None else fq
self.z0 = z0
self.first_element = first_element
self.targets = targets or {}
self.objective = objective
self.x_limit = float(x_limit)
self.action_step = float(action_step)
self.max_steps = int(max_steps)
self.success_reward = float(success_reward)
self.name = name
if active_rows is None:
self.active_rows = list(range(1, n + 1))
else:
self.active_rows = list(active_rows)
self.dim = 2 * len(self.active_rows)
# 状態: log倍率 x, Sパラメータ評価値, および目標との誤差
self.target_keys = sorted(list(self.targets.keys()))
obs_dim = self.dim + 4 + len(self.target_keys)
low = np.zeros(obs_dim, dtype=np.float32)
high = np.zeros(obs_dim, dtype=np.float32)
low[:self.dim] = -self.x_limit
high[:self.dim] = self.x_limit
low[self.dim:] = 0.0
high[self.dim:] = 10.0
self.observation_space = spaces.Box(
low=low,
high=high,
dtype=np.float32,
)
# 行動: xへの増分。実際の増分は action_step を掛ける
self.action_space = spaces.Box(
low=-1.0,
high=1.0,
shape=(self.dim,),
dtype=np.float32,
)
self.x = None
self.step_count = 0
self.best_reward = -np.inf
self.best_x = None
self.best_analysis = None
self.best_LC = None
def reset(self, seed=None, options=None):
super().reset(seed=seed)
self.x = np.zeros(self.dim, dtype=np.float32)
self.step_count = 0
analysis, LC_current = self._evaluate_current()
reward, details = bpf_score.score_bpf_analysis(
analysis,
targets=self.targets,
objective=self.objective,
)
self.best_reward = reward
self.best_x = self.x.copy()
self.best_analysis = analysis
self.best_LC = LC_current.copy()
obs = self._get_obs(self.x, analysis)
info = {
"analysis": analysis,
"LC_elements": LC_current,
"reward_details": details,
"best_reward": self.best_reward,
}
return obs, info
def step(self, action):
self.step_count += 1
action = np.asarray(action, dtype=np.float32)
action = np.clip(action, -1.0, 1.0)
self.x = self.x + self.action_step * action
self.x = np.clip(self.x, -self.x_limit, self.x_limit)
analysis, LC_current = self._evaluate_current()
reward, details = bpf_score.score_bpf_analysis(
analysis,
targets=self.targets,
objective=self.objective,
)
if reward > self.best_reward:
self.best_reward = reward
self.best_x = self.x.copy()
self.best_analysis = analysis
self.best_LC = LC_current.copy()
success = self._check_success(analysis)
if success:
reward += 10.0 # 制約を満たしているステップではボーナス
terminated = False
truncated = bool(self.step_count >= self.max_steps)
obs = self._get_obs(self.x, analysis)
info = {
"analysis": analysis,
"LC_elements": LC_current,
"reward_details": details,
"success": success,
"best_reward": self.best_reward,
"best_analysis": self.best_analysis,
"best_LC_elements": self.best_LC,
}
return obs, reward, terminated, truncated, info
def _evaluate_current(self):
LC_current = bpf_score.apply_lc_log_multipliers(
self.LC_base,
self.x,
active_rows=self.active_rows,
)
ntwk = bpf.lossy_bpf(
n=self.n,
LC_elements=LC_current,
Q_values=self.Q_values,
fq=self.fq,
fstart=self.fstart,
fstop=self.fstop,
points=self.points,
z0=self.z0,
first_element=self.first_element,
name=self.name,
)
analysis = bpf.evaluate_bpf(
ntwk,
f1=self.f1,
f2=self.f2,
f0=self.f0,
)
return analysis, LC_current
def _check_success(self, analysis):
"""
全制約を満たせば成功。
"""
for key, spec in self.targets.items():
value = analysis[key]
if not np.isfinite(value):
return False
lower = spec.get("lower", None)
upper = spec.get("upper", None)
if lower is not None and value < lower:
return False
if upper is not None and value > upper:
return False
return True
def get_best_result(self):
return {
"best_reward": self.best_reward,
"best_x": self.best_x,
"best_LC_elements": self.best_LC,
"best_analysis": self.best_analysis,
}
def _get_obs(self, x, analysis):
obs_x = np.array(x, dtype=np.float32)
il = analysis.get("insertion_loss_at_f0_dB", 20.0)
rl = analysis.get("min_return_loss_dB_in_design_band", 0.0)
ripple = analysis.get("passband_ripple_dB_in_design_band", 10.0)
fbw = analysis.get("measured_fbw", 0.0)
if not np.isfinite(il): il = 20.0
if not np.isfinite(rl): rl = 0.0
if not np.isfinite(ripple): ripple = 10.0
if not np.isfinite(fbw): fbw = 0.0
metrics = np.array([
np.clip(il, 0.0, 20.0) / 20.0,
np.clip(rl, 0.0, 40.0) / 40.0,
np.clip(ripple, 0.0, 10.0) / 10.0,
np.clip(fbw, 0.0, 0.5) / 0.5
], dtype=np.float32)
errors = []
for key in self.target_keys:
spec = self.targets[key]
val = analysis.get(key, np.nan)
if not np.isfinite(val):
errors.append(10.0)
continue
lower = spec.get("lower", None)
upper = spec.get("upper", None)
scale = spec.get("scale", 1.0)
err = 0.0
if lower is not None and val < lower:
err += (lower - val) / scale
if upper is not None and val > upper:
err += (val - upper) / scale
errors.append(np.clip(err, 0.0, 10.0))
obs_errors = np.array(errors, dtype=np.float32)
return np.concatenate([obs_x, metrics, obs_errors])
最近のコメント