Tổ chức pipeline xử lý ảnh bằng hàm, dataclass và cấu hình JSON

Thử nghiệm thuật toán trong một file dài 300 dòng thì nhanh, nhưng mang lên trạm thật sẽ gặp đủ thứ: muốn đổi ngưỡng phải sửa code, không biết bước nào chậm, không chạy lại được trên ảnh cũ để tìm lỗi. Bài này dạy cách tổ chức pipeline xử lý ảnh như kỹ sư chuyên nghiệp: chia bước rõ ràng, tham số trong file cấu hình, kết quả có cấu trúc, đo thời gian từng bước. Đây là khung bạn sẽ dùng cho mọi dự án về sau.

1. Pipeline là gì?

Hầu hết chương trình kiểm tra có cùng một dòng chảy:

Ảnh vào → Tiền xử lý → Phân tích (tìm vật, đo) → Đánh giá (so tiêu chí) → Kết quả + Ảnh kết quả

Mỗi bước là một hàm: nhận dữ liệu từ bước trước, trả kết quả cho bước sau, không đọc biến toàn cục, không tự in ra màn hình. Nhờ vậy bạn thay thế, kiểm thử, đo thời gian từng bước độc lập.

2. Tham số vào một chỗ

Mọi con số "ma thuật" (ngưỡng 128, diện tích tối thiểu 500, ROI…) phải nằm trong cấu hình, không rải rác trong code. Dùng dataclass để có giá trị mặc định, gợi ý code trong VS Code và kiểm tra hợp lệ:

import json
from dataclasses import dataclass, asdict, field
from pathlib import Path


@dataclass
class CauHinh:
    roi: tuple = (0, 0, 640, 480)          # (x, y, w, h)
    lam_mo: int = 5                         # kích thước bộ lọc Gaussian, số lẻ
    nguong: int = 0                         # 0 = tự chọn bằng Otsu
    dien_tich_min: int = 300
    so_vat_dung: int = 4                    # số vật phải có trên khay
    ghi_chu: str = ""

    def kiem_tra(self):
        if self.lam_mo % 2 == 0 or self.lam_mo < 1:
            raise ValueError("lam_mo phải là số lẻ ≥ 1")
        if not 0 <= self.nguong <= 255:
            raise ValueError("nguong phải trong 0–255")
        return self

    @classmethod
    def doc(cls, duong_dan):
        p = Path(duong_dan)
        if not p.exists():
            p.write_text(json.dumps(asdict(cls()), indent=2, ensure_ascii=False), encoding="utf-8")
            return cls()
        du_lieu = json.loads(p.read_text(encoding="utf-8"))
        du_lieu["roi"] = tuple(du_lieu.get("roi", cls.roi))
        return cls(**du_lieu).kiem_tra()


cfg = CauHinh.doc("pipeline.json")
print(cfg)

CauHinh(roi=(0, 0, 640, 480), lam_mo=5, nguong=0, dien_tich_min=300, so_vat_dung=4, ghi_chu='')

cls(**du_lieu) sẽ báo lỗi ngay nếu file cấu hình có khóa gõ sai (ví dụ "nguog"), thay vì âm thầm dùng giá trị mặc định. Lỗi cấu hình phát hiện lúc khởi động tốt hơn nhiều so với kết quả sai lúc đang chạy.

3. Kết quả có cấu trúc

@dataclass
class KetQua:
    ok: bool
    so_vat: int
    ly_do: str = ""
    cac_vat: list = field(default_factory=list)      # mỗi vật: dict tâm, diện tích, hộp
    thoi_gian_ms: dict = field(default_factory=dict) # thời gian từng bước

Trả về một đối tượng thay vì một chuỗi "OK"/"NG" giúp bước sau (vẽ ảnh, lưu CSV, gửi PLC) có đủ thông tin mà không phải tính lại.

4. Các bước xử lý

import time
import cv2
import numpy as np


def tien_xu_ly(anh, cfg):
    x, y, w, h = cfg.roi
    vung = anh[y:y + h, x:x + w]
    xam = cv2.cvtColor(vung, cv2.COLOR_BGR2GRAY) if vung.ndim == 3 else vung
    return cv2.GaussianBlur(xam, (cfg.lam_mo, cfg.lam_mo), 0)


def tim_vat(xam, cfg):
    kieu = cv2.THRESH_BINARY + (cv2.THRESH_OTSU if cfg.nguong == 0 else 0)
    _, nhi_phan = cv2.threshold(xam, cfg.nguong, 255, kieu)
    contours, _ = cv2.findContours(nhi_phan, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    cac_vat = []
    for c in contours:
        dt = cv2.contourArea(c)
        if dt < cfg.dien_tich_min:
            continue
        m = cv2.moments(c)
        cac_vat.append({"tam": (round(m["m10"] / m["m00"]), round(m["m01"] / m["m00"])),
                        "dien_tich": round(dt), "hop": cv2.boundingRect(c)})
    return cac_vat


def danh_gia(cac_vat, cfg):
    if len(cac_vat) != cfg.so_vat_dung:
        return False, f"có {len(cac_vat)} vật, cần {cfg.so_vat_dung}"
    return True, ""

Những hàm threshold, findContours, moments sẽ học kỹ ở chặng 3. Ở đây hãy tập trung vào cấu trúc: mỗi hàm nhận dữ liệu + cấu hình, trả kết quả.

5. Hàm điều phối: chạy và đo thời gian

def kiem_tra(anh, cfg):
    tg = {}
    t = time.perf_counter()

    xam = tien_xu_ly(anh, cfg)
    tg["tien_xu_ly"] = (time.perf_counter() - t) * 1000; t = time.perf_counter()

    cac_vat = tim_vat(xam, cfg)
    tg["tim_vat"] = (time.perf_counter() - t) * 1000; t = time.perf_counter()

    ok, ly_do = danh_gia(cac_vat, cfg)
    tg["danh_gia"] = (time.perf_counter() - t) * 1000
    tg["tong"] = sum(tg.values())
    return KetQua(ok, len(cac_vat), ly_do, cac_vat, {k: round(v, 2) for k, v in tg.items()})


# Ảnh thử: khay có 4 vật sáng
anh = np.full((480, 640, 3), 50, np.uint8)
for x in (120, 260, 400, 540):
    cv2.circle(anh, (x, 240), 40, (210, 210, 210), -1)

kq = kiem_tra(anh, cfg)
print(kq.ok, kq.so_vat, kq.ly_do)
print([v["tam"] for v in kq.cac_vat])
print("Thời gian:", kq.thoi_gian_ms)

True 4
[(540, 240), (400, 240), (260, 240), (120, 240)]
Thời gian: {'tien_xu_ly': 1.21, 'tim_vat': 0.85, 'danh_gia': 0.0, 'tong': 2.06}

Thời gian thực tế tùy máy. Khi dây chuyền yêu cầu xử lý dưới 50 ms mà bạn đang ở 80 ms, bảng thời gian từng bước cho biết tối ưu chỗ nào thay vì đoán mò. Để ý thứ tự vật đang ngược (contour trả về theo thứ tự bất kỳ); nếu cần thứ tự từ trái sang phải, sắp xếp: sorted(cac_vat, key=lambda v: v["tam"][0]).

6. Chạy lại trên ảnh đã lưu (chế độ offline)

Pipeline tách rời camera cho phép chạy lại trên thư mục ảnh đã lưu: đây là cách bạn tìm nguyên nhân khi khách hàng báo "hôm qua trạm loại nhầm". Chỉ cần thêm một hàm duyệt thư mục:

def chay_thu_muc(thu_muc, cfg):
    tong, ng = 0, []
    for p in sorted(Path(thu_muc).glob("*.png")):
        anh = cv2.imread(str(p))
        if anh is None:
            continue
        kq = kiem_tra(anh, cfg)
        tong += 1
        if not kq.ok:
            ng.append((p.name, kq.ly_do))
    return tong, ng


Path("anh_thu").mkdir(exist_ok=True)
for i, so in enumerate([4, 4, 3, 4]):            # ảnh thứ 3 thiếu một vật
    a = np.full((480, 640, 3), 50, np.uint8)
    for x in (120, 260, 400, 540)[:so]:
        cv2.circle(a, (x, 240), 40, (210, 210, 210), -1)
    cv2.imwrite(f"anh_thu/khay_{i}.png", a)

print(chay_thu_muc("anh_thu", cfg))

(4, [('khay_2.png', 'có 3 vật, cần 4')])

Khi sửa thuật toán, chạy lại trên bộ ảnh này để chắc chắn sửa lỗi mới không làm hỏng trường hợp cũ. Đây chính là kiểm thử hồi quy, và bạn sẽ dùng một bộ "ảnh vàng" như vậy khi nghiệm thu trạm (chặng 5).

7. Cấu trúc thư mục dự án gợi ý

tram_dem_vat/
├── pipeline.json          ← cấu hình (người vận hành chỉnh được)
├── pipeline.py            ← CauHinh, KetQua, các bước, kiem_tra()
├── ve.py                  ← vẽ ảnh kết quả (bài trước)
├── chay_camera.py         ← vòng lặp camera, gọi kiem_tra()
├── chay_thu_muc.py        ← chạy lại offline trên ảnh đã lưu
└── anh_vang/              ← bộ ảnh kiểm thử có kết quả biết trước

pipeline.py không biết gì về camera hay giao diện; chay_camera.py không biết gì về thuật toán. Tách như vậy, khi đổi camera hay đổi thuật toán bạn chỉ sửa một file.

Tóm tắt

• Pipeline = chuỗi hàm: tiền xử lý → phân tích → đánh giá → kết quả.
• Tham số trong file JSON, đọc qua dataclass có kiểm tra hợp lệ.
• Kết quả là đối tượng có cấu trúc, kèm lý do NG và thời gian từng bước.
• Tách thuật toán khỏi camera để chạy lại offline và kiểm thử hồi quy.

Bài tập

@dataclass
class CauHinh2(CauHinh):
    dien_tich_vat_min: int = 4000
    dien_tich_vat_max: int = 6000


def danh_gia2(cac_vat, cfg):
    if len(cac_vat) != cfg.so_vat_dung:
        return False, f"có {len(cac_vat)} vật, cần {cfg.so_vat_dung}"
    for v in sorted(cac_vat, key=lambda v: v["tam"][0]):
        if not cfg.dien_tich_vat_min <= v["dien_tich"] <= cfg.dien_tich_vat_max:
            return False, f"vật tại {v['tam']} có diện tích {v['dien_tich']}"
    return True, ""


cfg2 = CauHinh2()
print(danh_gia2(tim_vat(tien_xu_ly(anh, cfg2), cfg2), cfg2))

(True, '')