3、Python基础回顾：面向对象编程、装饰器、多线程与多进程在视觉任务中的应用

各位同学，咱们直接进入正题。做边缘设备上的视觉应用，Python 基础不牢，后面会踩很多坑。今天聊的这三个东西——面向对象、装饰器、多线程与多进程，说白了就是你在嵌入式视觉战场上的三把武器。我当年刚入行时，觉得 Python 嘛，写个脚本跑通就行了。结果第一次做实时行人检测，帧率上不去，代码改得想哭。后来才明白，这些基础玩不转，根本没法做工程。

3.1 面向对象编程：把摄像头和检测器封装起来

面向对象编程，OOP，听起来高大上。其实核心就一句话：把数据和操作数据的方法打包在一起。在视觉任务里，你想想看，摄像头有分辨率、帧率这些属性，还有打开、关闭、读取帧这些操作。用类来封装，再合适不过。

我个人习惯，每个硬件模块都写成一个类。比如摄像头，我绝不会在全局代码里直接调 cv2.VideoCapture。为什么？因为项目一复杂，你到处改摄像头参数，改到后面自己都忘了哪个是哪个。

class Camera:
    def __init__(self, device_id=0, width=640, height=480):
        self.device_id = device_id
        self.width = width
        self.height = height
        self.cap = None

    def open(self):
        self.cap = cv2.VideoCapture(self.device_id)
        self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width)
        self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height)
        if not self.cap.isOpened():
            raise RuntimeError(f"摄像头 {self.device_id} 打不开，检查连接")

    def read(self):
        ret, frame = self.cap.read()
        if not ret:
            return None
        return frame

    def release(self):
        if self.cap:
            self.cap.release()

你看，这样一封装，摄像头初始化、读取、释放都在一个类里。我在项目中遇到过一个问题：多个检测器共用同一个摄像头，结果有人直接调 cap.read()，有人又调了 cap.release()，程序跑着跑着就崩了。用类封装后，加个引用计数，问题就解决了。

再说说继承。做行人检测，你可能会有不同的检测器：YOLO、SSD、或者传统的 HOG+SVM。它们都有共同的接口：加载模型、推理、后处理。这时候用基类定义接口，子类实现细节，代码就干净了。

我的建议：在嵌入式视觉项目里，把摄像头、检测器、显示模块都写成类。哪怕你只写一个脚本，也养成这个习惯。后面加功能、换硬件，你会感谢自己的。

3.2 装饰器：给函数加点“料”

装饰器，说白了就是一个函数，用来修改另一个函数的行为。在视觉任务里，最常见的场景就是计时、日志、权限检查。

举个例子，你写了一个行人检测函数，想知道它跑了多久。最笨的办法是在函数开头记时间，结尾算差值。但如果你有十个函数都要计时呢？每个都写一遍？太累了。装饰器就是干这个的。

import time

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 耗时 {end - start:.3f} 秒")
        return result
    return wrapper

@timer
def detect_pedestrians(frame):
    # 这里是检测逻辑
    time.sleep(0.1)  # 模拟耗时
    return []

嗯，这里要注意：装饰器里的 wrapper 函数，参数要用 *args 和 **kwargs，这样才能适配任意函数。我曾经犯过一个错，装饰器写死了参数个数，结果换了个函数就报错，排查了半天。

在边缘设备上，装饰器还有一个妙用：缓存推理结果。比如你检测到某帧没有行人，下一帧可能也没有，你可以用装饰器缓存结果，减少模型推理次数。当然，这要看具体场景，不能乱用。

避坑指南：我曾经在装饰器里直接修改了被装饰函数的返回值，结果调用方拿到的数据格式变了，导致后续处理全错。记住，装饰器最好只做“附加”操作，不要改变函数的核心行为。

3.3 多线程与多进程：让边缘设备跑起来

边缘设备算力有限，但我们可以通过并发来榨干每一滴性能。多线程和多进程，很多人搞混。我简单说：多线程适合 I/O 密集型任务，比如读取摄像头、网络请求；多进程适合 CPU 密集型任务，比如模型推理。

为什么？因为 Python 有 GIL（全局解释器锁），多线程在 CPU 密集任务上反而可能更慢。但在 I/O 等待时，GIL 会释放，所以多线程能提高效率。

在行人检测项目里，我常用的架构是：一个线程专门读摄像头，一个线程做检测，一个线程做显示。这样读帧和检测可以并行，不会因为检测慢了就卡住画面。

import threading
import queue

class Pipeline:
    def __init__(self, camera, detector):
        self.camera = camera
        self.detector = detector
        self.frame_queue = queue.Queue(maxsize=2)
        self.result_queue = queue.Queue(maxsize=2)

    def capture_thread(self):
        while True:
            frame = self.camera.read()
            if frame is None:
                break
            self.frame_queue.put(frame)

    def detect_thread(self):
        while True:
            frame = self.frame_queue.get()
            results = self.detector.detect(frame)
            self.result_queue.put((frame, results))

    def start(self):
        t1 = threading.Thread(target=self.capture_thread)
        t2 = threading.Thread(target=self.detect_thread)
        t1.start()
        t2.start()

你看，用队列来传递数据，线程之间解耦。我建议队列大小不要太大，2-3 就够了，否则延迟会很高。实时系统最怕延迟。

再说多进程。如果你要在边缘设备上同时跑多个模型，比如一个检测行人，一个检测车辆，那就用多进程。每个进程有自己的 GIL，可以真正并行。但要注意，进程间通信开销大，数据要序列化，所以别频繁传递大图片。

特性 多线程 多进程
适用场景 I/O 密集型（读摄像头、网络） CPU 密集型（模型推理）
GIL 影响 有，但 I/O 时释放 无，每个进程独立
数据共享 容易，共享内存 困难，需序列化
启动开销 小 大
崩溃影响 可能影响主进程 独立，不影响其他进程

我的经验：在树莓派或 Jetson Nano 上做实时检测，我通常用多线程架构。一个线程读帧，一个线程做预处理（缩放、归一化），一个线程跑模型推理。如果模型太大，推理时间超过 100ms，那就考虑用多进程，或者换更轻量的模型。

3.4 综合实战：一个简单的行人检测流水线

好了，我们把前面三个知识点串起来。写一个简单的行人检测流水线，用类封装摄像头和检测器，用装饰器计时，用多线程并行处理。

import cv2
import time
import threading
import queue

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__}: {time.time()-start:.3f}s")
        return result
    return wrapper

class PedestrianDetector:
    def __init__(self, model_path):
        self.net = cv2.dnn.readNet(model_path)

    @timer
    def detect(self, frame):
        # 简化的检测逻辑
        blob = cv2.dnn.blobFromImage(frame, 1/255.0, (416, 416))
        self.net.setInput(blob)
        outputs = self.net.forward()
        return outputs

class Pipeline:
    def __init__(self, camera_id=0):
        self.camera = Camera(camera_id)
        self.detector = PedestrianDetector("yolo.weights")
        self.frame_queue = queue.Queue(maxsize=2)
        self.result_queue = queue.Queue(maxsize=2)

    def capture(self):
        self.camera.open()
        while True:
            frame = self.camera.read()
            if frame is None:
                break
            self.frame_queue.put(frame)

    def process(self):
        while True:
            frame = self.frame_queue.get()
            results = self.detector.detect(frame)
            self.result_queue.put((frame, results))

    def run(self):
        t1 = threading.Thread(target=self.capture)
        t2 = threading.Thread(target=self.process)
        t1.start()
        t2.start()
        t1.join()
        t2.join()

if __name__ == "__main__":
    pipe = Pipeline()
    pipe.run()

这段代码虽然简单，但结构清晰。你想想看，如果以后要换摄像头，只需要改 Camera 类；要换检测器，只需要改 PedestrianDetector 类。这就是面向对象的好处。装饰器让你一眼就能看到每个函数的耗时，方便调优。多线程让读帧和检测并行，帧率自然就上去了。

嗯，今天就到这里。这些基础看似简单，但真正用好，需要你在项目中反复打磨。下一章我们会深入 YOLO 模型在边缘设备上的部署，到时候这些知识都会用上。

课后练习：把上面的 Pipeline 改成多进程版本。注意进程间如何传递图片数据。提示：可以用 multiprocessing.Queue，但图片要先压缩再传。

← 上一章 📖 返回目录 下一章 →