AI计算机视觉实战指南：从YOLO到多模型Pipeline

import bpy
import numpy as np

def render_object_views(obj_path, output_dir, num_views=100):
    bpy.ops.import_scene.obj(filepath=obj_path)
    obj = bpy.context.selected_objects[0]

    for i in range(num_views):
        # 随机旋转
        obj.rotation_euler = np.random.uniform(0, 6.28, 3)
        # 随机光源位置
        light_object.location = np.random.uniform(-5, 5, 3)
        # 渲染
        bpy.context.scene.render.filepath = f"{output_dir}/view_{i:04d}.png"
        bpy.ops.render.render(write_still=True)

pip install labelImg
labelImg raw_images/ labels/

results = model.train(
    augment=True,
    hsv_h=0.015,    # 色调
    hsv_s=0.7,      # 饱和度
    hsv_v=0.4,      # 亮度
    degrees=15.0,   # 旋转
    translate=0.1,  # 平移
    scale=0.5,      # 缩放
    fliplr=0.5,     # 左右翻转
    mosaic=1.0,     # Mosaic
    mixup=0.1       # Mixup
)

from ultralytics import YOLO

model = YOLO('yolov8n.pt')

results = model.train(
    data='data.yaml',
    epochs=100,
    imgsz=640,
    batch=16,
    device=0,
    project='runs/train',
    name='defect_detection',
    patience=10  # 早停
)

path: dataset/
train: images/train/
val: images/val/
test: images/test/

names:
  0: normal
  1: crack
  2: missing_part
  3: wrong_orientation

from sklearn.utils.class_weight import compute_class_weight

class_weights = compute_class_weight(
    'balanced',
    classes=np.unique(labels),
    y=labels
)

results = model.train(
    ...,
    cls=class_weights.tolist()
)

model.export(
    format='onnx',
    opset=12,
    simplify=True,
    dynamic=False,
    imgsz=640
)

import cv2
import numpy as np
import onnxruntime as ort

class DefectDetector:
    def __init__(self, model_path):
        self.session = ort.InferenceSession(model_path)
        self.input_name = self.session.get_inputs()[0].name
        self.output_names = [o.name for o in self.session.get_outputs()]
        self.class_names = ['normal', 'crack', 'missing_part', 'wrong_orientation']

    def preprocess(self, image):
        img = cv2.resize(image, (640, 640))
        img = img.astype(np.float32) / 255.0
        img = np.transpose(img, (2, 0, 1))
        img = np.expand_dims(img, axis=0)
        return img

    def postprocess(self, outputs, conf_threshold=0.5, iou_threshold=0.45):
        predictions = outputs[0]
        valid_mask = predictions[:, 4] >= conf_threshold
        predictions = predictions[valid_mask]

        boxes = predictions[:, :4]
        scores = predictions[:, 4]
        class_ids = predictions[:, 5:].argmax(axis=1)

        indices = cv2.dnn.NMSBoxes(
            boxes.tolist(), scores.tolist(),
            conf_threshold, iou_threshold
        )

        results = []
        for i in indices:
            results.append({
                'box': list(map(int, boxes[i])),
                'class_id': int(class_ids[i]),
                'class_name': self.class_names[int(class_ids[i])],
                'confidence': float(scores[i])
            })
        return results

    def detect(self, image):
        input_data = self.preprocess(image)
        outputs = self.session.run(
            self.output_names,
            {self.input_name: input_data}
        )
        return self.postprocess(outputs)

import tensorrt as trt

def build_engine(onnx_path):
    TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
    builder = trt.Builder(TRT_LOGGER)
    network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
    parser = trt.OnnxParser(network, TRT_LOGGER)

    with open(onnx_path, 'rb') as model:
        parser.parse(model.read())

    config = builder.create_builder_config()
    config.max_workspace_size = 1 << 30  # 1GB
    return builder.build_engine(network, config)

def preprocess_with_clahe(image):
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
    equalized = clahe.apply(gray)
    return cv2.cvtColor(equalized, cv2.COLOR_GRAY2BGR)

def is_oil_stain(image, box):
    x1, y1, x2, y2 = box
    roi = image[y1:y2, x1:x2]
    gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY)
    glcm = cv2.calcHist([gray], [0], None, [256], [0, 256])
    smoothness = np.sum(glcm[1:255]) / np.sum(glcm[1:256])
    return smoothness > 0.85

# 后处理
for result in results:
    if result['class_name'] in ['crack', 'missing_part']:
        if is_oil_stain(image, result['box']):
            result['confidence'] *= 0.5

def preprocess_image(image_path, max_size=2048):
    img = cv2.imread(image_path)
    if img is None:
        raise ValueError("无法读取图片")

    # 保持宽高比调整大小
    h, w = img.shape[:2]
    scale = min(max_size / max(h, w), 1.0)
    img = cv2.resize(img, (int(w * scale), int(h * scale)))

    # 去噪
    img = cv2.fastNlMeansDenoisingColored(img, None, 10, 10, 7, 21)

    # CLAHE 增强对比度
    lab = cv2.cvtColor(img, cv2.COLOR_BGR2LAB)
    l, a, b = cv2.split(lab)
    l = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)).apply(l)
    img = cv2.cvtColor(cv2.merge([l, a, b]), cv2.COLOR_LAB2BGR)

    return img

from concurrent.futures import ThreadPoolExecutor

class CVModelPool:
    def __init__(self):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.models = {
            'nsfw': self._load_nsfw_model(),
            'ocr': self._load_ocr_model(),
            'detection': self._load_detection_model()
        }

    def infer_all(self, image):
        results = {}
        with ThreadPoolExecutor(max_workers=len(self.models)) as executor:
            futures = {
                name: executor.submit(self.models[name].predict, image)
                for name in self.models
            }
            for name, future in futures.items():
                results[name] = future.result()
        return results

class ResultFusion:
    def __init__(self):
        self.weights = {
            'nsfw': 0.8,
            'ocr': 0.9,
            'detection': 0.7,
            'face': 0.6
        }

    def fuse_results(self, raw_results):
        final_score = 0.0
        reasons = []

        if raw_results['nsfw']['nsfw_score'] > 0.7:
            final_score += self.weights['nsfw']
            reasons.append("疑似违规内容")

        forbidden_words = self._check_forbidden_words(raw_results['ocr']['text'])
        if forbidden_words:
            final_score += self.weights['ocr']
            reasons.append(f"敏感词: {', '.join(forbidden_words)}")

        return {
            'score': min(final_score, 1.0),
            'reasons': reasons,
            'action': self._decide_action(final_score)
        }

    def _decide_action(self, score):
        if score >= 0.8: return 'reject'
        elif score >= 0.5: return 'manual_review'
        else: return 'approve'

import hashlib
import pickle
import os

class CachedInference:
    def __init__(self, model, cache_dir='./cache', ttl=3600):
        self.model = model
        self.cache_dir = cache_dir
        self.ttl = ttl
        os.makedirs(cache_dir, exist_ok=True)

    def _get_image_hash(self, image):
        return hashlib.md5(image.tobytes()).hexdigest()

    def predict(self, image):
        img_hash = self._get_image_hash(image)
        cache_file = os.path.join(self.cache_dir, f"{img_hash}.pkl")

        # 检查缓存
        if os.path.exists(cache_file):
            if time.time() - os.path.getmtime(cache_file) < self.ttl:
                with open(cache_file, 'rb') as f:
                    return pickle.load(f)

        # 推理
        result = self.model.predict(image)
        with open(cache_file, 'wb') as f:
            pickle.dump(result, f)
        return result

FROM pytorch/pytorch:2.0.1-cuda11.7-cudnn8-runtime

RUN apt-get update && apt-get install -y \
    libgl1-mesa-glx \
    libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY src/ /app/src/
WORKDIR /app

CMD ["python", "src/main.py"]

from prometheus_client import Counter, Histogram, Gauge

inference_counter = Counter('cv_inference_total', 'Total inferences', ['model', 'status'])
inference_latency = Histogram('cv_inference_duration_seconds', 'Duration', ['model'])
gpu_usage = Gauge('gpu_usage_percent', 'GPU usage')

class DetectionMonitor:
    def __init__(self, log_file='detection_log.jsonl'):
        self.log_file = log_file

    def log(self, result, inference_time, light_intensity=None):
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'detections': len(result),
            'defect_count': sum(1 for r in result if r['class_name'] != 'normal'),
            'inference_time_ms': inference_time * 1000,
            'light_intensity': light_intensity
        }
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(log_entry) + '\n')