« 高周波・RFニュース 2026年8月19日 IEEE Antennas and Propagation Magazine8月号発行、東京農大とロームがテラヘルツ用薄型レンズ開発、I-PEXの細線同軸ケーブル解説ウェビナー、NokiaとMediaTekが3GPPベースGNSS RTKでデシメートル精度を達成 | トップページ | 高周波・RFニュース 2026年8月20日 Pythonの高周波ライブラリscikit-rfがv2.1.0でTouchstone v2.1に対応、Mini-CircutsがSバンドアップコンバータ製作解説、6Gフォーラムは10月6日開催、Google Pixel 11分解動画で5Gミリ波AiP確認、LitePointとSTMがUWBでコラボ »

2026年8月19日 (水)

高周波エンジニアのためのAI・機械学習入門(GPU編42)強化学習でLCバンドパスフィルタの素子値を最適化する。今回はPythonで報酬とgymnasium環境を作る。

前回は特性評価するライブラリを作った。今回は報酬とgymnasium環境を作る。
報酬はこんな感じ。指数関数を使って素子値を更新するのがポイント(Copilot Chatに考えてもらった)。


import numpy as np

def constraint_penalty(value, lower=None, upper=None, weight=1.0, scale=1.0):
    """
    value が lower 以上、upper 以下に入るようにペナルティを返す。
    制約を満たしていれば0。
    """
    penalty = 0.0

    if lower is not None and value < lower:
        penalty += ((lower - value) / scale) ** 2

    if upper is not None and value > upper:
        penalty += ((value - upper) / scale) ** 2

    return weight * penalty


def score_bpf_analysis(
    analysis,
    targets,
    objective="min_insertion_loss",
):
    """
    analysis辞書から報酬を計算する。

    targets の例:
    targets = {
        "insertion_loss_at_f0_dB": {"upper": 2.0, "weight": 10.0, "scale": 1.0},
        "passband_ripple_dB_in_design_band": {"upper": 0.5, "weight": 5.0, "scale": 0.2},
        "min_return_loss_dB_in_design_band": {"lower": 15.0, "weight": 5.0, "scale": 5.0},
        "measured_fbw": {"lower": 0.09, "upper": 0.11, "weight": 3.0, "scale": 0.01},
    }

    reward は大きいほど良い。
    """

    penalty = 0.0

    for key, spec in targets.items():
        if key not in analysis:
            raise KeyError(f"{key} is not found in analysis.")

        value = analysis[key]

        if not np.isfinite(value):
            penalty += 1e6
            continue

        lower = spec.get("lower", None)
        upper = spec.get("upper", None)
        weight = spec.get("weight", 1.0)
        scale = spec.get("scale", 1.0)

        penalty += constraint_penalty(
            value=value,
            lower=lower,
            upper=upper,
            weight=weight,
            scale=scale,
        )

    # 目的関数
    if objective == "min_insertion_loss":
        obj = analysis["insertion_loss_at_f0_dB"]
    elif objective == "min_worst_insertion_loss":
        obj = analysis["worst_insertion_loss_dB_in_design_band"]
    elif objective == "max_return_loss":
        obj = -analysis["min_return_loss_dB_in_design_band"]
    elif objective == "min_ripple":
        obj = analysis["passband_ripple_dB_in_design_band"]
    else:
        obj = 0.0

    if not np.isfinite(obj):
        obj = 1e6

    reward = -(obj + penalty)

    return reward, {
        "objective_value": obj,
        "penalty": penalty,
        "reward": reward,
    }

def apply_lc_log_multipliers(
    LC_base,
    x,
    active_rows=None,
):
    """
    LC_base の L,C に対して exp(x) の倍率をかける。

    x の長さは 2 * n。
    x = [x_L1, x_C1, x_L2, x_C2, ..., x_Ln, x_Cn]

    active_rows:
        最適化対象の行番号。
        Noneなら 1〜n 行を対象にする。
    """
    LC_new = np.array(LC_base, dtype=float, copy=True)
    n = LC_base.shape[0] - 2

    if active_rows is None:
        active_rows = list(range(1, n + 1))

    expected_len = 2 * len(active_rows)

    if len(x) != expected_len:
        raise ValueError(f"x length must be {expected_len}.")

    idx = 0
    for row in active_rows:
        LC_new[row, 0] = LC_base[row, 0] * np.exp(x[idx])
        LC_new[row, 1] = LC_base[row, 1] * np.exp(x[idx + 1])
        idx += 2

    return LC_new

gymnasium環境はこんな感じ。


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])

これを使っていろいろな強化学習のアルゴリズムを試してみよう。

« 高周波・RFニュース 2026年8月19日 IEEE Antennas and Propagation Magazine8月号発行、東京農大とロームがテラヘルツ用薄型レンズ開発、I-PEXの細線同軸ケーブル解説ウェビナー、NokiaとMediaTekが3GPPベースGNSS RTKでデシメートル精度を達成 | トップページ | 高周波・RFニュース 2026年8月20日 Pythonの高周波ライブラリscikit-rfがv2.1.0でTouchstone v2.1に対応、Mini-CircutsがSバンドアップコンバータ製作解説、6Gフォーラムは10月6日開催、Google Pixel 11分解動画で5Gミリ波AiP確認、LitePointとSTMがUWBでコラボ »

パソコン・インターネット」カテゴリの記事

学問・資格」カテゴリの記事

日記・コラム・つぶやき」カテゴリの記事

コメント

コメントを書く

(ウェブ上には掲載しません)

« 高周波・RFニュース 2026年8月19日 IEEE Antennas and Propagation Magazine8月号発行、東京農大とロームがテラヘルツ用薄型レンズ開発、I-PEXの細線同軸ケーブル解説ウェビナー、NokiaとMediaTekが3GPPベースGNSS RTKでデシメートル精度を達成 | トップページ | 高周波・RFニュース 2026年8月20日 Pythonの高周波ライブラリscikit-rfがv2.1.0でTouchstone v2.1に対応、Mini-CircutsがSバンドアップコンバータ製作解説、6Gフォーラムは10月6日開催、Google Pixel 11分解動画で5Gミリ波AiP確認、LitePointとSTMがUWBでコラボ »

最近の記事

2026年9月
    1 2 3 4 5
6 7 8 9 10 11 12
13 14 15 16 17 18 19
20 21 22 23 24 25 26
27 28 29 30      

最近のコメント

無料ブログはココログ
フォト