Python通过clr库实现与.NET互操作教程

目录

• Pythonclr库使用详解
• 1. 安装和基础配置
• 安装
• 基础导入
• 2. 加载 .NET 程序集
• 方法一：加载 GAC 中的程序集
• 方法二：加载特定版本程序集
• 方法三：加载本地 DLL 文件
• 3. 基本类型转换
• 数值类型
• 字符串处理
• 数组和集合
• 4. 创建和使用 .NET 对象
• 实例化对象
• 调用静态方法和属性
• 5. 事件处理
• 订阅 .NET 事件
• 使用委托
• 6. 异常处理
• 捕获 .NET 异常
• 7. 高级功能
• 反射和动态调用
• 泛型使用
• 多线程编程
• 8. 实际应用案例
• 案例1：文件监控
• 案例2：Windows 服务交互
• 9. 调试和故障排除
• 调试技巧
• 10. 性能优化建议
• 总结

clr 是 pythonnet 包提供的模块，用于在 Python 和 .NET 之间进行互操作。

pip install pythonnet

import clr
import sys
import os

import clr

# 加载系统程序集
clr.AddReference("System")
clr.AddReference("System.Core")
clr.AddReference("System.Data")
clr.AddReference("System.Windows.Forms")

from System import *
from System.IO import *
from System.Diagnostics import *

import clr

# 加载特定版本
clr.AddReference("System, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089")

# 或者使用 LoadFrom
clr.AddReferenceByName("System, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089")

import clr
import os

# 添加 DLL 搜索路径
clr.AddReference("YourAssembly")  # 如果 DLL 在 Python 路径中

# 或者指定完整路径
clr.AddReferenceToFile(r"C:pathtoYourAssembly.dll")

# 或者使用 LoadFrom
clr.AddReferenceToFileAndPath(r"C:pathtoYourAssembly.dll")

# 添加整个目录到搜索路径
clr.AddReference(".")  # 当前目录

import clr
from System import *

# Python 到 .NET
int_value = Int32(42)
double_value = Double(3.14)
bool_value = Boolean(True)

# .NET 到 Python
python_int = int(int_value)
python_float = float(double_value)
python_bool = bool(bool_value)

print(f".NET Int32: {int_value}, Python int: {python_int}")
print(f".NET Double: {double_value}, Python float: {python_float}")

import clr
from System import *

# 字符串转换
net_string = String("Hello from .NET")
python_string = str(net_string)

# 字符串操作
net_string = String("Hello World")
length = net_string.Length                    # 属性访问
substring = net_string.Substring(0, 5)       # 方法调用
contains = net_string.Contains("Hello")      # 布尔方法

print(f"Length: {length}")
print(f"Substring: {substring}")
print(f"Contains 'Hello': {contains}")

import clr
from System import *

# 创建 .NET 数组
int_array = Array.CreateInstance(Int32, 5)
for i in range(5):
    int_array[i] = i * 10

# 或者使用更简单的方法
from System import Array
int_array2 = Array[int]([1, 2, 3, 4, 5])

# 列表转换
from System.Collections.Generic import List
net_list = List[int]()
net_list.Add(1)
net_list.Add(2)
net_list.Add(3)

# Python 列表到 .NET 列表
python_list = [10, 20, 30]
net_list_from_python = List[int](python_list)

print(f".NET 数组: {[x for x in int_array]}")
print(f".NET 列表: {[x for x in net_list]}")

import clr
from System import *
from System.IO import *
from System.Text import *

# 创建各种对象
sb = StringBuilder()
sb.Append("Hello")
sb.Append(" World")
result = sb.ToString()

# 使用 FileInfo
file_info = FileInfo("test.txt")
print(f"文件存在: {file_info.Exists}")
print(f"文件扩展名: {file_info.Extension}")

# 使用 DateTime
now = DateTime.Now
tomorrow = now.AddDays(1)
time_span = tomorrow - now

print(f"现在: {now}")
print(f"明天: {tomorrow}")
print(f"时间差: {time_span.TotalHours} 小时")

import clr
from System import *
from System.Math import *
from System.Environment import *

# 静态方法
max_value = Max(10, 20)
sqrt_value = Sqrt(16.0)
pi_value = PI

# 静态属性
machine_name = MachineName
os_version = OSVersion
current_directory = CurrentDirectory

print(f"Math.Max(10, 20) = {max_value}")
print(f"Math.Sqrt(16) = {sqrt_value}")
print(f"PI = {pi_value}")
print(f"机器名: {machine_name}")
print(f"操作系统: {os_version}")
print(f"当前目录: {current_directory}")

import clr
from System import *
from System.Timers import *

class TimerHandler:
    def __init__(self):
        self.count = 0
    
    def OnTimerElapsed(self, sender, e):
        self.count += 1
        print(f"定时器触发 {self.count} 次 - {DateTime.Now}")

# 创建定时器
timer = Timer(1000)  # 1秒间隔
handler = TimerHandler()

# 订阅事件
timer.Elapsed += handler.OnTimerElapsed
timer.Start()

print("定时器已启动，按 Enter 停止...")
input()
timer.Stop()

import clr
from System import *

# 定义委托处理程序
def message_handler(message):
    print(f"收到消息: {message}")

# 使用 Action 委托
action_delegate = Action[str](message_handler)
action_delegate.Invoke("Hello from delegate!")

# 使用 Func 委托
def add_numbers(a, b):
    return a + b

func_delegate = Func[int, int, int](add_numbers)
result = func_delegate.Invoke(5, 3)
print(f"5 + 3 = {result}")

import clr
from System import *
from System.IO import *

def file_operations():
    try:
        # 尝试读取不存在的文件
        file_content = File.ReadAllText("nonexistent_file.txt")
        print(file_content)
        
    except FileNotFoundException as e:
        print(f"文件未找到异常: {e.Message}")
        
    except UnauthorizedAccessException as e:
        print(f"访问被拒绝: {e.Message}")
        
    except Exception as e:  # 捕获所有 .NET 异常
        print(f"其他异常: {e.GetType().Name}: {e.Message}")

def division_operations():
    try:
        from System import DivideByZeroException
        result = 10 / 0
        
    except DivideByZeroException as e:
        print(f"除零异常: {e.Message}")
        
    except Exception as e:
        print(f"Python 异常: {e}")

if __name__ == "__main__":
    file_operations()
    division_operations()

import clr
from System import *
from System.Reflection import *

class ReflectionDemo:
    def explore_assembly(self, assembly_name):
        # 加载程序集
        clr.AddReference(assembly_name)
        assembly = Assembly.Load(assembly_name)
        
        print(f"程序集: {assembly.FullName}")
        print("类型列表:")
        
        for type_obj in assembly.GetTypes():
            if type_obj.IsPublic:
                print(f"  {type_obj.FullName}")
                
                # 显示方法
                methods = type_obj.GetMethods(BindingFlags.Public | BindingFlags.Instance | BindingFlags.Static)
                for method in methods[:3]:  # 只显示前3个方法
                    print(f"    方法: {method.Name}")

# 使用示例
demo = ReflectionDemo()
demo.explore_assembly("System")

import clr
from System import *
from System.Collections.Generic import *

def generic_collections():
    # 创建泛型字典
    dictionary = Dictionary[str, int]()
    dictionary["one"] = 1
    dictionary["two"] = 2
    dictionary["three"] = 3
    
    # 遍历字典
    for kvp in dictionary:
        print(f"Key: {kvp.Key}, Value: {kvp.Value}")
    
    # 泛型列表
    list_of_strings = List[str]()
    list_of_strings.Add("Apple")
    list_of_strings.Add("Banana")
    list_of_strings.Add("Cherry")
    
    # 使用 LINQ (需要引用 System.Linq)
    try:
        clr.AddReference("System.Core")
        from System.Linq import Enumerable
        
        filtered = Enumerable.Where(list_of_strings, lambda x: x.startswith("A"))
        print("以 A 开头的项目:")
        for item in filtered:
            print(f"  {item}")
            
    except Exception as e:
        print(f"LINQ 不可用: {e}")

if __name__ == "__main__":
    generic_collections()

import clr
from System import *
from System.Threading import *
from System.ComponentModel import *

class BackgroundWorkerDemo:
    def __init__(self):
        self.worker = BackgroundWorker()
        self.worker.WorkerReportsProgress = True
        self.worker.WorkerSupportsCancellation = True
        
        self.worker.DoWork += self.on_do_work
        self.worker.ProgressChanged += self.on_progress_changed
        self.worker.RunWorkerCompleted += self.on_work_completed
    
    def on_do_work(self, sender, e):
        worker = sender
        for i in range(1, 101):
            if worker.CancellationPending:
                e.Cancel = True
                return
            
            Thread.Sleep(50)  # 模拟工作
            worker.ReportProgress(i, f"处理项目 {i}")
    
    def on_progress_changed(self, sender, e):
        print(f"进度: {e.ProgressPercentage}% - {e.UserState}")
    
    def on_work_completed(self, sender, e):
        if e.Cancelled:
            print("操作被取消")
        elif e.Error:
            print(f"发生错误: {e.Error.Message}")
        else:
            print("操作完成!")
    
    def start(self):
        self.worker.RunWorkerAsync()
    
    def cancel(self):
        self.worker.CancelAsync()

# 使用示例
demo = BackgroundWorkerDemo()
demo.start()

print("后台工作已启动，按 Enter 取消...")
input()
demo.cancel()

import clr
from System import *
from System.IO import *

class FileMonitor:
    def __init__(self, path):
        self.watcher = FileSystemWatcher()
        self.watcher.Path = path
        self.watcher.IncludeSubdirectories = True
        
        # 订阅事件
        self.watcher.Created += self.on_file_created
        self.watcher.Changed += self.on_file_changed
        self.watcher.Deleted += self.on_file_deleted
        self.watcher.Renamed += self.on_file_renamed
    
    def on_file_created(self, sender, e):
        print(f"文件创建: {e.FullPath}")
    
    def on_file_changed(self, sender, e):
        print(f"文件修改: {e.FullPath}")
    
    def on_file_deleted(self, sender, e):
        print(f"文件删除: {e.FullPath}")
    
    def on_file_renamed(self, sender, e):
        print(f"文件重命名: {e.OldFullPath} -> {e.FullPath}")
    
    def start(self):
        self.watcher.EnableRaisingEvents = True
        print(f"开始监控目录: {self.watcher.Path}")
    
    def stop(self):
        self.watcher.EnableRaisingEvents = False
        print("停止监控")

# 使用示例
monitor = FileMonitor(".")
monitor.start()

print("文件监控已启动，按 Enter 停止...")
input()
monitor.stop()

import clr
from System import *
from System.ServiceProcess import *

class ServiceManager:
    def list_services(self):
        services = ServiceController.GetServices()
        print("系统服务列表:")
        for service in services:
            status = service.Status
            print(f"  {service.ServiceName} - {service.DisplayName} - {status}")
    
    def control_service(self, service_name, action):
        try:
            service = ServiceController(service_name)
            
            if action == "start":
                if service.Status == ServiceControllerStatus.Stopped:
                    service.Start()
                    service.WaitForStatus(ServiceControllerStatus.Running, TimeSpan.FromSeconds(30))
                    print(f"服务 {service_name} 已启动")
                else:
                    print(f"服务 {service_name} 不是停止状态")
            
            elif action == "stop":
                if service.Status == ServiceControllerStatus.Running:
                    service.Stop()
                    service.WaitForStatus(ServiceControllerStatus.Stopped, TimeSpan.FromSeconds(30))
                    print(f"服务 {service_name} 已停止")
                else:
                    print(f"服务 {service_name} 不是运行状态")
            
            elif action == "restart":
                if service.Status == ServiceControllerStatus.Running:
                    service.Stop()
                    service.WaitForStatus(ServiceControllerStatus.Stopped, TimeSpan.FromSeconds(30))
                    service.Start()
                    service.WaitForStatus(ServiceControllerStatus.Running, TimeSpan.FromSeconds(30))
                    print(f"服务 {service_name} 已重启")
                else:
                    print(f"服务 {service_name} 不是运行状态")
                    
        except Exception as e:
            print(f"操作服务时出错: {e}")

# 使用示例
manager = ServiceManager()
manager.list_services()

# 注意：需要管理员权限
# manager.control_service("Themes", "stop")

import clr
from System import *
import traceback

def debug_net_interop():
    try:
        # 查看已加载的程序集
        print("已加载的程序集:")
        for assembly in clr.ListAssemblies():
            print(f"  {assembly}")
        
        # 检查类型信息
        clr.AddReference("System")
        from System import String
        
        string_type = type(String("test"))
        print(f"String 类型: {string_type}")
        print(f"String 方法: {[method for method in dir(String) if not method.startswith('_')][:5]}")
        
    except Exception as e:
        print(f"调试过程中出错: {e}")
        traceback.print_exc()

def memory_management():
    """内存管理示例"""
    import gc
    
    # 创建大量 .NET 对象
    objects = []
    for i in range(1000):
        sb = StringBuilder()
        sb.Append(f"String {i}")
        objects.append(sb)
    
    print(f"创建了 {len(objects)} 个对象")
    
    # 强制垃圾回收
    gc.collect()
    print("垃圾回收完成")

if __name__ == "__main__":
    debug_net_interop()
    memory_management()

import clr
from System import *
import time

class PerformanceOptimizer:
    @staticmethod
    def benchmark_operations():
        """性能基准测试"""
        from System.Text import StringBuilder
        
        # 测试字符串拼接性能
        start_time = time.time()
        
        # 方法1：使用 Python 字符串拼接
        result1 = ""
        for i in range(10000):
            result1 += str(i)
        
        time1 = time.time() - start_time
        
        # 方法2：使用 .NET StringBuilder
        start_time = time.time()
        sb = StringBuilder()
        for i in range(10000):
            sb.Append(str(i))
        result2 = sb.ToString()
        
        time2 = time.time() - start_time
        
        print(f"Python 字符串拼接: {time1:.4f} 秒")
        print(f".NET StringBuilder: {time2:.4f} 秒")
        print(f"性能提升: {time1/time2:.2f}x")

    @staticmethod
    def batch_operations():
        """批量操作优化"""
        from System.Collections.Generic import List
        
        # 不推荐的写法：逐个添加
        slow_list = List[int]()
        start_time = time.time()
        for i in range(10000):
            slow_list.Add(i)
        slow_time = time.time() - start_time
        
        # 推荐的写法：批量添加
        fast_list = List[int]()
        start_time = time.time()
        fast_list.AddRange(range(10000))
        fast_time = time.time() - start_time
        
        print(f"逐个添加: {slow_time:.4f} 秒")
        print(f"批量添加: {fast_time:.4f} 秒")
        print(f"性能提升: {slow_time/fast_time:.2f}x")

if __name__ == "__main__":
    PerformanceOptimizer.benchmark_operations()
    PerformanceOptimizer.batch_operations()

clr 库提供了强大的 Python 与 .NET 互操作能力：

1. 程序集加载：支持 GAC、本地文件等多种方式
2. 类型转换：自动处理基本类型转换
3. 对象操作：创建实例、调用方法、访问属性 
4. 事件处理：订阅 .NET 事件和委托
5. 异常处理：捕获和处理 .NET 异常
6. 高级特性：反射、泛型、多线程支持

使用时的注意事项：

• 确保架构匹配（x86/x64）
• 处理内存管理
• 注意性能优化
• 适当处理异常

这个库使得在 Python 中利用丰富的 .NET 生态系统成为可能。

以上就是Python使用clr库实现与.NET互操作教程的详细内容，更多关于Python clr与.NET互操作的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python利用SNMP与Netmiko实现网络监控应用实践
• 在.NET项目中嵌入Python代码的实践指南
• Python错误AttributeError: 'NoneType' object has no attribute问题的彻底解决方法
• Python封装Netcat打造跨平台文件传输利器