# ✅ 推荐命名风格
image_input = cv2.imread("input.jpg")     # 图像变量用 snake_case
gray_image = cv2.cvtColor(image_input, cv2.COLOR_BGR2GRAY)
binary_mask = cv2.threshold(gray_image, 127, 255, cv2.THRESH_BINARY)[1]
contour_list, _ = cv2.findContours(binary_mask, cv2.RETR_EXTERNAL,
                                     cv2.CHAIN_APPROX_SIMPLE)

# 常量用 UPPER_CASE
MAX_FEATURES = 1000
GAUSSIAN_KERNEL_SIZE = (5, 5)
CONFIDENCE_THRESHOLD = 0.5

# ❌ 避免
img = cv2.imread("input.jpg")         # 太短，不清晰
IMG = cv2.imread("input.jpg")         # 图像不是常量
MyImage = cv2.imread("input.jpg")     # 不用 CamelCase

# ✅ 单一职责，清晰的输入输出
def preprocess_image(image: np.ndarray,
                     target_size: tuple = (640, 640),
                     normalize: bool = True) -> np.ndarray:
    """
    图像预处理流水线。

    参数:
        image: BGR 格式输入图像 (H, W, 3), dtype=uint8
        target_size: 目标尺寸 (W, H)
        normalize: 是否归一化到 [0, 1]

    返回:
        预处理后的 float32 图像
    """
    # 调整大小（保持比例 + 填充）
    h, w = image.shape[:2]
    scale = min(target_size[0] / w, target_size[1] / h)
    new_w, new_h = int(w * scale), int(h * scale)
    resized = cv2.resize(image, (new_w, new_h),
                          interpolation=cv2.INTER_LINEAR)

    # 填充到目标尺寸
    padded = np.full((*target_size[::-1], 3), 114, dtype=np.uint8)
    x_off = (target_size[0] - new_w) // 2
    y_off = (target_size[1] - new_h) // 2
    padded[y_off:y_off+new_h, x_off:x_off+new_w] = resized

    if normalize:
        return padded.astype(np.float32) / 255.0
    return padded

class ImageProcessor:
    """可复用的图像处理器"""

    def __init__(self, config: dict = None):
        self.config = config or {}
        self._initialize()

    def _initialize(self):
        """初始化资源（模型、滤波器等）"""
        self.gaussian_kernel = cv2.getGaussianKernel(5, 1.0)
        self.morph_kernel = cv2.getStructuringElement(
            cv2.MORPH_RECT, (5, 5)
        )

    def __call__(self, image: np.ndarray) -> np.ndarray:
        """使实例可调用"""
        return self.process(image)

    def process(self, image: np.ndarray) -> np.ndarray:
        """处理流水线"""
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        blurred = cv2.GaussianBlur(gray, (5, 5), 0)
        _, binary = cv2.threshold(blurred, 0, 255,
                                  cv2.THRESH_BINARY + cv2.THRESH_OTSU)
        cleaned = cv2.morphologyEx(binary, cv2.MORPH_CLOSE,
                                    self.morph_kernel)
        return cleaned

import cv2
import numpy as np

# ❌ 每次循环创建新对象
while True:
    result = np.zeros((1080, 1920, 3), dtype=np.uint8)  # 分配 + 清零
    # ...

# ✅ 预分配，复用
result = np.zeros((1080, 1920, 3), dtype=np.uint8)
while True:
    result[:] = 0  # 原地清零（更快）
    # 或 result.fill(0)
    # ...

# ✅ 使用引用（零拷贝）
roi = img[100:300, 200:400]         # 切片是视图，不拷贝
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)  # 输出是新数组

# ✅ 使用 inplace 操作
cv2.addWeighted(img1, 0.5, img2, 0.5, 0, dst=result)  # 写入 dst
cv2.GaussianBlur(img, (5, 5), 0, dst=blurred)          # 写入 blurred

# ❌ 不必要的拷贝
img_copy = img.copy()  # 只在需要独立副本时使用

import cv2
import numpy as np

# ❌ 逐张处理
for path in image_paths:
    img = cv2.imread(path)
    result = process(img)
    cv2.imwrite(path.replace("input", "output"), result)

# ✅ 批量处理（共享模型初始化）
detector = YOLODetector("model.onnx")  # 只加载一次

for path in image_paths:
    img = cv2.imread(path)
    detections = detector.detect(img)
    draw_detections(img, detections)

import cv2
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def process_image_async(path):
    loop = asyncio.get_event_loop()
    img = await loop.run_in_executor(executor, cv2.imread, path)
    result = await loop.run_in_executor(executor, process, img)
    await loop.run_in_executor(executor, cv2.imwrite,
                                path.replace("input", "output"), result)
    return result

# 批量异步处理
async def batch_process(paths):
    tasks = [process_image_async(p) for p in paths]
    results = await asyncio.gather(*tasks)
    return results

import cv2
import numpy as np
from multiprocessing import Pool

def process_worker(path):
    """工作进程"""
    img = cv2.imread(path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    edges = cv2.Canny(gray, 50, 150)
    output_path = path.replace("input", "output")
    cv2.imwrite(output_path, edges)
    return output_path

# 使用进程池
if __name__ == "__main__":
    with Pool(processes=8) as pool:
        results = pool.map(process_worker, image_paths)
    print(f"处理完成: {len(results)} 个文件")

"""
config.py — 项目配置管理
"""
from dataclasses import dataclass, field
from typing import Tuple
import yaml

@dataclass
class ModelConfig:
    path: str = "models/yolov8n.onnx"
    input_size: Tuple[int, int] = (640, 640)
    conf_threshold: float = 0.5
    iou_threshold: float = 0.4
    use_cuda: bool = False

@dataclass
class PreprocessConfig:
    normalize: bool = True
    mean: Tuple[float, float, float] = (0.485, 0.456, 0.406)
    std: Tuple[float, float, float] = (0.229, 0.224, 0.225)
    max_image_size: int = 4096

@dataclass
class AppConfig:
    model: ModelConfig = field(default_factory=ModelConfig)
    preprocess: PreprocessConfig = field(default_factory=PreprocessConfig)
    output_dir: str = "output/"
    log_level: str = "INFO"

def load_config(path: str = "config.yaml") -> AppConfig:
    with open(path) as f:
        data = yaml.safe_load(f)
    return AppConfig(**data)

# 使用
# config = load_config("config.yaml")
# print(config.model.conf_threshold)

"""
tests/test_preprocess.py
"""
import cv2
import numpy as np
import pytest

from cv_project.pipeline.preprocess import preprocess_image

@pytest.fixture
def sample_image():
    """创建测试图像"""
    img = np.random.randint(0, 256, (480, 640, 3), dtype=np.uint8)
    cv2.circle(img, (320, 240), 100, (255, 255, 255), -1)
    return img

def test_output_shape(sample_image):
    result = preprocess_image(sample_image, target_size=(640, 640))
    assert result.shape == (640, 640, 3)

def test_output_dtype(sample_image):
    result = preprocess_image(sample_image, normalize=True)
    assert result.dtype == np.float32

def test_normalized_range(sample_image):
    result = preprocess_image(sample_image, normalize=True)
    assert result.min() >= 0.0
    assert result.max() <= 1.0

def test_none_input():
    with pytest.raises(ValueError):
        preprocess_image(None)

def test_empty_image():
    empty = np.zeros((0, 0, 3), dtype=np.uint8)
    with pytest.raises(ValueError):
        preprocess_image(empty)

def test_grayscale_input():
    gray = np.zeros((480, 640), dtype=np.uint8)
    # 应自动转为三通道或抛出明确错误
    with pytest.raises(ValueError):
        preprocess_image(gray)

# 运行: pytest tests/ -v

# .github/workflows/ci.yml
name: CI

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10", "3.11", "3.12"]

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install dependencies
        run: |
          pip install --upgrade pip
          pip install -e ".[dev]"

      - name: Lint
        run: |
          ruff check src/ tests/

      - name: Test
        run: |
          pytest tests/ -v --cov=cv_project --cov-report=xml

      - name: Build Docker image
        run: |
          docker build -t cv-project:${{ github.sha }} .
          docker run --rm cv-project:${{ github.sha }} python -c "import cv2; print(cv2.__version__)"

# ONNX 模型优化
import onnxruntime as ort

# 使用 ONNX Runtime（比 OpenCV DNN 更快）
session = ort.InferenceSession(
    "model.onnx",
    providers=["CUDAExecutionProvider", "CPUExecutionProvider"]
)

# 输入输出信息
print("输入:", [(i.name, i.shape, i.type) for i in session.get_inputs()])
print("输出:", [(o.name, o.shape, o.type) for o in session.get_outputs()])

# 推理
import numpy as np
input_data = np.random.randn(1, 3, 640, 640).astype(np.float32)
outputs = session.run(None, {"input": input_data})

# model_manifest.yaml
models:
  - name: yolov8n
    version: "1.1"
    path: yolov8n_v1.1.onnx
    input_size: [640, 640]
    classes: 80
    metrics:
      mAP: 37.3
      latency_ms: 15.2

  - name: resnet50
    version: "2.0"
    path: resnet50_v2.0.onnx
    input_size: [224, 224]
    classes: 1000
    metrics:
      top1_acc: 76.1
      latency_ms: 8.5

"""
scripts/benchmark.py — 性能基准测试
"""
import cv2
import numpy as np
import time
import json

def benchmark_pipeline(image_sizes, n_iterations=100):
    results = []
    for size_name, (w, h) in image_sizes.items():
        img = np.random.randint(0, 256, (h, w, 3), dtype=np.uint8)

        # 预热
        for _ in range(5):
            process(img)

        # 计时
        times = []
        for _ in range(n_iterations):
            start = time.perf_counter()
            process(img)
            times.append((time.perf_counter() - start) * 1000)

        times = np.array(times)
        result = {
            "size": size_name,
            "resolution": f"{w}x{h}",
            "mean_ms": float(times.mean()),
            "median_ms": float(np.median(times)),
            "p95_ms": float(np.percentile(times, 95)),
            "p99_ms": float(np.percentile(times, 99)),
            "fps": float(1000 / times.mean()),
        }
        results.append(result)
        print(f"{size_name}: {result['mean_ms']:.1f}ms, {result['fps']:.1f} FPS")

    return results

if __name__ == "__main__":
    sizes = {
        "720p": (1280, 720),
        "1080p": (1920, 1080),
        "4K": (3840, 2160),
    }
    results = benchmark_pipeline(sizes)
    with open("benchmark_results.json", "w") as f:
        json.dump(results, f, indent=2)

def validate_upload(file_bytes: bytes, max_size_mb: int = 10):
    """上传文件安全验证"""
    # 1. 大小检查
    if len(file_bytes) > max_size_mb * 1024 * 1024:
        raise ValueError(f"文件超过 {max_size_mb}MB 限制")

    # 2. 格式检查（魔术字节）
    magic_bytes = {
        b'\xff\xd8\xff': 'jpg',
        b'\x89PNG': 'png',
        b'BM': 'bmp',
    }
    detected = None
    for magic, fmt in magic_bytes.items():
        if file_bytes[:len(magic)] == magic:
            detected = fmt
            break

    if detected is None:
        raise ValueError("不支持的文件格式")

    # 3. 解码验证
    nparr = np.frombuffer(file_bytes, np.uint8)
    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
    if img is None:
        raise ValueError("无法解码图像")

    # 4. 尺寸检查
    h, w = img.shape[:2]
    if w > 8192 or h > 8192:
        raise ValueError("图像尺寸过大")

    return img