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 自动化仿真框架

Logo

魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。

更多推荐