基于 Python 的 ADS 自动化仿真框架与 API 使用指南

1. 自动化数据提取工具库详解

为了简化 ADS 仿真程控的开发难度，我提供了一个通用的自动化工具库 auto_simulator.py。该工具库封装了从环境配置、参数更新、仿真运行到结果提取的全流程，使得用户只需关注“如何将参数应用到电路”这一核心逻辑。

""" 通用自动化仿真模块 提供通用的ADS仿真自动化框架，支持批量参数扫描和结果提取。 用户只需实现参数更新接口即可使用。 """import pandas as pd import numpy as np import json import os from pathlib import Path from abc import ABC, abstractmethod from typing import Dict, List, Any, Optional, Callable from keysight.ads.de import db_uu as db from keysight.ads import dataset import utils classParameterUpdater(ABC):"""参数更新抽象接口"""@abstractmethoddefupdate_parameters(self, design: db.Design, param_row: pd.Series)->None:""" 更新电路参数 Args: design: ADS设计对象 param_row: 包含参数值的pandas Series """passclassAutoSimulator:"""通用自动化仿真器"""def__init__(self, parameter_updater: ParameterUpdater):""" 初始化仿真器 Args: parameter_updater: 参数更新器实例 """ self.parameter_updater = parameter_updater self.design =Nonedefload_parameter_samples(self, filepath:str)-> pd.DataFrame:"""加载参数样本数据，仅支持输出JSON格式""" filepath = Path(filepath)ifnot filepath.exists():raise FileNotFoundError(f"参数文件不存在: {filepath}")if filepath.suffix.lower()==".json":withopen(filepath,"r", encoding="utf-8")as f: data = json.load(f)# 如果JSON数据是字典格式，转换为DataFrameifisinstance(data,dict):if"samples"in data:return pd.DataFrame(data["samples"])else:return pd.DataFrame([data])elifisinstance(data,list):return pd.DataFrame(data)else:raise ValueError("JSON数据格式不支持，需要字典或列表格式")elif filepath.suffix.lower()==".csv":return pd.read_csv(filepath)elif filepath.suffix.lower()in[".xlsx",".xls"]:return pd.read_excel(filepath)else:raise ValueError(f"不支持的文件格式: {filepath.suffix}")defsetup_ads_environment( self, workspace_path:str, library_name:str, design_name:str)->None:"""设置ADS工作环境"""print("设置ADS工作环境...")try: workspace = utils.get_workspace(workspace_path) library = utils.get_library(workspace, library_name) self.design = utils.get_design( library, design_name, utils.ViewType.SCHEMATIC )print("ADS环境设置成功")except Exception as e:raise RuntimeError(f"ADS环境设置失败: {e}")defextract_simulation_results( self, results: dataset.Dataset, target_variables: List[str], debug:bool=True)-> Dict[str, Any]:""" 通用的仿真结果提取函数 Args: results: ADS仿真结果对象 target_variables: 目标变量名列表 debug: 是否输出调试信息 Returns: dict: 提取结果字典，格式为 {变量名: 值/数组, ...} """ extraction_results ={}if debug:print(f" 开始提取仿真结果，目标变量: {target_variables}")# 初始化所有目标变量为 Nonefor var_name in target_variables: extraction_results[var_name]=Nonetry:# 遍历所有变量块for vb_name in results.varblock_names:try: varblock = results[vb_name] dependent_vars =[] independent_vars =[]# 使用 ivars 和 dvars 属性ifhasattr(varblock,"ivars")andhasattr(varblock,"dvars"):try: independent_vars =[ivar.name for ivar in varblock.ivars] dependent_vars =[dvar.name for dvar in varblock.dvars]except Exception as e:if debug:print(f" 使用 ivars/dvars 失败: {e}")if debug:print(f" 检查变量块: {vb_name}")print(f" 依赖变量: {dependent_vars}")# 主要在依赖变量中搜索目标变量 search_vars =( dependent_vars if dependent_vars else(independent_vars + dependent_vars))# 检查每个目标变量for target_var in target_variables:if extraction_results[target_var]isnotNone:continueif target_var in search_vars:try: df = varblock.to_dataframe(dvar_names=[target_var])if target_var in df.columns andlen(df)>0: data_series = df[target_var]iflen(data_series)==1: value =float(data_series.iloc[0]) extraction_results[target_var]= value if debug:print(f" ✓ 提取单值变量: {vb_name}.{target_var} = {value:.3f}")else: value_array = data_series.values.tolist() extraction_results[target_var]= value_array if debug:print(f" ✓ 提取数组变量: {vb_name}.{target_var} (长度: {len(value_array)})")except Exception as e:if debug:print(f" ❌ 提取 {vb_name}.{target_var} 失败: {e}")continueexcept Exception as e:if debug:print(f" ❌ 访问变量块 {vb_name} 失败: {e}")continue# 显示提取结果摘要if debug: successful_vars =[ var for var, val in extraction_results.items()if val isnotNone] failed_vars =[ var for var, val in extraction_results.items()if val isNone]print(" === 提取结果摘要 ===")print(f" 成功提取: {successful_vars}")if failed_vars:print(f" 未找到变量: {failed_vars}")return extraction_results except Exception as e:if debug:print(f" ❌ 仿真结果提取异常: {e}")return extraction_results defrun_simulation_and_extract_results( self, target_variables: List[str], debug:bool=True)-> Dict[str, Any]:""" 运行仿真并提取多个目标变量的结果 Args: target_variables: 目标变量名列表 debug: 是否输出调试信息 Returns: dict: 提取结果字典 """try:# 创建仿真管理器 sim_manager = utils.SimulationManager(self.design)# 运行仿真 sim_manager.run_design_simulation()# 提取仿真结果with sim_manager.get_simulation_results()as results:return self.extract_simulation_results( results, target_variables, debug=debug )except Exception as e:if debug:print(f"仿真失败: {e}")return{var:Nonefor var in target_variables}defrun_batch_simulation( self, data_filepath:str, workspace_path:str, library_name:str, design_name:str, target_variables: List[str], goal_variables: List[str]=None, result_processor: Optional[Callable]=None, max_samples: Optional[int]=None, save_to_original:bool=True,)-> pd.DataFrame:""" 运行批量仿真 Args: data_filepath: 参数数据文件路径 workspace_path: ADS工作空间路径 library_name: 库名称 design_name: 设计名称 target_variables: 仿真提取变量列表 goal_variables: 目标变量列表 result_processor: 结果处理函数，接收results字典，返回处理后的结果 max_samples: 最大样本数限制 save_to_original: 是否保存结果到原始数据文件 Returns: 包含仿真结果的DataFrame """print("=== 自动化仿真开始 ===")print(f"数据文件: {data_filepath}")print(f"工作空间: {workspace_path}")print(f"库名称: {library_name}")print(f"设计名称: {design_name}")# 1. 加载参数样本数据print(f"加载参数数据: {data_filepath}") df = self.load_parameter_samples(data_filepath)print(f"成功加载 {len(df)} 个样本")# 应用样本数限制if max_samples isnotNone: df = df.head(max_samples)print(f"测试模式：只处理前 {len(df)} 个样本")# 2. 设置ADS工作环境 self.setup_ads_environment(workspace_path, library_name, design_name)# 3. 循环处理每个样本 simulation_results =[]for idx, row in df.iterrows(): sample_id = row.get("sample_id", idx +1)print(f"\n处理样本 {sample_id} ({idx+1}/{len(df)})...")try:# 更新电路参数 self.parameter_updater.update_parameters(self.design, row)print(" 参数更新完成")if os.path.exists( os.path.join(self.design.cell.path,"data", design_name +".ds")): os.remove( os.path.join(self.design.cell.path,"data", design_name +".ds"))# 运行仿真并提取结果 results = self.run_simulation_and_extract_results(target_variables)# 处理结果if result_processor: processed_results = result_processor(results)else: processed_results = results simulation_results.append(processed_results)# 显示结果for var_name, value in processed_results.items():if value isnotNone:ifisinstance(value,(int,float)):print(f" {var_name} = {value:.3f}")else:print(f" {var_name} = {type(value).__name__}(长度: {len(value)ifhasattr(value,'__len__')else'N/A'})")except Exception as e:print(f" 样本 {sample_id} 处理失败: {e}") simulation_results.append({var:Nonefor var in target_variables})if goal_variables isNone: goal_variables = target_variables # 4. 将结果添加到原始数据并保存if save_to_original: self._save_results_to_original_file( df, simulation_results, goal_variables, data_filepath )else: df["simulation_results"]= simulation_results # 5. 显示统计信息 self._print_simulation_statistics(df, goal_variables)print("=== 自动化仿真完成 ===")return df def_save_results_to_original_file( self, df: pd.DataFrame, simulation_results: List[Dict[str, Any]], goal_variables: List[str], data_filepath:str,)->None:""" 将仿真结果保存到原始数据文件中，追加到现有的simulation_results中 Args: df: 原始数据DataFrame simulation_results: 仿真结果列表 goal_variables: 目标变量列表 data_filepath: 原始数据文件路径 """print("\n保存结果...")# 处理新的仿真结果 new_formatted_results =[]for result_dict in simulation_results: new_result ={}for var_name in goal_variables: value = result_dict.get(var_name)if value isnotNone:# 处理不同类型的数据 processed_value = self._process_value_for_json(value) new_result[f"{var_name}"]= processed_value else: new_result[f"{var_name}"]=None new_formatted_results.append(new_result)# 合并现有的simulation_results和新结果 merged_results =[]for i,(_, row)inenumerate(df.iterrows()):# 获取现有的simulation_results（如果存在） existing_results = row.get("simulation_results",{})# 如果现有结果不是字典，初始化为空字典ifnotisinstance(existing_results,dict): existing_results ={}# 获取对应的新结果 new_result =( new_formatted_results[i]if i <len(new_formatted_results)else{})# 合并结果：新结果会覆盖同名的现有结果 merged_result = existing_results.copy()# 保留现有数据 merged_result.update(new_result)# 添加/覆盖新数据 merged_results.append(merged_result)# 将合并后的结果添加到DataFrame df["simulation_results"]= merged_results # 保存到原始文件withopen(data_filepath,"w", encoding="utf-8")as f: json.dump(df.to_dict(orient="records"), f, indent=2, ensure_ascii=False)print(f"结果已保存到: {data_filepath}")def_process_value_for_json(self, value: Any)-> Any:""" 处理值以确保JSON可序列化 Args: value: 要处理的值 Returns: JSON可序列化的值 """# 处理复数ifisinstance(value,complex):return{"real":float(value.real),"imag":float(value.imag),"magnitude":float(abs(value)),"phase_deg":float(np.angle(value, deg=True)),}# 处理numpy数组或列表elifisinstance(value,(list, np.ndarray)):iflen(value)==0:return[]eliflen(value)==1:# 单个值，递归处理return self._process_value_for_json(value[0])else:# 多个值，处理每个元素 processed_list =[]for item in value: processed_list.append(self._process_value_for_json(item))return processed_list # 处理numpy标量elifisinstance(value, np.number):if np.iscomplex(value):return self._process_value_for_json(complex(value))else:returnfloat(value)# 处理NaN和无穷大elifisinstance(value,float):if np.isnan(value)or np.isinf(value):returnNoneelse:return value # 其他类型直接返回else:return value def_print_simulation_statistics( self, df: pd.DataFrame, target_variables: List[str])->None:"""打印仿真统计信息"""print("\n=== 仿真统计 ===")print(f"总样本数: {len(df)}")# 统计成功的仿真 successful_count =0for results in df["simulation_results"]:ifany(results.get(var)isnotNonefor var in target_variables): successful_count +=1print(f"成功仿真: {successful_count}")print(f"失败样本: {len(df)- successful_count}")# 显示各变量的统计信息for var_name in target_variables: values =[]for results in df["simulation_results"]: value = results.get(var_name)if value isnotNoneandisinstance(value,(int,float)): values.append(value)if values: values = np.array(values)print(f"{var_name}范围: {values.min():.3f} ~ {values.max():.3f}")print(f"{var_name}均值: {values.mean():.3f}")

AutoSimulator 采用 模板方法模式 (Template Method Pattern) 设计：

• 不变的部分：ADS 工作空间的打开/关闭、仿真器的实例化、结果文件的遍历与读取、异常处理、数据保存（追加到原始 JSON）。这些通用逻辑被封装在基类和工具函数中。
• 变化的部分：不同的电路有不同的参数需要更新（如电阻值、信号文件路径、偏置电压）。这部分逻辑通过 ParameterUpdater 接口暴露给用户实现。

1.2 核心类说明

utils.auto_simulator.py 中主要包含两个核心类：ParameterUpdater 和 AutoSimulator。

1. ParameterUpdater (抽象基类)

设计意图：这是一个策略接口（Strategy Pattern），将"如何更新电路参数"这一变化点从框架中剥离出来。由于每个 ADS 设计的元件名称、变量控件结构、参数命名规则都不同，这部分逻辑无法通用化，必须由用户根据自己的电路原理图来实现。

类定义：

from abc import ABC, abstractmethod from keysight.ads.de import db_uu as db import pandas as pd classParameterUpdater(ABC):"""参数更新抽象接口"""@abstractmethoddefupdate_parameters(self, design: db.Design, param_row: pd.Series)->None:""" 更新电路参数（用户必须实现） Args: design: ADS 设计对象，指向当前打开的原理图 param_row: 包含本次仿真所需全部参数的 Pandas Series """pass

核心方法详解：

实现时需要掌握的 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() 可让原理图界面同步显示新值（非必需但推荐）

2. 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 格式（见 _save_results_to_original_file）。如需将结果回写到 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(self, data_filepath, workspace_path, library_name, design_name, target_variables, goal_variables=None, result_processor=None, max_samples=None, save_to_original=True) -> pd.DataFrame

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

执行流程：

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(self, df, simulation_results, goal_variables, data_filepath) -> None

将当前批次的仿真结果与原文件中已有的 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.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 classAmplifierUpdater(ParameterUpdater):""" 放大器参数更新器 假设原理图中包含： - VAR1 控件：定义 Vbias（偏置电压）、R_source（源阻抗） - C1 元件：耦合电容 - L1 元件：匹配电感 """defupdate_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 编写主控制脚本

defmain():"""主函数：配置参数并启动批量仿真"""# ========== 配置区 ========== 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 defprocess_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 isnotNoneandisinstance(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.4 为什么推荐使用此模式？

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

1.5 utils 工具库

先前在自动化工具库 auto_simulator.py中也用到了一些我自己二次封装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 # 定义枚举classViewType(Enum): SCHEMATIC ="schematic" SYMBOL ="symbol" LAYOUT ="layout"def__str__(self)->str:return self.value # 定义Literal类型（用于纯字符串输入） ViewNameLiteral = Literal["schematic","symbol","layout"]# 组合类型 ViewNameType = Union[ViewType, ViewNameLiteral]defdesign_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)defget_workspace(workspace_path:str)-> de.Workspace:# Ensure there isn't already a workspace openif de.workspace_is_open(): de.close_workspace()# Cannot create a workspace if the directory already existsif os.path.exists(workspace_path):# raise RuntimeError(f"Workspace directory already 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 finishedreturn workspace defget_library(workspace: de.Workspace, name:str)-> de.Library:# assert workspace.path is not None# Libraries can only be added to an open workspaceassert 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 defget_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 defadd_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 listfor i inrange(len(eq_list)-1): measeqn.parameters["Meas"].repeats.append( db.ParamItemString("Meas","SingleTextLine", eq_list[i +1])) measeqn.update_item_annotation()classManagedCircuitSimulator:"""带有完整资源管理的电路仿真器"""def__init__(self, hpeesof_dir=None): self.simulator = ads.CircuitSimulator(hpeesof_dir)defrun_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:passclassSimulationManager:"""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)defvalidate_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")returnFalsereturnTruedefrun_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 isNone: 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 nameif validate_name: self.validate_design_name()try:# Generate netlistprint("Generating netlist...") self.netlist = self.design.generate_netlist()ifnot self.netlist orlen(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 cleanupprint("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 managerprint(f"Simulation completed, result file: {self.result_path}") self.data_file_is_exist =Trueexcept Exception as e:print(f"Simulation failed: {e}")raisedefget_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 """ifnot self.data_file_is_exist:raise RuntimeError("Simulation has not been run or failed, cannot get results")return dataset.open(self.result_path)