ADS .ds 数据集结构与Python访问实践 ADS的python程控
·
1. 文件层面:.ds文件的物理存储结构
1.1 .ds文件格式
.ds文件 (ADS Dataset File)
├── 文件头 (Header)
│ ├── 版本信息
│ ├── 创建时间
│ └── 元数据
├── 变量块索引 (VariableBlock Index)
│ ├── 块名称列表
│ ├── 块偏移量
│ └── 块大小信息
└── 数据块 (Data Blocks)
├── VariableBlock_1
│ ├── 独立变量定义
│ ├── 依赖变量定义
│ └── 数据矩阵
├── VariableBlock_2
└── ...
1.2 典型的ADS仿真.ds文件内容(以例子说明)
from keysight.ads import de
from keysight.ads.de import db_uu as db
from keysight.edatoolbox import ads
import keysight.ads.dataset as dataset
import os
import matplotlib.pyplot as plt
from IPython.core import getipython
from pathlib import Path
import numpy as np
workspace_path = "D:/ADS_Python_Tutorials/tutorial4_wrk"
cell_name = "python_schematic"
library_name = "tutorial4_lib"
def create_and_open_an_empty_workspace(workspace_path: str):
# 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):
raise RuntimeError(f"Workspace directory already exists: {workspace_path}")
# Create the workspace
workspace = de.create_workspace(workspace_path)
# Open the workspace
workspace.open()
# Return the open workspace and close when it finished
return workspace
def create_a_library_and_add_it_to_the_workspace(workspace: de.Workspace):
# 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_path = workspace.path / library_name
# Create the library
de.create_new_library(library_name, 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)
lib.setup_schematic_tech()
return lib
ws = create_and_open_an_empty_workspace(workspace_path)
# Create and add library to the empty workspace using the pointer to the workspace
lib = create_a_library_and_add_it_to_the_workspace(ws)
def create_schematic(library: de.Library):
design = db.create_schematic(f"{library_name}:{cell_name}:schematic")
num_inds = 5
num_caps = num_inds - 1
for i in range(num_inds):
ind = design.add_instance("ads_rflib:L:symbol", (i * 2, 0))
ind.parameters["L"].value = f"L{i + 1} nH"
ind.update_item_annotation()
design.add_wire([(i * 2 + 1, 0), (i * 2 + 2, 0)])
for i in range(num_caps):
cap = design.add_instance("ads_rflib:C:symbol", (i * 2 + 1.5, -1), angle=-90)
cap.parameters["C"].value = f"C{i + 1} pF"
cap.update_item_annotation()
design.add_wire([(i * 2 + 1.5, 0), (i * 2 + 1.5, -1)])
design.add_instance("ads_rflib:GROUND:symbol", (i * 2 + 1.5, -2), angle=-90)
design.add_instance("ads_simulation:TermG:symbol", (-1, -1), angle=-90)
design.add_instance("ads_simulation:TermG:symbol", (10, -1), angle=-90)
design.add_wire([(-1, -1), (-1, 0), (0, 0)])
design.add_wire([(10, -1), (10, 0)])
sp = design.add_instance("ads_simulation:S_Param:symbol", (0, 2))
sp.parameters["Start"].value = "0.01 GHz"
sp.parameters["Stop"].value = "0.5 GHz"
sp.parameters["Step"].value = "0.001 GHz"
sp.update_item_annotation()
design.save_design()
return design
# Create schematic with the lib object/pointer
design = create_schematic(lib)
##### VAR Definition #####
L_values = ["100", "40", "100", "40", "100"]
C_values = ["30", "10", "30", "10"]
var_inst = design.add_instance(
("ads_datacmps", "VAR", "symbol"), (3.5, 1.875), name="VAR1", angle=90
)
for val in range(len(L_values)):
var_inst.vars[f"L{val + 1}"] = L_values[val]
del var_inst.vars["X"]
var_inst = design.add_instance(
("ads_datacmps", "VAR", "symbol"), (5, 1.875), name="VAR2", angle=90
)
for val in range(len(C_values)):
var_inst.vars[f"C{val + 1}"] = C_values[val]
del var_inst.vars["X"]
##### Measurement Equation Block #####
eq_list = [
"groupdelay=(-1/360)*diff(unwrap(phase(S(2,1))))/diff(freq)",
"s21mag=mag(S(2,1))",
"s21phase=phase(S(2,1))",
]
def add_measeqn(design, eq_name, eq_list):
# add MeasEqn to the schmatic
measeqn = design.add_instance(
("ads_simulation", "MeasEqn", "symbol"), (6.5, 1.875), name=eq_name, angle=-90
)
# 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()
add_measeqn(design, "Meas1", eq_list)
### Netlist Creation and Simulation ###
netlist = design.generate_netlist()
simulator = ads.CircuitSimulator()
target_output_dir = os.path.join(workspace_path, "data")
simulator.run_netlist(netlist, output_dir=target_output_dir)
##### Data Processing & Plot #####
output_data = dataset.open(
Path(os.path.join(target_output_dir, f"{cell_name}" + ".ds"))
)
# Inspect available data blocks in the dataset
print("Available Data Blocks: ", output_data.varblock_names)
# Finding relevant data block containing our results
for datablock in output_data.find_varblocks_with_var_name("groupdelay"):
print("Group Delay expression is found in:", datablock.name)
gd = datablock.name
for datablock in output_data.find_varblocks_with_var_name("S[2,1]"):
print("S21 measurement is found in:", datablock.name)
sp = datablock.name
# Convert SP1.SP datablock to the pandas dataframe
mydata = output_data[sp].to_dataframe().reset_index()
# Convert Group Delay datablock to the pandas dataframe
mygd = output_data[gd].to_dataframe().reset_index()
# Extract data and convert S21 & S11 to dB
freq = mydata["freq"] / 1e6
S21 = 20 * np.log10(abs(mydata["S[2,1]"]))
S11 = 20 * np.log10(abs(mydata["S[1,1]"]))
# Plot results using inline plot from matplotlib
ipython = getipython.get_ipython()
ipython.run_line_magic("matplotlib", "inline")
_, ax = plt.subplots()
ax.set_title("Python Filter Response")
plt.xlabel("Frequency (MHz)")
plt.ylabel("S21 and S11 (dB)")
plt.grid(True)
plt.plot(freq, S21)
plt.plot(freq, S11)
# Plot Group Delay results using inline plot from matplotlib
freq = mygd["freq"] / 1e6
groupdelay = mygd["groupdelay"] / 1e-9
ipython = getipython.get_ipython()
ipython.run_line_magic("matplotlib", "inline")
_, ax = plt.subplots()
ax.set_title("Filter Group Delay Response")
plt.xlabel("Frequency (MHz)")
plt.ylabel("Group Delay (nsec)")
plt.grid(True)
plt.plot(freq, groupdelay)
电路仿真数据结构分析:
CellName.ds
├── SP1.SP (S参数块)
│ ├── ivars: [freq]
│ ├── dvars: [S[1,1], S[2,1], S[1,2], S[2,2]]
│ └── data: 频率 × S参数矩阵
├── Meas1.m (测量方程块)
│ ├── ivars: [freq]
│ ├── dvars: [groupdelay, s21mag, s21phase]
│ └── data: 频率 × 计算结果矩阵
└── 其他仿真块...
2. 类层次关系详解
2.1 类继承和包含关系图
Dataset (collections.abc.Mapping)
├── 包含多个 VariableBlock 对象
│ ├── VariableBlock_1 ("SP1.SP")
│ ├── VariableBlock_2 ("Meas1.m")
│ └── VariableBlock_N (...)
│
每个 VariableBlock 包含:
├── ivars: List[Variable] # 独立变量列表
├── dvars: List[Variable] # 依赖变量列表
│
每个 Variable 包含:
├── name: str # 变量名
├── data_type: Type # Python数据类型
├── dtype: np.dtype # NumPy数据类型
├── is_indep: bool # 是否为独立变量
├── attrs: VariableAttributes # 变量属性
└── variable_type: str # 变量类型标识
3. 数据组织详解
3.1 .ds文件到Dataset对象的映射
def demonstrate_file_to_dataset_mapping():
"""演示文件到Dataset对象的映射过程"""
# 1. 物理文件层面
ds_file_path = "workspace/data/python_schematic.ds"
# 2. Dataset对象创建
with dataset.open(ds_file_path) as ds:
# Dataset对象封装了底层的C++实现
print(f"Dataset._impl: {type(ds._impl)}")
print(f"Dataset._path: {ds._path}")
# 3. 文件内容映射到Python对象
print("文件内容映射:")
for vb_name in ds.varblock_names:
print(f" 文件块 '{vb_name}' -> VariableBlock对象")
vb = ds[vb_name]
print(f" 包含 {vb.ivars_count} 个独立变量")
print(f" 包含 {vb.dvars_count} 个依赖变量")
3.2 Dataset中多个VariableBlock的组织
class DatasetStructureAnalyzer:
"""数据集结构分析器"""
def __init__(self, ds: dataset.Dataset):
self.ds = ds
def analyze_varblock_organization(self):
"""分析VariableBlock的组织方式"""
print("=== VariableBlock组织结构 ===")
# Dataset实现了Mapping接口
print(f"Dataset作为Mapping: {isinstance(self.ds, dict)}")
print(f"支持的操作:")
print(f" - len(ds): {len(self.ds)}")
print(f" - 'SP1.SP' in ds: {'SP1.SP' in self.ds}")
print(f" - list(ds.keys()): {list(self.ds.keys())}")
# 遍历所有VariableBlock
print(f"\n遍历方式:")
print("1. 通过varblock_names:")
for name in self.ds.varblock_names:
vb = self.ds[name]
print(f" {name} -> {type(vb)}")
print("2. 通过items():")
for name, vb in self.ds.items():
print(f" {name} -> {type(vb)}")
print("3. 通过varblocks属性:")
for name, vb in self.ds.varblocks.items():
print(f" {name} -> {type(vb)}")
def find_related_varblocks(self):
"""查找相关的VariableBlock"""
# 查找包含特定变量的块
freq_blocks = list(self.ds.find_varblocks_with_var_name("freq"))
s_param_blocks = list(self.ds.find_varblocks_with_var_name("S[2,1]"))
print(f"\n=== 相关VariableBlock查找 ===")
print(f"包含'freq'的块: {[vb.name for vb in freq_blocks]}")
print(f"包含'S[2,1]'的块: {[vb.name for vb in s_param_blocks]}")
return freq_blocks, s_param_blocks
3.3 VariableBlock中变量的组织
def analyze_variable_organization(vb: dataset.VariableBlock):
"""分析VariableBlock中变量的组织"""
print(f"=== VariableBlock '{vb.name}' 变量组织 ===")
# 独立变量和依赖变量的分离
print(f"独立变量 (ivars): {vb.ivars_count} 个")
print(f"依赖变量 (dvars): {vb.dvars_count} 个")
# 变量访问方式
print(f"\n变量访问方式:")
# 1. 通过索引访问
if vb.ivars:
first_ivar = vb.ivars[0]
print(f"第一个独立变量: {first_ivar.name}")
# 2. 通过名称查找
try:
freq_var = vb.var("freq")
print(f"频率变量: {freq_var.name}")
except KeyError:
print("未找到频率变量")
# 3. 遍历所有变量
print(f"\n所有变量列表:")
all_vars = vb.ivars + vb.dvars
for i, var in enumerate(all_vars):
var_type = "独立" if var.is_indep else "依赖"
print(f" [{i}] {var.name} ({var_type}变量, {var.data_type})")
4. 实际示例:S参数仿真数据
4.1 典型的S参数数据结构
def analyze_s_parameter_data(ds_path: str):
"""分析S参数仿真数据的完整结构"""
with dataset.open(ds_path) as ds:
print("=== S参数数据结构分析 ===")
# 查找S参数变量块
s_param_vb = None
for vb in ds.find_varblocks_with_var_name("S[2,1]"):
s_param_vb = vb
break
if not s_param_vb:
print("未找到S参数数据")
return
print(f"S参数变量块: {s_param_vb.name}")
# 分析独立变量 (通常是频率)
print(f"\n独立变量分析:")
for ivar in s_param_vb.ivars:
print(f" 变量名: {ivar.name}")
print(f" 数据类型: {ivar.data_type}")
print(f" 变量类型: {ivar.variable_type}")
# 如果是频率变量,可能有单位信息
if 'units' in ivar.attrs:
print(f" 单位: {ivar.attrs['units']}")
# 分析依赖变量 (S参数)
print(f"\n依赖变量分析:")
s_parameters = []
for dvar in s_param_vb.dvars:
print(f" 变量名: {dvar.name}")
print(f" 数据类型: {dvar.data_type}") # 通常是complex
print(f" 变量类型: {dvar.variable_type}") # 通常是"s-parameters"
if dvar.name.startswith('S['):
s_parameters.append(dvar.name)
print(f"\n检测到的S参数: {s_parameters}")
# 数据提取示例
df = s_param_vb.to_dataframe().reset_index()
print(f"\nDataFrame结构:")
print(f" 形状: {df.shape}")
print(f" 列名: {df.columns.tolist()}")
print(f" 数据类型:")
for col in df.columns:
print(f" {col}: {df[col].dtype}")
return analyze_s_parameter_values(df)
def analyze_s_parameter_values(df):
"""分析S参数数值特性"""
print(f"\n=== S参数数值分析 ===")
# 频率分析
if 'freq' in df.columns:
freq = df['freq']
print(f"频率范围: {freq.min():.2e} - {freq.max():.2e} Hz")
print(f"频率点数: {len(freq)}")
print(f"频率步长: {(freq.max() - freq.min()) / (len(freq) - 1):.2e} Hz")
# S参数分析
s_param_cols = [col for col in df.columns if col.startswith('S[')]
for s_param in s_param_cols:
s_data = df[s_param]
# 复数数据分析
magnitude = np.abs(s_data)
phase = np.angle(s_data, deg=True)
magnitude_db = 20 * np.log10(magnitude)
print(f"\n{s_param} 分析:")
print(f" 数据类型: {s_data.dtype}")
print(f" 幅度范围: {magnitude.min():.6f} - {magnitude.max():.6f}")
print(f" 幅度(dB)范围: {magnitude_db.min():.2f} - {magnitude_db.max():.2f} dB")
print(f" 相位范围: {phase.min():.2f} - {phase.max():.2f} 度")
# 检查数据完整性
if s_data.isna().any():
print(f" ⚠️ 包含 {s_data.isna().sum()} 个NaN值")
if np.isinf(s_data).any():
print(f" ⚠️ 包含 {np.isinf(s_data).sum()} 个无穷值")
4.2 测量方程数据结构
def analyze_measurement_equation_data(ds_path: str):
"""分析测量方程数据结构"""
with dataset.open(ds_path) as ds:
print("=== 测量方程数据结构分析 ===")
# 查找测量方程变量块
meas_blocks = []
for vb_name in ds.varblock_names:
if '.m' in vb_name.lower() or 'meas' in vb_name.lower():
meas_blocks.append(ds[vb_name])
for meas_vb in meas_blocks:
print(f"\n测量方程块: {meas_vb.name}")
# 分析计算得出的变量
print(f"计算变量:")
for dvar in meas_vb.dvars:
print(f" {dvar.name}: {dvar.data_type}")
# 特殊变量分析
if 'groupdelay' in dvar.name.lower():
print(f" -> 群延迟变量")
elif 'mag' in dvar.name.lower():
print(f" -> 幅度变量")
elif 'phase' in dvar.name.lower():
print(f" -> 相位变量")
# 转换为DataFrame进行分析
df = meas_vb.to_dataframe().reset_index()
print(f" 数据形状: {df.shape}")
# 分析计算结果的数值特性
for col in df.columns:
if col != 'freq': # 跳过频率列
data = df[col]
if np.issubdtype(data.dtype, np.number):
print(f" {col}: 范围 [{data.min():.3e}, {data.max():.3e}]")
5. API使用:完整的数据遍历和提取
5.1 完整的数据遍历示例
def complete_data_traversal(ds_path: str):
"""完整的数据遍历示例"""
with dataset.open(ds_path) as ds:
print("=== 完整数据遍历 ===")
# 第一层:Dataset级别
print(f"Dataset: {ds.path}")
print(f"包含 {len(ds)} 个变量块")
# 第二层:VariableBlock级别
for vb_name, vb in ds.items():
print(f"\n VariableBlock: {vb_name}")
print(f" 独立变量: {vb.ivars_count}, 依赖变量: {vb.dvars_count}")
# 第三层:Variable级别
print(f" 独立变量详情:")
for ivar in vb.ivars:
print(f" - {ivar.name} ({ivar.data_type})")
# 第四层:Variable属性
if ivar.attrs:
for attr_name, attr_value in ivar.attrs.items():
print(f" 属性 {attr_name}: {attr_value}")
print(f" 依赖变量详情:")
for dvar in vb.dvars:
print(f" - {dvar.name} ({dvar.data_type})")
if dvar.attrs:
for attr_name, attr_value in dvar.attrs.items():
print(f" 属性 {attr_name}: {attr_value}")
# 数据提取
try:
df = vb.to_dataframe()
print(f" 数据矩阵: {df.shape}")
print(f" 内存使用: {df.memory_usage(deep=True).sum() / 1024:.1f} KB")
except Exception as e:
print(f" 数据提取失败: {e}")
5.2 高效的数据提取策略
class EfficientDataExtractor:
"""高效的数据提取器"""
def __init__(self, ds_path: str):
self.ds_path = ds_path
self._ds = None
def __enter__(self):
self._ds = dataset.open(self.ds_path)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self._ds:
self._ds.__exit__(exc_type, exc_val, exc_tb)
def extract_frequency_data(self):
"""提取频率数据"""
freq_blocks = list(self._ds.find_varblocks_with_var_name("freq"))
if not freq_blocks:
return None
# 选择第一个包含频率的块
vb = freq_blocks[0]
freq_var = vb.var("freq")
# 只提取频率列
df = vb.to_dataframe(dvar_names=[]) # 只要独立变量
return df['freq'].values
def extract_s_parameters(self, s_param_names=None):
"""提取指定的S参数"""
if s_param_names is None:
s_param_names = ['S[1,1]', 'S[2,1]', 'S[1,2]', 'S[2,2]']
# 查找S参数块
s_param_vb = None
for vb in self._ds.find_varblocks_with_var_name("S[2,1]"):
s_param_vb = vb
break
if not s_param_vb:
return None
# 只提取需要的S参数
available_s_params = [name for name in s_param_names
if any(dvar.name == name for dvar in s_param_vb.dvars)]
df = s_param_vb.to_dataframe(dvar_names=available_s_params)
return df.reset_index()
def extract_measurement_results(self, meas_names=None):
"""提取测量方程结果"""
results = {}
for vb_name in self._ds.varblock_names:
if '.m' in vb_name.lower():
vb = self._ds[vb_name]
if meas_names:
# 只提取指定的测量结果
available_meas = [name for name in meas_names
if any(dvar.name == name for dvar in vb.dvars)]
if available_meas:
df = vb.to_dataframe(dvar_names=available_meas)
results[vb_name] = df.reset_index()
else:
# 提取所有测量结果
df = vb.to_dataframe()
results[vb_name] = df.reset_index()
return results
# 使用示例
def efficient_extraction_example(ds_path: str):
"""高效提取示例"""
with EfficientDataExtractor(ds_path) as extractor:
# 只提取需要的数据
freq = extractor.extract_frequency_data()
s_params = extractor.extract_s_parameters(['S[2,1]', 'S[1,1]'])
measurements = extractor.extract_measurement_results(['groupdelay'])
print(f"频率点数: {len(freq) if freq is not None else 0}")
print(f"S参数数据形状: {s_params.shape if s_params is not None else 'None'}")
print(f"测量结果块数: {len(measurements)}")
return freq, s_params, measurements
6. 数据流向图
物理文件 (.ds)
↓
dataset.open() -> Dataset 对象
↓
遍历 varblock_names -> 获取 VariableBlock 对象
↓
检查 ivars/dvars -> 确认目标变量位置
↓
varblock.to_dataframe() -> Pandas DataFrame
↓
NumPy/Python List -> 数据分析/存储
这个层次结构设计使得ADS的仿真数据可以方便地与Python的科学计算生态系统集成,同时保持了数据的完整性和访问效率。
7.一个完备的ADS批量仿真工具库
为了简化 ADS 仿真程控的开发难度,我提供了一个通用的自动化工具库。该工具库封装了从环境配置、参数更新、仿真运行到结果提取的全流程,使得用户只需关注“如何将参数应用到电路”这一核心逻辑。
文章链接:ADS 自动化仿真框架
魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。
更多推荐


所有评论(0)