Python 驱动的 ADS 自动化仿真框架与 API 实战指南

实现时需要掌握的 ADS Python API：

使用事务保证原子性（推荐）：

from keysight.ads.de.db import Transaction
with Transaction(design) as transaction:
    # 在此块内进行所有参数修改
    inst1.vars.update({...})
    inst2.parameters["C"].value = "10 pF"
    # 提交事务，使修改生效
    transaction.commit()

刷新原理图标注（更新后建议调用）：

inst.update_item_annotation()

更新元件参数（用于电阻、电容、信号源等具体元件）：

inst.parameters["参数名"].value = "值（字符串，可带单位）"
# 例如：inst.parameters["R"].value = "100 Ohm"
# 例如：inst.parameters["File"].value = '"E:/data/signal.txt"' # 文件路径需额外引号

更新 VAR 控件中的变量（用于 VAR / VAR2 等变量定义控件）：

inst.vars.update({"变量名 1": "值（字符串）", "变量名 2": "值（字符串）",})

获取元件实例：

inst = design.get_instance("VAR1") # 按原理图中的 Instance Name 获取

注意事项：

• 所有参数值必须转换为字符串格式，ADS 内部会自行解析单位
• 文件路径类参数（如 DAC1 的 File 参数）需要用双引号包裹，即 "'path'"
• 若更新多个元件，建议包裹在 Transaction 中，确保要么全部成功、要么全部回滚
• 更新后调用 update_item_annotation() 可让原理图界面同步显示新值（非必需但推荐）

AutoSimulator (核心控制器)

这是自动化仿真的主引擎（Driver）。它不关心电路的具体细节，只负责流程控制：加载数据 -> 设置环境 -> 循环仿真 -> 提取结果 -> 保存文件。

__init__(self, parameter_updater: ParameterUpdater)

构造函数，负责将用户实现的参数更新器'注册'到控制器中。

内部行为：将 parameter_updater 保存为成员变量，同时初始化 self.design = None（后续由 setup_ads_environment 填充）。

load_parameter_samples(self, filepath: str) -> pd.DataFrame

从文件加载参数样本数据，统一转换为 Pandas DataFrame 供后续遍历使用。

支持的输入格式：

异常情况：

• FileNotFoundError：文件路径不存在
• ValueError：JSON 内容既非 dict 也非 list，或扩展名不在上述列表中

注意：虽然加载支持多种格式，但目前结果保存仅实现了 JSON 格式。如需将结果回写到 CSV/Excel，用户可继承 AutoSimulator 并重写保存逻辑。

setup_ads_environment(self, workspace_path: str, library_name: str, design_name: str) -> None

初始化 ADS 工作环境，完成工作空间 -> 库 -> 设计的三级打开流程。

内部流程：

1. utils.get_workspace(workspace_path) —— 打开/连接工作空间
2. utils.get_library(workspace, library_name) —— 加载指定库
3. utils.get_design(library, design_name, ViewType.SCHEMATIC) —— 打开原理图视图

成功后 self.design 将指向该设计对象，后续的参数更新和仿真都基于此对象进行。

异常情况：路径错误、库/设计不存在时抛出 RuntimeError。

extract_simulation_results(self, results: dataset.Dataset, target_variables: List[str], debug: bool = True) -> Dict[str, Any]

从仿真结果数据集中智能提取用户指定的目标变量，是本工具库最核心的数据处理函数。

提取策略（不依赖硬编码的块名）：

1. 遍历 results.varblock_names 获取所有 VariableBlock
2. 对每个块，通过 varblock.ivars / varblock.dvars 获取其独立变量和依赖变量名列表
3. 若目标变量出现在 dvars 中，调用 varblock.to_dataframe(dvar_names=[target_var]) 精准提取该列
4. 根据数据长度自动判断类型：
   • 长度 = 1 → 标量，转为 Python float
   • 长度 > 1 → 向量，转为 Python list

返回值：Dict[str, Any]，格式为 {变量名：值或列表}。未找到的变量对应值为 None。

run_simulation_and_extract_results(self, target_variables: List[str], debug: bool = True) -> Dict[str, Any]

执行单次仿真并提取结果的快捷方法，内部封装了仿真管理器的创建与结果获取。

内部流程：

1. 创建 SimulationManager(self.design)
2. 调用 sim_manager.run_design_simulation() 执行仿真
3. 使用上下文管理器 with sim_manager.get_simulation_results() as results 获取结果
4. 调用 extract_simulation_results(results, target_variables, debug) 提取数据

异常处理：若仿真过程出错（如不收敛、License 问题），捕获异常并返回 {var: None for var in target_variables}，不会中断程序。

run_batch_simulation(...)

批量仿真的主入口，串联整个自动化流程。大多数情况下用户只需调用此方法。

执行流程：

1. 加载数据：调用 load_parameter_samples(data_filepath) 读取参数表
2. 限制样本：若设置了 max_samples，取前 N 行
3. 初始化环境：调用 setup_ads_environment(...) 打开 ADS 设计
4. 循环处理每个样本：
   • 调用 parameter_updater.update_parameters(design, row) 更新电路参数
   • 删除旧仿真数据：移除 {cell_path}/data/{design_name}.ds，防止读到历史缓存
   • 调用 run_simulation_and_extract_results(target_variables) 执行仿真并提取结果
   • 若提供了 result_processor，对结果进行后处理
   • 将结果追加到列表
5. 保存结果：若 save_to_original=True，调用 _save_results_to_original_file 合并写回
6. 打印统计：调用 _print_simulation_statistics 输出成功/失败计数与数值范围

返回值：带有 simulation_results 列的 DataFrame。

_save_results_to_original_file(...)

将当前批次的仿真结果与原文件中已有的 simulation_results 字段合并后写回，支持增量覆盖（新结果覆盖同名旧结果，旧结果中不重名的字段保留）。

内部逻辑：

1. 遍历 simulation_results 列表，对每条记录中的 goal_variables 调用 _process_value_for_json 转换
2. 遍历原 DataFrame，取出每行的 simulation_results（若不存在则初始化为空字典）
3. 使用 dict.update() 将新结果合并进去
4. 最终以 json.dump(df.to_dict(orient="records"), ...) 写回文件

限制：目前仅支持 JSON 格式输出。如需 CSV/Excel 输出，可重写此方法或在调用后手动导出。

_process_value_for_json(self, value) -> Any

递归处理各种数据类型，确保最终结果可被 json.dump 序列化。

_print_simulation_statistics(self, df, target_variables) -> None

在批量仿真结束后打印统计摘要，便于快速评估整体质量。

输出内容包括：

• 总样本数 / 成功数 / 失败数
• 对于每个 target_variable，若为数值型则显示最小值、最大值、均值

实战示例：开发一个自定义仿真器

本节以一个射频放大器增益仿真为例，完整演示从数据准备到结果提取的全流程。

1.3.1 输入数据格式

首先准备参数样本文件（JSON 格式），每条记录代表一组待仿真的电路参数：

[{"sample_id":1,"Vbias":3.3,"R_source":50,"C_coupling":100,"L_match":2.2},{"sample_id":2,"Vbias":3.0,"R_source":75,"C_coupling":150,"L_match":1.8}]

字段说明：

• sample_id：样本编号，用于日志输出和结果追踪（可选，若不提供则自动使用行索引）
• 其余字段：与原理图中需要更新的变量/参数一一对应

1.3.2 实现参数更新器

根据你的 ADS 原理图结构，继承 ParameterUpdater 并实现 update_parameters 方法：

""" amplifier_simulator.py 射频放大器批量仿真脚本 """
import sys
from pathlib import Path

# 将项目根目录加入 Python 路径（确保能 import 该工具库）
project_root = Path(__file__).parent.parent
sys.path.append(str(project_root))

import pandas as pd
from utils.auto_simulator import AutoSimulator, ParameterUpdater
from keysight.ads.de import db_uu as db
from keysight.ads.de.db import Transaction

class AmplifierUpdater(ParameterUpdater):
    """
    放大器参数更新器
    假设原理图中包含：
    - VAR1 控件：定义 Vbias（偏置电压）、R_source（源阻抗）
    - C1 元件：耦合电容
    - L1 元件：匹配电感
    """
    def update_parameters(self, design: db.Design, param_row: pd.Series) -> None:
        """
        将 param_row 中的参数值写入 ADS 原理图
        Args:
            design: ADS 设计对象
            param_row: 当前样本的参数（Pandas Series）
        """
        # 使用 Transaction 确保所有修改原子性提交
        with Transaction(design) as transaction:
            # ========== 1. 更新 VAR 控件中的变量 ==========
            # 获取名为 "VAR1" 的变量控件实例
            inst_var = design.get_instance("VAR1")
            # 批量更新变量值（必须转为字符串）
            inst_var.vars.update({
                "Vbias": str(param_row["Vbias"]),
                "R_source": str(param_row["R_source"]),
            })
            # 刷新原理图标注（可选，但推荐）
            inst_var.update_item_annotation()

            # ========== 2. 更新具体元件的参数 ==========
            # 更新耦合电容 C1 的电容值
            inst_c1 = design.get_instance("C1")
            inst_c1.parameters["C"].value = f"{param_row['C_coupling']} pF"
            inst_c1.update_item_annotation()

            # 更新匹配电感 L1 的电感值
            inst_l1 = design.get_instance("L1")
            inst_l1.parameters["L"].value = f"{param_row['L_match']} nH"
            inst_l1.update_item_annotation()

            # ========== 3. 提交事务 ==========
            transaction.commit()

代码要点：

1. Transaction 上下文管理器：将所有修改包裹在事务中，调用 commit() 后才真正生效
2. VAR 控件 vs 元件参数：VAR 控件用 .vars.update()，普通元件用 .parameters["名称"].value
3. 单位处理：电容用 pF、电感用 nH，ADS 会自动解析
4. 字符串转换：所有值必须是字符串，使用 str() 或 f-string

1.3.3 编写主控制脚本

def main():
    """主函数：配置参数并启动批量仿真"""
    # ========== 配置区 ==========
    config = {
        # 输入：参数样本文件路径
        "data_filepath": "E:/project/data/amplifier_samples.json",
        # ADS 环境配置
        "workspace_path": "E:/ADS2026_wrk/MyAmplifier_wrk",
        "library_name": "MyAmplifier_lib",
        "design_name": "Amp_Gain_Schematic", # 原理图名称（不含 .dsn 后缀）
        # 输出：要提取的仿真结果变量
        # 这些变量名需与 ADS 仿真结果中的变量名完全一致
        "target_variables": [
            "S[2,1]", # S21 参数（增益）
            "S[1,1]", # S11 参数（输入反射）
            "Gain_dB", # 测量方程计算的增益（dB）
            "NF_dB", # 噪声系数
        ],
        # 可选：限制处理样本数（调试时使用）
        "max_samples": None, # None 表示处理全部，设为 5 则只跑前 5 个
    }
    print("=" * 60)
    print("射频放大器批量仿真")
    print("=" * 60)
    print(f"参数文件：{config['data_filepath']}")
    print(f"目标变量：{config['target_variables']}")
    print()

    # ========== 执行仿真 ==========
    try:
        # 1. 创建参数更新器实例
        updater = AmplifierUpdater()
        # 2. 创建自动仿真器，注入更新器
        simulator = AutoSimulator(updater)
        # 3. 运行批量仿真
        final_df = simulator.run_batch_simulation(
            data_filepath=config["data_filepath"],
            workspace_path=config["workspace_path"],
            library_name=config["library_name"],
            design_name=config["design_name"],
            target_variables=config["target_variables"],
            max_samples=config["max_samples"],
            save_to_original=True, # 结果追加写回原 JSON 文件
        )
        print()
        print("=" * 60)
        print(f"仿真完成！共处理 {len(final_df)} 个样本")
        print(f"结果已保存到：{config['data_filepath']}")
        print("=" * 60)
    except Exception as e:
        print(f"\n❌ 仿真过程发生错误：{e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()

1.3.4 输出结果格式

仿真完成后，原始 JSON 文件会被追加 simulation_results 字段：

[
  {
    "sample_id": 1,
    "Vbias": 3.3,
    "R_source": 50,
    "C_coupling": 100,
    "L_match": 2.2,
    "simulation_results": {
      "S[2,1]": {"real": 2.51, "imag": 1.33, "magnitude": 2.84, "phase_deg": 27.9},
      "S[1,1]": {"real": -0.12, "imag": 0.05, "magnitude": 0.13, "phase_deg": 157.4},
      "Gain_dB": 9.07,
      "NF_dB": 2.3
    }
  },
  {
    "sample_id": 2,
    "Vbias": 3.0,
    "R_source": 75,
    "C_coupling": 150,
    "L_match": 1.8,
    "simulation_results": {
      "S[2,1]": {...},
      "S[1,1]": {...},
      "Gain_dB": 8.45,
      "NF_dB": null
    }
  }
]

结果字段说明：

• 复数类型（如 S 参数）：自动拆分为 real、imag、magnitude、phase_deg
• 标量类型（如 Gain_dB）：直接存储数值
• 提取失败：对应字段值为 null

1.3.5 进阶：自定义结果后处理

若需要对原始提取结果进行计算（如将复数 S21 转为 dB），可传入 result_processor 回调：

import numpy as np

def process_results(raw_results: dict) -> dict:
    """
    自定义结果后处理函数
    Args:
        raw_results: extract_simulation_results 返回的原始字典
    Returns:
        处理后的结果字典
    """
    processed = raw_results.copy()
    # 示例：将复数 S21 转为 dB
    s21 = raw_results.get("S[2,1]")
    if s21 is not None and isinstance(s21, complex):
        processed["S21_dB"] = 20 * np.log10(abs(s21))
    return processed

# 在 run_batch_simulation 中使用
final_df = simulator.run_batch_simulation(..., result_processor=process_results)

为什么推荐使用此模式？

1. 稳健性：内部统一捕获仿真异常，单个样本失败不会拖垮全流程。
2. 断点续传：结果实时写回原始文件，结合简单的跳过逻辑即可实现增量仿真。
3. 结果对齐：自动将输入参数与输出结果一一对应，便于后续数据分析与模型训练。

辅助工具库 (utils)

先前在自动化工具库中也用到了一些二次封装 ADS API 的类和方法，这里一并给出供参考。

import os
import keysight.ads.dataset as dataset
from keysight.ads import de
from keysight.ads.de import db_uu as db
from enum import Enum
from typing import Literal, Union
from keysight.edatoolbox import ads
from keysight.ads.de.experimental.text_maker import TextMaker

# 定义枚举
class ViewType(Enum):
    SCHEMATIC = "schematic"
    SYMBOL = "symbol"
    LAYOUT = "layout"

    def __str__(self) -> str:
        return self.value

# 定义 Literal 类型（用于纯字符串输入）
ViewNameLiteral = Literal["schematic", "symbol", "layout"]
# 组合类型
ViewNameType = Union[ViewType, ViewNameLiteral]

def design_exists(library: de.Library, name: str, view_type: ViewNameType = ViewType.SCHEMATIC) -> bool:
    design_ref = f"{library.name}:{name}:{view_type.value}"
    return de.design_exists(design_ref)

def get_workspace(workspace_path: str) -> de.Workspace:
    # Ensure there isn't already a workspace open
    if de.workspace_is_open():
        de.close_workspace()
    # Cannot create a workspace if the directory already exists
    if os.path.exists(workspace_path):
        print(f"Workspace directory already exists: {workspace_path}")
        de.open_workspace(workspace_path)
        return de.active_workspace()
    
    # Create the workspace
    workspace = de.create_workspace(workspace_path)
    print(f"Workspace created: {workspace_path}")
    # Open the workspace
    workspace.open()
    # Return the open workspace and close when it finished
    return workspace

def get_library(workspace: de.Workspace, name: str) -> de.Library:
    # assert workspace.path is not None
    # Libraries can only be added to an open workspace
    assert workspace.is_open
    # We'll create a library in the directory of the workspace
    library_name = name
    library_path = workspace.path / library_name
    if os.path.exists(library_path):
        if de.library_is_open(library_name):
            de.close_library(library_name)
        print(f"Library directory already exists: {library_path}")
        return workspace.open_library(library_name, library_path, de.LibraryMode.SHARED)
    
    # Create the library
    de.create_new_library(library_name, library_path)
    print(f"Library created: {library_path}")
    # And add it to the workspace (update lib.defs)
    workspace.add_library(library_name, library_path, de.LibraryMode.SHARED)
    lib = workspace.open_library(library_name, library_path, de.LibraryMode.SHARED)
    return lib

def get_design(
    library: de.Library,
    name: str,
    view_type: ViewNameType = ViewType.SCHEMATIC,
    mode: db.DesignMode = db.DesignMode.APPEND,
) -> db.Design:
    """创建设计，如果已存在则打开现有设计"""
    design_ref = f"{library.name}:{name}:{view_type.value}"
    if design_exists(library, name, view_type):
        return db.open_design(design_ref, mode=mode)
    else:
        if view_type == ViewType.SCHEMATIC:
            design = db.create_schematic(design_ref)
        elif view_type == ViewType.SYMBOL:
            design = db.create_symbol(design_ref)
        elif view_type == ViewType.LAYOUT:
            design = db.create_layout(design_ref)
        else:
            raise ValueError(f"Invalid view type: {view_type}")
        print(f"Design created: {design_ref}")
        return design

def add_measeqn(
    design: db.Design,
    eq_name: str,
    eq_list: list[str],
    origin: tuple[float, float] = (0, 0),
    angle: float = 0,
) -> None:
    # add MeasEqn to the schmatic
    measeqn = design.add_instance(("ads_simulation", "MeasEqn", "symbol"), origin, name=eq_name, angle=angle)
    # change first existing equation
    measeqn.parameters["Meas"].value = [eq_list[0]]
    # add new equations with rest of equation list
    for i in range(len(eq_list) - 1):
        measeqn.parameters["Meas"].repeats.append(
            db.ParamItemString("Meas", "SingleTextLine", eq_list[i + 1])
        )
    measeqn.update_item_annotation()

class ManagedCircuitSimulator:
    """带有完整资源管理的电路仿真器"""
    def __init__(self, hpeesof_dir=None):
        self.simulator = ads.CircuitSimulator(hpeesof_dir)

    def run_netlist(self, netlist: str, output_dir: str, **kwargs) -> None:
        working_dir = kwargs.get("working_dir", output_dir)
        try:
            return self.simulator.run_netlist(
                netlist=netlist,
                output_dir=output_dir,
                **kwargs
            )
        finally:
            # 清理所有可能的临时文件
            self._cleanup_all_temp_files(working_dir)

    def _cleanup_all_temp_files(self, directory) -> None:
        """Clean up all ADS temporary files in directory"""
        import glob
        patterns = ["circ*.ckt", "circ*.out", "*.tmp"]
        for pattern in patterns:
            for temp_file in glob.glob(os.path.join(directory, pattern)):
                try:
                    os.remove(temp_file)
                    print(f"Cleaned temporary file: {os.path.basename(temp_file)}")
                except OSError:
                    pass

class SimulationManager:
    """Complete simulation manager with temporary file cleanup and path safety handling"""
    def __init__(self, design: db.Design):
        self.design = design
        self.output_dir = os.path.join(self.design.cell.path, "data")
        self.netlist = None
        self.result = None
        self.result_path = os.path.join(self.output_dir, f"{self.design.cell_name}.ds")
        self.managed_simulator = ManagedCircuitSimulator()
        self.data_file_is_exist = os.path.exists(self.result_path)
        if self.data_file_is_exist:
            self.result = dataset.open(self.result_path)

    def validate_design_name(self) -> bool:
        """Validate design name, check for uppercase letters that may cause path encoding issues"""
        cell_name = self.design.cell_name
        has_uppercase = any(c.isupper() for c in cell_name)
        if has_uppercase:
            print(f"Warning: Cell name '{cell_name}' contains uppercase letters")
            print(" ADS maps uppercase letters to %letter format, which may cause path issues")
            print(" Recommend using lowercase letters for Cell names")
        return False

    def run_design_simulation(
        self,
        output_dir: Union[str, None] = None,
        output_file: str = None,
        validate_name: bool = True,
        **sim_kwargs,
    ) -> None:
        """Run design simulation with automatic temporary file management"""
        if output_dir is None:
            output_dir = self.output_dir
        os.makedirs(output_dir, exist_ok=True)
        print(f"=== Running simulation: {self.design.design_name} ===")
        print(f"Output directory: {output_dir}")
        # Validate design name
        if validate_name:
            self.validate_design_name()
        try:
            # Generate netlist
            print("Generating netlist...")
            self.netlist = self.design.generate_netlist()
            if not self.netlist or len(self.netlist.strip()) == 0:
                raise RuntimeError("Generated netlist is empty")
            print(f"Netlist generated successfully, length: {len(self.netlist)} characters")
            # Run simulation using managed simulator with auto cleanup
            print("Running simulation...")
            self.managed_simulator.run_netlist(
                netlist=self.netlist,
                output_dir=output_dir,
                output_file=output_file,
                **sim_kwargs,
            )
            print("Simulation executed successfully")
            # Record result path, do not open dataset immediately
            # Dataset should be used through context manager
            print(f"Simulation completed, result file: {self.result_path}")
            self.data_file_is_exist = True
        except Exception as e:
            print(f"Simulation failed: {e}")
            raise

    def get_simulation_results(self) -> dataset.Dataset:
        """Get simulation results
        Returns a context manager, should be used like this:
        with sim_mgr.get_simulation_results() as results:
            print(results.varblock_names) # Process results...
            # Dataset will be automatically closed
        """
        if not self.data_file_is_exist:
            raise RuntimeError("Simulation has not been run or failed, cannot get results")
        return dataset.open(self.result_path)