Dự án: đếm sản phẩm trên băng tải bằng OpenCV, không đếm trùng

Chương trình chính. Đặt HIEN_THI = True để xem trực tiếp (nhấn q để dừng); mặc định tắt để chạy nhanh hết video. Phần cuối chạy lại hai cấu hình "sai có chủ đích" để thấy vai trò của closing và tracker.

import csv
import os
import time

import cv2
import numpy as np

CAU_HINH = {
    "video": "bang_tai.mp4",
    "roi": (0, 10, 640, 220),          # x, y, w, h của dải băng tải
    "nguong": 110,                     # ngưỡng cố định: băng tải ~45-70, sản phẩm ~135-210
    "k_mo": 5,                         # opening xóa nhiễu
    "k_dong": (15, 5),                 # closing ngang: nối hai nửa hộp có băng keo (khe 7 px)
    "dien_tich_min": 1200,             # hộp nhỏ nhất 50x40 = 2000 px²
    "vach_x": 320,                     # vạch đếm (tọa độ ảnh)
    "max_dist": 40,                    # khoảng cách ghép tối đa (px)
    "max_mat": 3,                      # số khung được phép mất dấu
    "thu_muc_minh_chung": "minh_chung",
    "file_log": "ket_qua.csv",
}
HIEN_THI = False
THI_NGHIEM = True


def tach_vat(gray, ch):
    """Trả về danh sách (cx, cy, diện tích, (x, y, w, h)) theo tọa độ ảnh đầy đủ."""
    x, y, w, h = ch["roi"]
    roi = cv2.GaussianBlur(gray[y:y + h, x:x + w], (5, 5), 0)
    _, m = cv2.threshold(roi, ch["nguong"], 255, cv2.THRESH_BINARY)
    if ch["k_mo"] > 1:
        m = cv2.morphologyEx(m, cv2.MORPH_OPEN,
                             cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (ch["k_mo"], ch["k_mo"])))
    if ch["k_dong"]:
        m = cv2.morphologyEx(m, cv2.MORPH_CLOSE, cv2.getStructuringElement(cv2.MORPH_RECT, ch["k_dong"]))
    cs, _ = cv2.findContours(m, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    ds = []
    for c in cs:
        A = cv2.contourArea(c)
        if A < ch["dien_tich_min"]:
            continue
        M = cv2.moments(c)
        bx, by, bw, bh = cv2.boundingRect(c)
        ds.append((M["m10"] / M["m00"] + x, M["m01"] / M["m00"] + y, A, (bx + x, by + y, bw, bh)))
    return ds


class TrackerTam:
    """Theo dõi tâm vật: dự đoán theo vận tốc, ghép tham lam theo khoảng cách."""

    def __init__(self, max_dist, max_mat):
        self.tracks, self.id_tiep, self.max_dist, self.max_mat = {}, 1, max_dist, max_mat

    def cap_nhat(self, ds):
        cap = []
        for tid, t in self.tracks.items():
            px, py = t["x"] + t["vx"], t["y"]                     # vị trí dự đoán
            for j, v in enumerate(ds):
                d = float(np.hypot(v[0] - px, v[1] - py))
                if d < self.max_dist:
                    cap.append((d, tid, j))
        cap.sort()
        track_da_ghep, vat_da_ghep = set(), set()
        for d, tid, j in cap:                                     # cặp gần nhất được ghép trước
            if tid in track_da_ghep or j in vat_da_ghep:
                continue
            track_da_ghep.add(tid)
            vat_da_ghep.add(j)
            t = self.tracks[tid]
            t.update(x_truoc=t["x"], vx=ds[j][0] - t["x"], x=ds[j][0], y=ds[j][1], mat=0, vat=ds[j])
        for tid in list(self.tracks):
            if tid not in track_da_ghep:
                self.tracks[tid]["mat"] += 1
                if self.tracks[tid]["mat"] > self.max_mat:
                    del self.tracks[tid]
        for j, v in enumerate(ds):
            if j not in vat_da_ghep:                              # vật mới vào khung
                self.tracks[self.id_tiep] = dict(x=v[0], y=v[1], vx=0.0, x_truoc=None,
                                                 mat=0, da_dem=False, vat=v)
                self.id_tiep += 1
        return self.tracks


def luu_minh_chung(frame, vat, stt, ch):
    bx, by, bw, bh = vat[3]
    x0, y0 = max(bx - 20, 0), max(by - 20, 0)
    cat = frame[y0:by + bh + 20, x0:bx + bw + 20].copy()
    cv2.rectangle(cat, (bx - x0, by - y0), (bx - x0 + bw - 1, by - y0 + bh - 1), (0, 255, 0), 1)
    cv2.putText(cat, f"#{stt}", (3, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 0, 255), 1)
    ten = f"sp_{stt:04d}.jpg"
    cv2.imwrite(os.path.join(ch["thu_muc_minh_chung"], ten), cat)
    return ten


def chay(ch, ghi_log=True):
    cap = cv2.VideoCapture(ch["video"])
    if not cap.isOpened():
        raise FileNotFoundError(f"không mở được video {ch['video']}")
    fps = cap.get(cv2.CAP_PROP_FPS) or 10.0
    tracker = TrackerTam(ch["max_dist"], ch["max_mat"])
    dem, k, frame = [], -1, None
    if ghi_log:
        os.makedirs(ch["thu_muc_minh_chung"], exist_ok=True)
    while True:
        ok, frame_moi = cap.read()
        if not ok:
            break
        frame, k = frame_moi, k + 1
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        for tid, t in tracker.cap_nhat(tach_vat(gray, ch)).items():
            if (not t["da_dem"] and t["mat"] == 0 and t["x_truoc"] is not None
                    and t["x_truoc"] < ch["vach_x"] <= t["x"]):
                t["da_dem"] = True
                stt = len(dem) + 1
                ten_anh = luu_minh_chung(frame, t["vat"], stt, ch) if ghi_log else ""
                dem.append({"stt": stt, "id_track": tid, "khung": k, "thoi_gian_s": round(k / fps, 1),
                            "phut": int(k / fps // 60) + 1, "dien_tich": int(t["vat"][2]), "anh": ten_anh})
        if HIEN_THI:
            ve = frame.copy()
            cv2.line(ve, (ch["vach_x"], 0), (ch["vach_x"], ve.shape[0] - 1), (0, 255, 255), 2)
            for tid, t in tracker.tracks.items():
                mau = (0, 255, 0) if t["da_dem"] else (0, 0, 255)
                cv2.circle(ve, (int(t["x"]), int(t["y"])), 4, mau, -1)
                cv2.putText(ve, str(tid), (int(t["x"]) + 6, int(t["y"])), cv2.FONT_HERSHEY_SIMPLEX, 0.4, mau, 1)
            cv2.putText(ve, f"Dem: {len(dem)}", (10, 25), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
            cv2.imshow("Dem san pham", ve)
            if cv2.waitKey(1) & 0xFF == ord("q"):
                break
    cap.release()
    if HIEN_THI:
        cv2.destroyAllWindows()
    if ghi_log:
        with open(ch["file_log"], "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=["stt", "id_track", "khung", "thoi_gian_s", "phut", "dien_tich", "anh"])
            w.writeheader()
            w.writerows(dem)
    return dem, k + 1, fps


bat_dau = time.perf_counter()
dem, so_khung, fps = chay(CAU_HINH)
giay = time.perf_counter() - bat_dau
print(f"đã xử lý {so_khung} khung ({so_khung / fps:.0f} s video), đếm được {len(dem)} sản phẩm")
print(f"tốc độ xử lý: {so_khung / giay:.0f} khung/s (thay đổi theo máy)")
print("5 dòng log đầu:", [(d["stt"], d["thoi_gian_s"], d["dien_tich"]) for d in dem[:5]])

if THI_NGHIEM:
    for ten, thay_doi in [("không closing", {"k_dong": None}), ("max_dist = 8", {"max_dist": 8})]:
        d2, _, _ = chay(dict(CAU_HINH, **thay_doi), ghi_log=False)
        print(f"thí nghiệm {ten:14s}: đếm được {len(d2)}")

đã xử lý 1800 khung (180 s video), đếm được 119 sản phẩm
tốc độ xử lý: 1333 khung/s (thay đổi theo máy)
5 dòng log đầu: [(1, 4.7, 3530), (2, 4.8, 2388), (3, 6.2, 2744), (4, 10.2, 3789), (5, 11.0, 2814)]
thí nghiệm không closing : đếm được 139
thí nghiệm max_dist = 8  : đếm được 72

Đọc kết quả:

• Cấu hình đúng đếm đúng bằng đáp án ở mục 2. Tốc độ xử lý trên máy thông thường cao hơn nhiều lần mức 10 khung/s cần thiết, còn dư cho camera nhanh hơn hoặc ảnh lớn hơn.
• Không closing: mỗi hộp có băng keo thành hai vật, hai track, hai lần đếm: đếm thừa. Đây là lỗi kinh điển "một vật đếm hai lần do phản xạ hoặc nhãn tối".
• max_dist quá nhỏ (8 px): khi hộp mới vào khung chưa có vận tốc, dự đoán lệch 12 px nên không ghép được; track cũ bị xóa, track mới sinh ra không có vị trí trước đúng lúc qua vạch, nên đếm thiếu. Đếm trùng và đếm sót đều đến từ việc chọn tham số tracker không dựa trên số liệu (tốc độ băng tải, khoảng cách giữa vật).
• Điều kiện t["mat"] == 0 đảm bảo chỉ đếm khi vật thật sự được nhìn thấy ở khung hiện tại, và x_truoc luôn là vị trí ghép được gần nhất.

Thống kê theo từng phút (bài thực hành của dự án). Khối này chỉ đọc log CSV, nên chạy lại được bất cứ lúc nào:

import csv
from collections import Counter

import cv2
import numpy as np

with open("ket_qua.csv", encoding="utf-8") as f:
    log = list(csv.DictReader(f))
theo_phut = Counter(int(d["phut"]) for d in log)
cac_phut = list(range(1, max(theo_phut, default=0) + 1))

with open("thong_ke_phut.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.writer(f)
    w.writerow(["phut", "so_san_pham"])
    for p in cac_phut:
        w.writerow([p, theo_phut[p]])

print("phút | sản phẩm")
for p in cac_phut:
    print(f"{p:4d} | {theo_phut[p]:4d} " + "#" * (theo_phut[p] // 2))
if cac_phut:
    print(f"tổng {len(log)} | trung bình {len(log) / len(cac_phut):.1f} sp/phút | "
          f"cao nhất phút {max(cac_phut, key=lambda p: theo_phut[p])}")

# Biểu đồ cột bằng OpenCV để đính kèm báo cáo ca
bd = np.full((220, 60 + 80 * max(len(cac_phut), 1), 3), 255, np.uint8)
cao_max = max(theo_phut.values(), default=1)
for i, p in enumerate(cac_phut):
    h = int(150 * theo_phut[p] / cao_max)
    cv2.rectangle(bd, (40 + 80 * i, 180 - h), (90 + 80 * i, 180), (200, 120, 40), -1)
    cv2.putText(bd, str(theo_phut[p]), (48 + 80 * i, 172 - h), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 1)
    cv2.putText(bd, f"P{p}", (52 + 80 * i, 200), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 1)
cv2.imwrite("bieu_do_phut.png", bd)

phút | sản phẩm
   1 |   36 ##################
   2 |   52 ##########################
   3 |   31 ###############
tổng 119 | trung bình 39.7 sp/phút | cao nhất phút 2

Kiểm tra với đáp án. So tổng, so từng phút, và so khung đếm của từng sản phẩm (ghép theo thứ tự thời gian) để biết chương trình đếm sớm hay muộn bao nhiêu:

import csv
import json
from collections import Counter

with open("ket_qua.csv", encoding="utf-8") as f:
    log = list(csv.DictReader(f))
dap_an = json.load(open("bang_tai_dap_an.json", encoding="utf-8"))["san_pham"]

that = Counter(d["phut"] for d in dap_an)
do = Counter(int(d["phut"]) for d in log)
print("phút | đáp án | đếm được | lệch")
for p in sorted(set(that) | set(do)):
    print(f"{p:4d} | {that[p]:6d} | {do[p]:8d} | {do[p] - that[p]:+4d}")
print(f"tổng | {len(dap_an):6d} | {len(log):8d} | {len(log) - len(dap_an):+4d}")

if len(log) == len(dap_an):
    lech = Counter(int(a["khung"]) - b for a, b in zip(log, sorted(d["khung"] for d in dap_an)))
    print("độ lệch khung đếm so với đáp án:", dict(sorted(lech.items())))

phút | đáp án | đếm được | lệch
   1 |     36 |       36 |   +0
   2 |     52 |       52 |   +0
   3 |     31 |       31 |   +0
tổng |    119 |      119 |   +0
độ lệch khung đếm so với đáp án: {0: 111, 1: 8}

Số đếm khớp tuyệt đối theo phút. Một số sản phẩm được đếm lệch một khung so với đáp án: đáp án dùng tâm hình học của hộp, còn chương trình dùng trọng tâm mặt nạ sau làm mịn và morphology, có thể lệch vài phần pixel, đủ để vượt vạch sớm hoặc muộn một khung (0,1 s). Với thống kê theo phút, lệch này chỉ có ảnh hưởng khi sản phẩm qua vạch đúng ranh giới phút; với tín hiệu gửi PLC để gạt sản phẩm, 0,1 s ở 120 px/s tương đương 12 px, phải tính đến khi thiết kế vị trí cơ cấu gạt (chặng 5, bài trigger, encoder và cycle time).