Bash
# 安装CANN工具包（推荐使用openEuler的AI容器镜像，环境最完整）
sudo docker pull openeuler/cann:latest

# 启动CANN开发容器
docker run --rm --network host \
           --device /dev/davinci0 \
           --device /dev/davinci_manager \
           -v /usr/local/Ascend/driver/lib64/:/usr/local/Ascend/driver/lib64/ \
           -ti openeuler/cann:latest

Bash
# 安装Toolkit开发套件包
sudo yum install -y Ascend-cann-toolkit-8.3.RC1

# 配置环境变量（关键步骤！）
source /usr/local/Ascend/ascend-toolkit/set_env.sh

# 安装NPU驱动和固件
sudo yum install -y Ascend-driver Ascend-firmware

Python
# check_npu_environment.py
import torch
import numpy as np
import subprocess
import os

def check_ascend_environment():
    print("=" * 60)
    print("openEuler + CANN NPU环境深度检查")
    print("=" * 60)
   
    # 检查NPU是否可用
    if torch.npu.is_available():
        device_count = torch.npu.device_count()
        print(f"✅ 检测到 {device_count} 个NPU设备")
       
        for i in range(device_count):
            device_name = torch.npu.get_device_name(i)
            device_capability = torch.npu.get_device_capability(i)
            print(f"  设备 {i}: {device_name}")
            print(f"  计算能力: {device_capability}")
       
        # NPU张量计算测试
        device = torch.npu.set_device(0)
        x = torch.randn(3, 3).npu()  # 关键：.npu()将数据放到NPU上
        y = torch.randn(3, 3).npu()
        z = torch.matmul(x, y)
       
        print(f"✅ NPU计算测试通过")
        print(f"  矩阵乘法结果形状: {z.shape}")
        print(f"  设备类型: {z.device}")
        print(f"  数据位置: {'NPU' if z.is_npu else 'CPU'}")
       
    else:
        print("❌ NPU不可用，请检查驱动和CANN安装")
        # 检查CANN环境变量
        cann_path = os.getenv('ASCEND_HOME')
        print(f"CANN路径: {cann_path if cann_path else '未设置'}")
        return False
   
    # 检查系统信息
    import platform
    print(f"✅ 操作系统: {platform.platform()}")
    print(f"✅ Python版本: {platform.python_version()}")
    print(f"✅ PyTorch版本: {torch.__version__}")
   
    # 检查npu-smi工具
    try:
        result = subprocess.run(['npu-smi'], capture_output=True, text=True)
        if result.returncode == 0:
            print("✅ npu-smi工具可用")
        else:
            print("❌ npu-smi工具异常")
    except FileNotFoundError:
        print("❌ npu-smi未安装")
   
    return True

if __name__ == "__main__":
    check_ascend_environment()

Python
# pcb_defect_detection_npu.py
import torch
import torch.nn as nn
import torch.optim as optim
import torchvision.transforms as transforms
from torch.utils.data import Dataset, DataLoader
import cv2
import numpy as np
from PIL import Image
import time

# 启用CANN的混合精度训练，大幅提升NPU计算效率
from torch.npu.amp import autocast, GradScaler

class PCBDefectDataset(Dataset):
    """PCB缺陷检测数据集"""
    def __init__(self, num_samples=1000, img_size=224):
        self.num_samples = num_samples
        self.img_size = img_size
        self.defect_types = ['missing_hole', 'mouse_bite', 'open_circuit',
                           'short', 'spur', 'spurious_copper']
        self.transform = transforms.Compose([
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                               std=[0.229, 0.224, 0.225])
        ])
   
    def __len__(self):
        return self.num_samples
   
    def generate_pcb_image(self):
        """生成模拟PCB图像"""
        # 创建绿色PCB底板
        img = np.ones((self.img_size, self.img_size, 3), dtype=np.uint8) * 60
        img[:, :, 1] = 120  # 增强绿色通道
       
        # 绘制基础电路
        cv2.rectangle(img, (30, 30), (180, 80), (200, 150, 80), 3)  # 焊盘
        cv2.line(img, (50, 100), (150, 100), (180, 130, 70), 2)     # 线路
        cv2.circle(img, (100, 150), 15, (200, 150, 80), -1)         # 过孔
       
        return img
   
    def add_defect(self, img, defect_type):
        """添加特定类型的缺陷"""
        if defect_type == 'missing_hole':
            # 漏孔：应该有孔的地方没有孔
            pass  # 这里我们故意不画某个孔
       
        elif defect_type == 'mouse_bite':
            # 鼠咬：线路边缘被咬掉
            cv2.circle(img, (120, 100), 8, (60, 120, 60), -1)
       
        elif defect_type == 'open_circuit':
            # 开路：线路断开
            cv2.line(img, (50, 100), (90, 100), (180, 130, 70), 2)
            cv2.line(img, (110, 100), (150, 100), (180, 130, 70), 2)
       
        elif defect_type == 'short':
            # 短路：不该连接的地方连接
            cv2.line(img, (80, 90), (80, 110), (220, 100, 60), 2)
       
        elif defect_type == 'spur':
            # 毛刺：线路上的凸起
            points = np.array([[140, 95], [150, 90], [155, 100]], np.int32)
            cv2.fillPoly(img, [points], (180, 130, 70))
       
        elif defect_type == 'spurious_copper':
            # 伪铜：多余的铜箔
            cv2.circle(img, (60, 60), 10, (180, 140, 90), -1)
       
        return img
   
    def __getitem__(self, idx):
        # 随机选择缺陷类型（10%的概率为正常样本）
        if idx % 10 == 0:
            defect_type = 'normal'
            label = 6  # 正常样本标签为6
        else:
            defect_type = self.defect_types[idx % len(self.defect_types)]
            label = self.defect_types.index(defect_type)
       
        # 生成图像
        img = self.generate_pcb_image()
        if defect_type != 'normal':
            img = self.add_defect(img, defect_type)
       
        # 应用变换
        img = Image.fromarray(img)
        img = self.transform(img)
       
        return img, label

class FastDefectDetector(nn.Module):
    """轻量级缺陷检测模型（针对NPU优化）"""
    def __init__(self, num_classes=7):
        super(FastDefectDetector, self).__init__()
       
        # 特征提取主干网络 - 使用适合NPU的卷积配置
        self.features = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, stride=2, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
           
            nn.Conv2d(32, 64, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
           
            nn.Conv2d(64, 128, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
           
            nn.Conv2d(128, 256, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.AdaptiveAvgPool2d((1, 1))
        )
       
        # 分类器
        self.classifier = nn.Sequential(
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(inplace=True),
            nn.Linear(128, num_classes)
        )
   
    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        x = self.classifier(x)
        return x

def train_model_npu():
    """在NPU上训练模型（使用混合精度）"""
    print("开始训练PCB缺陷检测模型（NPU加速）...")
   
    # 设置NPU设备
    device = torch.device('npu:0')
    print(f"使用设备: {device}")
    print(f"设备名称: {torch.npu.get_device_name(0)}")
   
    # 准备数据
    dataset = PCBDefectDataset(num_samples=1000)
    train_loader = DataLoader(dataset, batch_size=32, shuffle=True)
   
    # 初始化模型并转移到NPU
    model = FastDefectDetector(num_classes=7).to(device)
    criterion = nn.CrossEntropyLoss().to(device)
    optimizer = optim.Adam(model.parameters(), lr=0.001)
   
    # NPU混合精度训练关键：GradScaler
    scaler = GradScaler()
   
    # 训练循环
    model.train()
    for epoch in range(20):
        total_loss = 0
        correct = 0
        total = 0
       
        for batch_idx, (data, target) in enumerate(train_loader):
            data, target = data.to(device), target.to(device)
           
            optimizer.zero_grad()
           
            # 使用混合精度前向传播
            with autocast():
                output = model(data)
                loss = criterion(output, target)
           
            # 使用scaler进行反向传播
            scaler.scale(loss).backward()
            scaler.step(optimizer)
            scaler.update()
           
            total_loss += loss.item()
            _, predicted = output.max(1)
            total += target.size(0)
            correct += predicted.eq(target).sum().item()
       
        accuracy = 100. * correct / total
        avg_loss = total_loss / len(train_loader)
        print(f'Epoch [{epoch+1}/20], Loss: {avg_loss:.4f}, Accuracy: {accuracy:.2f}%')
   
    # 保存模型
    torch.save(model.state_dict(), 'pcb_defect_model_npu.pth')
    print("NPU模型训练完成，已保存为 pcb_defect_model_npu.pth")
   
    return model

if __name__ == "__main__":
    # 设置默认NPU设备
    torch.npu.set_device(0)
    model = train_model_npu()

Python
# inference_npu_optimized.py
import torch
import time
import numpy as np
import cv2
from pcb_defect_detection_npu import FastDefectDetector, PCBDefectDataset

class PCBDefectInferenceNPU:
    """PCB缺陷检测推理类（NPU极致优化版）"""
    def __init__(self, model_path):
        # 强制使用NPU设备
        self.device = torch.device('npu:0')
        print(f"推理设备: {self.device}")
        print(f"设备名称: {torch.npu.get_device_name(0)}")
       
        # 加载模型到NPU
        self.model = FastDefectDetector(num_classes=7).to(self.device)
        self.model.load_state_dict(torch.load(model_path))
        self.model.eval()  # 设置为评估模式
       
        self.defect_names = ['漏孔', '鼠咬', '开路', '短路', '毛刺', '伪铜', '正常']
       
        # NPU性能优化配置
        self.setup_npu_optimization()
       
    def setup_npu_optimization(self):
        """设置NPU专属优化"""
        # 启用NPU推理模式
        torch.npu.set_compile_mode(jit_compile=True)
        # 设置NPU随机数种子
        torch.npu.manual_seed(42)
       
    def preprocess_image_optimized(self, image_path):
        """优化的图像预处理（模拟AIPP功能）"""
        # 读取图像
        image = cv2.imread(image_path)
       
        # 使用OpenCV进行高效预处理（模拟NPU AIPP的预处理流程）
        # AIPP（AI Pre-Processing）可以在NPU内部完成预处理，避免CPU-NPU数据传输
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
       
        # 调整尺寸
        image = cv2.resize(image, (224, 224))
       
        # 归一化 - 使用与训练相同的参数
        # 在真实部署中，这部分可以通过CANN的AIPP在模型内部完成
        image = image.astype(np.float32) / 255.0
        mean = np.array([0.485, 0.456, 0.406])
        std = np.array([0.229, 0.224, 0.225])
        image = (image - mean) / std
       
        # 转换维度: HWC -> CHW
        image = np.transpose(image, (2, 0, 1))
        image = np.expand_dims(image, axis=0)
       
        # 直接创建NPU张量，避免CPU到NPU的数据传输
        return torch.from_numpy(image).to(self.device)
   
    def benchmark_performance(self, num_runs=1000):
        """性能基准测试（NPU专属）"""
        print("开始NPU性能测试...")
       
        # 生成NPU测试数据
        dummy_input = torch.randn(1, 3, 224, 224).npu()
       
        # NPU Warm-up（重要：NPU首次推理需要初始化时间）
        print("NPU预热运行...")
        for _ in range(100):
            with torch.no_grad():
                _ = self.model(dummy_input)
        torch.npu.synchronize()  # 等待所有NPU操作完成
       
        # 正式测试
        print(f"正式测试 {num_runs} 次...")
        start_time = time.time()
       
        for i in range(num_runs):
            with torch.no_grad():
                output = self.model(dummy_input)
           
            # 每200次同步一次，避免频繁同步影响性能
            if (i + 1) % 200 == 0:
                torch.npu.synchronize()
                elapsed = time.time() - start_time
                avg_time = elapsed / (i + 1) * 1000
                print(f"已运行 {i + 1} 次，平均推理时间: {avg_time:.2f} ms")
       
        torch.npu.synchronize()
        total_time = time.time() - start_time
        avg_inference_time = total_time / num_runs * 1000
       
        # NPU性能统计
        print("=" * 50)
        print(f"NPU性能测试结果:")
        print(f"总运行次数: {num_runs}")
        print(f"总耗时: {total_time:.2f} 秒")
        print(f"平均推理时间: {avg_inference_time:.2f} ms")
        print(f"吞吐量: {1000 / avg_inference_time:.2f} FPS")
        print(f"设备类型: {dummy_input.device}")
        print("=" * 50)
       
        return avg_inference_time
   
    def detect_defect(self, image_path):
        """缺陷检测（NPU加速版）"""
        # 预处理
        input_tensor = self.preprocess_image_optimized(image_path)
       
        # NPU推理
        start_time = time.time()
        with torch.no_grad():
            output = self.model(input_tensor)
        torch.npu.synchronize()  # 确保推理完成
        inference_time = (time.time() - start_time) * 1000
       
        # 后处理
        probabilities = torch.softmax(output, dim=1)
        confidence, predicted = torch.max(probabilities, 1)
       
        defect_type = self.defect_names[predicted.item()]
        confidence_score = confidence.item()
       
        print(f"检测结果: {defect_type}")
        print(f"置信度: {confidence_score:.4f}")
        print(f"NPU推理时间: {inference_time:.2f} ms")
        print(f"设备利用率: {'NPU' if input_tensor.is_npu else 'CPU'}")
       
        return defect_type, confidence_score, inference_time

def compare_cpu_npu_performance():
    """对比CPU和NPU性能差异"""
    print("\n" + "=" * 60)
    print("CPU vs NPU 性能对比测试")
    print("=" * 60)
   
    # NPU性能
    print("测试NPU性能...")
    detector_npu = PCBDefectInferenceNPU('pcb_defect_model_npu.pth')
    npu_time = detector_npu.benchmark_performance(num_runs=500)
   
    # CPU性能（在NPU设备上创建CPU模型进行对比）
    print("\n测试CPU性能...")
    model_cpu = FastDefectDetector(num_classes=7)
    model_cpu.load_state_dict(torch.load('pcb_defect_model_npu.pth', map_location='cpu'))
    model_cpu.eval()
   
    dummy_input_cpu = torch.randn(1, 3, 224, 224)
   
    # CPU Warm-up
    for _ in range(50):
        with torch.no_grad():
            _ = model_cpu(dummy_input_cpu)
   
    # CPU测试
    start_time = time.time()
    for i in range(500):
        with torch.no_grad():
            _ = model_cpu(dummy_input_cpu)
    cpu_time = (time.time() - start_time) * 1000 / 500
   
    # 性能对比
    speedup = cpu_time / npu_time
   
    print("\n" + "=" * 50)
    print("性能对比结果:")
    print(f"CPU平均推理时间: {cpu_time:.2f} ms")
    print(f"NPU平均推理时间: {npu_time:.2f} ms")
    print(f"NPU加速比: {speedup:.2f}x")
    print(f"性能提升: {(speedup - 1) * 100:.1f}%")
    print("=" * 50)

def main():
    """主函数"""
    # 初始化NPU推理引擎
    detector = PCBDefectInferenceNPU('pcb_defect_model_npu.pth')
   
    # NPU性能测试
    print("开始NPU性能基准测试...")
    avg_time = detector.benchmark_performance(num_runs=1000)
   
    # 性能对比
    compare_cpu_npu_performance()
   
    # 模拟真实检测场景
    print("\n模拟真实NPU检测场景...")
    dataset = PCBDefectDataset(num_samples=10)
   
    for i in range(5):
        print(f"\n--- 样本 {i+1} ---")
        sample, label = dataset[i]
       
        # 保存临时图像用于测试
        temp_image = (sample.permute(1, 2, 0).numpy() * 255).astype(np.uint8)
        cv2.imwrite(f'temp_sample_{i}.jpg', temp_image)
       
        # 执行NPU检测
        defect_type, confidence, inference_time = detector.detect_defect(f'temp_sample_{i}.jpg')
       
        print(f"真实标签: {dataset.defect_names[label] if label < 6 else '正常'}")
        print(f"预测结果: {defect_type}")
        print(f"置信度: {confidence:.4f}")
        print(f"NPU推理时间: {inference_time:.2f} ms")

if __name__ == "__main__":
    main()

Python
代码块
# 定期清理NPU缓存，防止内存碎片
torch.npu.empty_cache()

# 使用NPU内存分析工具
torch.npu.memory_summary()

# 监控NPU内存使用
allocated = torch.npu.memory_allocated(0) / 10243
cached = torch.npu.memory_cached(0) / 10243
print(f"NPU内存使用: 已分配 {allocated:.2f}GB, 缓存 {cached:.2f}GB")

Bash
# 使用npu-smi监控NPU状态
npu-smi info

# 监控NPU利用率
watch -n 1 npu-smi info