oapy/labs/src/lab17_5_threadusemodel.py

#!/usr/bin/env python3
"""
Lab 17-5: 线程使用模型 (Thread Use Model)

本示例演示 OpenAccess 的线程使用模型。
注意：这里不使用多线程，仅展示如何设置和重置线程模型。

线程模型类型：
- oacSingleThreadUseModel: 单线程模式
- oacMultipleReadersThreadUseModel: 多读线程模式（只读）
- oacMultipleWritersThreadUseModel: 多写线程模式
"""

import os
import sys
import shutil

# 导入工具函数和 OA 绑定
from utils import init_oa, c_str, make_oa_string, make_oa_name, get_namespace
from oapy._oa import _base, _dm, _design


# ============================================================================
# 全局配置
# ============================================================================

class Globals:
    """全局配置类，管理库路径和名称"""
    
    _instance = None
    
    @classmethod
    def get_instance(cls):
        """获取全局单例"""
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
    
    def __init__(self):
        """初始化全局配置"""
        # 初始化设计数据库
        _design.oaDesignInit()
        ns = get_namespace("native")
        
        # 库路径和名称
        self.lib_path_str = "../data/LibDir"
        self.lib_name_str = "LibDir"
        self.cell_name_str = "Cell"
        self.view_name_str = "View"
        self.sn_lib = make_oa_name(ns, self.lib_name_str)
        self.sn_cell = make_oa_name(ns, self.cell_name_str)
        self.sn_view = make_oa_name(ns, self.view_name_str)
        self.str_lib_path = make_oa_string(self.lib_path_str)
        
        # 网络名称
        self.sn_net1 = make_oa_name(ns, "Net1")
        self.sn_net2 = make_oa_name(ns, "Net2")
        self.sn_net3 = make_oa_name(ns, "Net3")
        
        # 获取视图类型（原理图）
        self.view_type = _dm.oaViewType.get(_dm.oaReservedViewType(_dm.oaReservedViewTypeEnum.oacSchematic))
        
        # 获取构建版本号
        packages = _base.oaBuildInfoArray()
        _base.oaBuildInfo.getPackages(packages)
        self.build_number = packages[0].getBuildNumber()
        
        print(f"\n*** 使用 OA 构建版本: {self.build_number} ***\n")


# ============================================================================
# 线程模型管理函数
# ============================================================================

def log_setting_thread_use_model(new_model):
    """记录线程模型设置操作"""
    model_name = new_model.getName()
    print(f"session.setThreadUseModel({model_name})")


def set_thread_use_model(new_model_enum):
    """
    设置线程使用模型
    
    参数:
        new_model_enum: oaThreadUseModelEnum 枚举值
    """
    session = _base.oaSession.get()
    current_model = session.getThreadUseModel()
    new_model = _base.oaThreadUseModel(new_model_enum)
    
    print(f"\n切换线程模型: {current_model.getName()} -> {new_model.getName()}")
    
    # 特殊处理：OA 22.41p004 版本的限制
    # 从 MultipleReaders 切换到其他模型时，必须先回到 SingleThread
    OA22_41_p004 = 25346
    globals_obj = Globals.get_instance()
    
    if globals_obj.build_number == OA22_41_p004:
        if (new_model_enum == _base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel and
            new_model_enum != _base.oaThreadUseModelEnum.oacSingleThreadUseModel):
            print("\n[警告] OA 22.41p004 版本限制:")
            print("  从 MultipleReaders 切换到其他模型时，必须先回到 SingleThread")
            print("  否则 API 会抛出 oacInternalError 异常")
            
            # 先切换到单线程模式
            log_setting_thread_use_model(_base.oaThreadUseModel(_base.oaThreadUseModelEnum.oacSingleThreadUseModel))
            session.setThreadUseModel(_base.oaThreadUseModel(_base.oaThreadUseModelEnum.oacSingleThreadUseModel))
    
    # 设置新的线程模型
    log_setting_thread_use_model(new_model)
    session.setThreadUseModel(new_model)
    
    # 确认当前模型
    final_model = session.getThreadUseModel()
    print(f"当前线程模型: {final_model.getName()}\n")


# ============================================================================
# 设计操作函数
# ============================================================================

def create_original_designs():
    """创建初始设计（库、单元、视图和网络）"""
    globals_obj = Globals.get_instance()
    
    # 清理并创建库目录
    if os.path.exists(globals_obj.lib_path_str):
        shutil.rmtree(globals_obj.lib_path_str)
    os.makedirs(globals_obj.lib_path_str, exist_ok=True)
    
    # 创建库（共享模式，使用 DMFileSys 插件）
    lib = _dm.oaLib.create(
        globals_obj.sn_lib,
        make_oa_string(globals_obj.lib_path_str),
        _dm.oaLibMode(_dm.oaLibModeEnum.oacSharedLibMode),
        make_oa_string("oaDMFileSys"),
        _dm.oaDMAttrArray(0)
    )
    
    # 打开设计进行写入
    design = _design.oaDesign.open(
        globals_obj.sn_lib,
        globals_obj.sn_cell,
        globals_obj.sn_view,
        globals_obj.view_type,
        'w'
    )
    
    assert design.isValid(), "设计打开失败"
    
    # 创建顶层 block
    block = _design.oaBlock.create(design, True)
    
    # 创建三个标量网络
    net1 = _design.oaScalarNet.create(block, globals_obj.sn_net1)
    net2 = _design.oaScalarNet.create(block, globals_obj.sn_net2)
    net3 = _design.oaScalarNet.create(block, globals_obj.sn_net3)
    
    print(f"\n已创建网络: {net1.getName()}, {net2.getName()}, {net3.getName()}")
    
    # 保存设计
    print("\n保存设计...")
    design.save()


def count_invalid_nets(block):
    """
    统计无效的 net 数量
    
    参数:
        block: 要检查的 block
        
    返回:
        无效 net 的数量
    """
    invalid_count = 0
    nets = block.getNets()
    
    for net in nets:
        if not net.isValid():
            invalid_count += 1
    
    return invalid_count


def modify_designs():
    """
    修改设计（创建新网络）
    
    行为取决于当前线程模型：
    - MultipleReaders 模式：尝试创建网络会失败（设计为只读）
    - 其他模式：正常创建网络
    """
    globals_obj = Globals.get_instance()
    
    # 查找已存在的设计
    design = _design.oaDesign.find(
        globals_obj.sn_lib,
        globals_obj.sn_cell,
        globals_obj.sn_view
    )
    
    assert design.isValid(), "设计查找失败"
    
    # 获取顶层 block
    block = design.getTopBlock()
    
    print("在设计中创建新标量网络...")
    
    session = _base.oaSession.get()
    current_model = session.getThreadUseModel()
    
    if current_model == _base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel:
        # MultipleReaders 模式下，设计为只读
        print("\n[警告] 当前为 MultipleReaders 模式，设计为只读")
        print("  尝试修改设计会导致未定义行为")
        print("  在 debug 模式下会抛出 oacInternalError 异常")
        print("  在 opt 模式下可能创建无效对象")
        
        # 统计修改前的状态
        before_invalid = count_invalid_nets(block)
        before_total = block.getNets().getCount()
        
        # 尝试创建网络（预期会失败）
        try:
            new_net = _design.oaScalarNet.create(block)
            print("  [注意] 网络创建成功（可能是 opt 模式）")
        except _base.oaException as e:
            print(f"  [预期] 创建失败: {e.getMsg()}")
        
        # 统计修改后的状态
        after_invalid = count_invalid_nets(block)
        after_total = block.getNets().getCount()
        
        print(f"  修改前: {before_total} 个网络 ({before_invalid} 个无效)")
        print(f"  修改后: {after_total} 个网络 ({after_invalid} 个无效)")
        
    else:
        # 其他模式下可以正常创建网络
        print(f"  当前模型: {current_model.getName()}，可以创建网络")
        new_net = _design.oaScalarNet.create(block)
        
        assert new_net.isValid(), "新网络创建失败"
        print(f"  成功创建网络: {c_str(new_net.getName())}")


# ============================================================================
# 测试函数
# ============================================================================

def test_thread_use_model():
    """
    测试不同的线程使用模型
    
    测试流程：
    1. SingleThread 模式：创建初始设计
    2. MultipleReaders 模式：尝试修改（预期失败）
    3. MultipleWriters 模式：修改设计
    4. 多次切换模型并修改
    """
    print("=" * 70)
    print("测试线程使用模型")
    print("=" * 70)
    
    # 测试 1: 单线程模式 - 创建初始设计
    print("\n[测试 1] 单线程模式 - 创建初始设计")
    set_thread_use_model(_base.oaThreadUseModelEnum.oacSingleThreadUseModel)
    create_original_designs()
    
    # 测试 2: 多读模式 - 尝试修改（应该失败）
    print("\n[测试 2] 多读模式 - 尝试修改设计")
    set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel)
    modify_designs()
    
    # 测试 3: 多写模式 - 可以修改
    print("\n[测试 3] 多写模式 - 修改设计")
    set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleWritersThreadUseModel)
    set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleWritersThreadUseModel)  # 重复设置测试
    modify_designs()
    
    # 测试 4: 切换回多读模式
    print("\n[测试 4] 多读模式 - 再次尝试修改")
    set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel)
    set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel)  # 重复设置测试
    modify_designs()
    
    # 测试 5: 回到单线程模式
    print("\n[测试 5] 单线程模式 - 最终修改")
    set_thread_use_model(_base.oaThreadUseModelEnum.oacSingleThreadUseModel)
    modify_designs()
    
    print("\n" + "=" * 70)
    print("线程模型测试完成")
    print("=" * 70)


# ============================================================================
# 主程序
# ============================================================================

def main():
    """主函数"""
    print("\n" + "=" * 70)
    print("Lab 17-5: OpenAccess 线程使用模型演示")
    print("=" * 70)
    
    # 初始化 OA 子系统
    init_oa()
    
    try:
        # 运行测试
        test_thread_use_model()
        
        print("\n........正常终止........\n")
        return 0
        
    except Exception as ex:
        print(f"\n[错误] 未预期的异常: {ex}")
        import traceback
        traceback.print_exc()
        return 1


if __name__ == "__main__":
    sys.exit(main())