⭐ Vision Engineer · Chặng 14Bài 3/5Dự án~480 phút

Capstone: kiểm tra lỗi bề mặt bằng AI, ONNX real-time, PLC và giám sát drift

Dự án portfolio: dữ liệu chia theo lô, huấn luyện CNN, chọn ngưỡng theo escape, xuất ONNX kiểm sai số, trạm ONNX Runtime trả PLC, lưu ảnh NG, giám sát drift.

Capstone: kiểm tra lỗi bề mặt bằng AI, ONNX real-time, PLC và giám sát drift

Học xong bài này, bạn sẽ

  • Dữ liệu, huấn luyện, chọn ngưỡng
  • Triển khai ONNX real-time
  • Tích hợp PLC, lưu ảnh NG, giám sát

Nên học trước: Capstone 2: trạm đọc DataMatrix và số seri

Nội dung bài viết
  1. 1. Bối cảnh
  2. 2. Yêu cầu và tiêu chí nghiệm thu
  3. 3. Kiến thức sử dụng
  4. 4. Kiến trúc tham chiếu
  5. 5. Lộ trình thực hiện theo mốc
  6. 6. Mã khởi đầu và phần lõi
  7. 7. Kiểm thử và nghiệm thu
  8. 8. Sản phẩm bàn giao cho portfolio
  9. 9. Tự đánh giá
  10. 10. Mở rộng
  11. Tóm tắt

Dự án cuối cùng ghép mọi chặng lại: một trạm kiểm tra lỗi bề mặt bằng AI chạy thật trên dây chuyền, từ dữ liệu tới PLC. Bạn sẽ tạo dữ liệu có chia theo lô vật liệu, huấn luyện một CNN nhỏ, chọn ngưỡng theo ràng buộc escape trên tập validation, đánh giá một lần trên tập test, xuất ONNX và kiểm tra sai số, chạy ONNX Runtime trong vòng lặp có đo thời gian từng bước, trả kết quả cho PLC qua Modbus TCP, lưu ảnh NG, và cuối cùng là phần mà đa số dự án AI trên mạng bỏ qua: giám sát sau khi triển khai, phát hiện khi camera mất nét trước khi trạm bắt đầu lọt lỗi. Dự án dùng kiến thức của các chặng Deep Learning, AI Vision, Detection / Segmentation / Anomaly Detection, Deployment và PLC / Robot / MES; hai dự án trước (trạm đo kích thước, trạm đọc mã) đã cho bạn khung tích hợp PLC và cách viết biên bản nghiệm thu.

Mục tiêu dự án(1) Dữ liệu, huấn luyện, chọn ngưỡng theo tiêu chí nhà máy. (2) Triển khai ONNX real-time, đo latency p50/p99. (3) Tích hợp PLC, lưu ảnh NG, giám sát trôi dữ liệu và tỉ lệ NG. Thời lượng dự kiến: khoảng 8 giờ.

1. Bối cảnh

Một nhà máy gia công vỏ thiết bị âm thanh làm nắp nhôm xước sợi (brushed aluminium, bề mặt có vân xước song song do đánh bóng bằng băng nhám) kích thước 120 × 80 mm, sau đó anod hóa. Lỗi cần bắt trước khi anod hóa: vết xước cắt ngang vân (dài từ khoảng 1 mm) và đốm (vết lõm, bụi kẹt, đường kính từ khoảng 0,2 mm). Kiểm tra bằng mắt hiện bỏ sót nhiều vì vân xước sợi "che" lỗi, và hai ca đánh giá rất khác nhau. Thử rule-based (ngưỡng, lọc hình thái) cho overkill cao vì độ sáng và hướng vân thay đổi theo lô nhôm: đây đúng là trường hợp bài "Rule-based hay Deep Learning" khuyên chuyển sang AI.

Thông sốGiá trị
Nhịp dây chuyền30 nắp/phút (2 s/nắp); PLC cho trạm 500 ms từ TRIGGER tới DONE
Quang học (trạm thật)Camera đơn sắc khoảng 5 MP, đèn chiếu xiên (low angle) để vết xước nổi lên, khoảng 0,05 mm/px
Cách kiểm traĐịnh vị nắp bằng rule-based, chia vùng kiểm tra thành các ô, mô hình chấm điểm NG từng ô; nắp NG nếu có ít nhất một ô NG
Tỉ lệ lỗi thậtKhoảng 3% số nắp
Dữ liệu có sẵnẢnh của 10 lô nhôm, đã được QC gán nhãn OK/NG (lô 1–6 cũ nhất, lô 9–10 mới nhất)
PLCModbus TCP, cùng bảng địa chỉ kiểu Capstone 1; thêm điểm NG (float32) và phiên bản mô hình
Phiên bản mô phỏng trong bàiĐể cả trang chạy trong khoảng 1 phút trên CPU, mỗi "sản phẩm" trong demo là một ô 48 × 48 px tổng hợp, và mô hình chỉ có khoảng 6 nghìn tham số. Trạm thật dùng ô lớn hơn (thường 128–256 px) và mô hình lớn hơn, thường là backbone đã huấn luyện sẵn như bài "Phân loại ảnh OK/NG bằng transfer learning với PyTorch". Quy trình (chia theo lô, chọn ngưỡng, ONNX, đo thời gian, PLC, giám sát) giữ nguyên.

2. Yêu cầu và tiêu chí nghiệm thu

MãYêu cầuTiêu chí đạtCách thử
R1Chia dữ liệu chống rò rỉTrain / val / test không chung lô vật liệuKiểm tra tập lô
R2Không lọt lỗiEscape = 0 trên tập test (2 lô chưa từng thấy)Ngưỡng chọn trên val, áp một lần lên test
R3Loại nhầm trong ngân sáchOverkill ≤ 2% trên testNhư R2
R4ONNX tương đương PyTorchSai lệch điểm ≤ 1e-4, không quyết định nào bị lật600 ô val + test
R5Tốc độSuy luận một nắp (lô 64 ô) p99 ≤ 50 ms trên PC trạmĐo 30 lần
R6Tích hợp PLCPLC nhận đúng kết quả 60/60 chu kỳ, kể cả chu kỳ camera không trả ảnh (LỖI)PLC giả, kịch bản biết trước
R7Lưu ảnh NGMọi ô NG có ảnh trên đĩa, tên file có chu kỳ và điểmĐếm file
R8Giám sátCảnh báo trôi dữ liệu xuất hiện trước escape đầu tiên khi camera mất nét dầnMô phỏng 2000 sản phẩm
R9Truy vết mô hìnhFile metadata đi kèm ONNX: phiên bản, ngưỡng, tiền xử lý, lô dùng để huấn luyện; PLC và CSDL nhận số phiên bảnXem xét

3. Kiến thức sử dụng

  • Deep Learning: bài "PyTorch cơ bản: tensor, autograd và vòng lặp huấn luyện"; bài "Overfitting và cách chia train/validation/test đúng" (vì sao chia theo lô); bài "Phân loại ảnh OK/NG bằng transfer learning với PyTorch".
  • AI Vision: bài "Rule-based hay Deep Learning", bài "Tự xây dataset ảnh công nghiệp", bài "Dữ liệu tổng hợp (synthetic data) khi thiếu ảnh lỗi", bài "Precision, Recall, F1 và ma trận nhầm lẫn", và bài chọn ngưỡng, đánh giá theo nhà máy của chặng này.
  • Detection / Segmentation / Anomaly Detection: dự án "Phát hiện lỗi bề mặt bằng Anomaly Detection (PatchCore)", phương án khi gần như không có ảnh lỗi (mục 10).
  • Deployment: các bài xuất ONNX và kiểm tra sai số, đo latency, logging và giám sát, đóng gói ứng dụng chạy 24/7; bài "Tăng tốc suy luận: ONNX Runtime, OpenVINO, TensorRT".
  • Industrial Vision và PLC / Robot / MES: escape/overkill, quy tắc 3/n, Modbus TCP, ảnh NG theo ngày.
pip install torch onnx onnxscript onnxruntime opencv-python numpy pymodbus
# Bài viết chạy với torch 2.14 (CPU), onnx 1.23, onnxscript 0.7, onnxruntime 1.31, OpenCV 5.0, pymodbus 3.15.0

4. Kiến trúc tham chiếu

Hai thế giới tách biệt: máy huấn luyện (PyTorch, dữ liệu, notebook) và máy trạm (chỉ ONNX Runtime, OpenCV, pymodbus). Thứ duy nhất đi từ bên này sang bên kia là cặp file mo_hinh.onnx + mo_hinh.json. Máy trạm không cài PyTorch: ít phụ thuộc hơn, khởi động nhanh hơn, và không ai "sửa nhanh" mô hình trên dây chuyền được.

MÁY HUẤN LUYỆN                                   MÁY TRẠM (24/7)
du_lieu.py → du_lieu.npz (ảnh, nhãn, lô)         ┌───────────────────────────────────────┐
huan_luyen.py → mo_hinh.pt                       │ PLC ─TRIGGER─► vòng lặp trạm           │
chon_nguong.py (val) → NGUONG                    │   chụp → cắt ô → BoSuyLuan.diem()      │
danh_gia (test, MỘT lần)                         │   → OK/NG/LỖI → ghi HR (1 lệnh) → DONE │
xuat_onnx.py → mo_hinh.onnx + mo_hinh.json ════► │   → SQLite + ảnh NG (anh_ng/ngày/)     │
                                                 │   → GiamSat: độ sáng, độ nét, tỉ lệ NG │
                                                 └──────────────┬────────────────────────┘
                                                                ▼ cảnh báo HMI / email / log

Cấu trúc repo đề xuất:

tram-ai-be-mat/
├── README.md, MODEL_CARD.md       mô hình học gì, từ lô nào, ngưỡng, giới hạn đã biết
├── train/                         du_lieu.py, huan_luyen.py, chon_nguong.py, xuat_onnx.py
├── models/v3/                     mo_hinh.onnx, mo_hinh.json  (mỗi phiên bản một thư mục)
├── station/
│   ├── suy_luan.py                BoSuyLuan: chỉ onnxruntime + numpy
│   ├── plc.py, luu_tru.py         như Capstone 1
│   ├── giam_sat.py                EWMA độ sáng/độ nét, tỉ lệ NG hai phía
│   └── main.py                    cấu hình, log xoay vòng, tự kết nối lại
├── eval/                          bao_cao.py → biên bản
└── deploy/                        dịch vụ chạy 24/7 (systemd/Windows service), Dockerfile

Bảng địa chỉ Modbus: coil 0 TRIGGER (PLC), coil 1–3 READY/BUSY/DONE (vision); HR 0–1 mã sản phẩm, số chu kỳ (PLC); HR 10 kết quả (1 OK, 2 NG, 3 LỖI), HR 11 chu kỳ trả lại, HR 12 mã lỗi (bit 0 NG do AI, bit 8 lỗi ảnh), HR 13 phiên bản mô hình, HR 14–15 điểm NG lớn nhất (float32, −1 khi không có ảnh). Gửi phiên bản mô hình cho PLC để mọi sản phẩm trong MES biết được kiểm bằng mô hình nào.

5. Lộ trình thực hiện theo mốc

MốcViệc chínhSản phẩmXong khiThời gian
M1Đặc tả lỗi bằng con số, bảng địa chỉ PLC, ngân sách thời gian; dữ liệu và cách chia theo lôdocs/dac_ta.md, du_lieu.npzĐịnh nghĩa được "lỗi nhỏ nhất phải bắt"; tập test là các lô mới nhất, chưa ai nhìn60 phút
M2Huấn luyện baseline, theo dõi loss và valhuan_luyen.py, mo_hinh.ptVal ổn định qua các epoch; ghi lại seed, phiên bản thư viện90 phút
M3Chọn ngưỡng trên val theo escape, đánh giá test một lầnchon_nguong.py, bảng ngưỡngR1–R3 đạt; giải thích được vì sao chọn ngưỡng đó60 phút
M4Xuất ONNX, kiểm tra sai số, metadata; lớp suy luận không phụ thuộc PyTorchmo_hinh.onnx, mo_hinh.jsonR4, R5, R9 đạt60 phút
M5Vòng lặp trạm: PLC giả, ảnh NG, SQLite, đo thời gian từng bướcstation/R6, R7 đạt; chạy lại 3 lần không treo, không lỗi cổng90 phút
M6Giám sát drift và tỉ lệ NG; thử bằng kịch bản mất nétgiam_sat.pyR8 đạt; ngưỡng cảnh báo không báo nhầm trong giai đoạn bình thường60 phút
M7Biên bản, model card, README, videoRepo hoàn chỉnhNgười khác chạy lại được từ đầu và ra cùng biên bản60 phút

6. Mã khởi đầu và phần lõi

6.1. Mã khởi đầu: bề mặt xước sợi tổng hợp, chia theo lô

Vân xước sợi được tạo bằng nhiễu làm mờ rất lệch (σ 4 px theo một chiều, 0,5 px theo chiều kia) rồi xoay theo hướng vân của lô. Mỗi lô có độ sáng, độ thô của vân, hướng vân và độ dốc chiếu sáng riêng: đó là lý do phải chia theo lô. Nếu trộn ngẫu nhiên, ô của cùng một lô nằm ở cả train và test, mô hình được "xem trước" lô test và con số đánh giá sẽ đẹp hơn thực tế. Tập dữ liệu được làm giàu lỗi (35%) để có đủ mẫu NG; dây chuyền thật chỉ khoảng 3%.

# ===== du_lieu.py — bề mặt nhôm xước sợi tổng hợp, chia theo lô, có nhãn và loại lỗi =====
import math

import cv2
import numpy as np

S = 48                                  # kích thước một ô ảnh (px)


def thong_so_lo(lo):
    """Mỗi lô vật liệu có độ sáng, độ thô sợi, hướng sợi và độ dốc chiếu sáng hơi khác nhau."""
    r = np.random.default_rng(1000 + lo)
    return {"sang": r.uniform(125, 150), "tho": r.uniform(5, 8), "huong": r.uniform(-10, 10),
            "doc": r.uniform(-12, 12, 2)}


def ve_o(rng, p, loi=None, do_sang=1.0):
    """Một ô 48 × 48: vân xước sợi + độ dốc sáng + nhiễu; loi ∈ {None, 'xuoc', 'dom'}."""
    B = S + 24
    van = cv2.GaussianBlur(rng.normal(0, 1, (B, B)).astype(np.float32), (0, 0), sigmaX=4.0, sigmaY=0.5)
    M = cv2.getRotationMatrix2D((B / 2, B / 2), p["huong"] + rng.uniform(-3, 3), 1.0)
    van = cv2.warpAffine(van, M, (B, B))[12:12 + S, 12:12 + S]
    van /= van.std() + 1e-6
    yy, xx = np.mgrid[0:S, 0:S] / S - 0.5
    anh = p["sang"] + p["tho"] * van + p["doc"][0] * xx + p["doc"][1] * yy + rng.normal(0, 2.5, (S, S))
    if loi is not None:
        lop = np.zeros((S, S), np.float32)
        if loi == "xuoc":                              # vết xước mảnh, dài 12–30 px, hướng bất kỳ
            a, L = rng.uniform(0, math.pi), rng.uniform(12, 30)
            cx, cy = rng.uniform(12, 36, 2)
            d = (L / 2 * math.cos(a), L / 2 * math.sin(a))
            cv2.line(lop, (int(cx - d[0]), int(cy - d[1])), (int(cx + d[0]), int(cy + d[1])), 1.0, 1, cv2.LINE_AA)
            anh -= rng.uniform(28, 50) * lop
        else:                                          # đốm (lõm, bụi kẹt) bán kính 2–3 px
            c = tuple(int(v) for v in rng.uniform(6, 42, 2))
            cv2.circle(lop, c, int(rng.integers(2, 4)), 1.0, -1, cv2.LINE_AA)
            anh -= rng.uniform(30, 55) * cv2.GaussianBlur(lop, (0, 0), 0.7)
    return np.clip(anh * do_sang, 0, 255).astype(np.uint8)


def tao_lo(lo, so_o, ti_le_loi, do_sang=1.0, seed=None):
    rng = np.random.default_rng(lo if seed is None else seed)
    p = thong_so_lo(lo)
    X, y, loai = [], [], []
    for _ in range(so_o):
        loi = None
        if rng.random() < ti_le_loi:
            loi = "xuoc" if rng.random() < 0.6 else "dom"
        X.append(ve_o(rng, p, loi, do_sang))
        y.append(int(loi is not None))
        loai.append(loi or "ok")
    return np.array(X), np.array(y), np.array(loai)


if __name__ == "__main__":
    # 10 lô × 150 ô, làm giàu lỗi 35% (dây chuyền thật chỉ vài %: xem mục 7)
    X, y, lo, loai = [], [], [], []
    for k in range(1, 11):
        a, b, c = tao_lo(k, 150, 0.35)
        X.append(a), y.append(b), loai.append(c), lo.append(np.full(len(b), k))
    X, y, lo, loai = map(np.concatenate, (X, y, lo, loai))
    np.savez_compressed("du_lieu.npz", X=X, y=y, lo=lo, loai=loai)
    tam = np.vstack([np.hstack(list(X[lo == k][:10])) for k in (1, 5, 9)])
    cv2.imwrite("mau_be_mat.png", cv2.resize(tam, None, fx=3, fy=3, interpolation=cv2.INTER_NEAREST))
    for ten, tap in (("train", [1, 2, 3, 4, 5, 6]), ("val", [7, 8]), ("test", [9, 10])):
        m = np.isin(lo, tap)
        print(f"{ten:5s}: lô {tap} | {m.sum()} ô | lỗi {y[m].sum()} (xước {(loai[m] == 'xuoc').sum()}, "
              f"đốm {(loai[m] == 'dom').sum()})")
train: lô [1, 2, 3, 4, 5, 6] | 900 ô | lỗi 305 (xước 190, đốm 115)
val  : lô [7, 8] | 300 ô | lỗi 92 (xước 57, đốm 35)
test : lô [9, 10] | 300 ô | lỗi 89 (xước 51, đốm 38)

Mở mau_be_mat.png: ba hàng là ba lô khác nhau. Một số vết xước gần song song với vân rất khó thấy bằng mắt, giống thực tế.

6.2. Khung chương trình bạn cần viết

tien_xu_ly(X_uint8) -> float32 (N, 1, 48, 48)     DÙNG CHUNG cho huấn luyện và trạm, ghi vào metadata
MangNho(nn.Module): forward(x) -> logit (N,)       sigmoid(logit) = điểm NG trong [0, 1]
huan_luyen: tăng cường (xoay 90°, lật), Adam, giảm lr dạng cosine, lưu state_dict
chon_nguong(diem_val, y_val) -> NGUONG           escape val = 0 VÀ overkill val ≤ 2%, ưu tiên ngưỡng thấp
xuat_onnx(mo_hinh) -> mo_hinh.onnx + mo_hinh.json; kiểm sai số trên 600 ô
BoSuyLuan(onnx, json).diem(cac_o) -> điểm NG      chỉ numpy + onnxruntime
vòng lặp trạm: TRIGGER↑ → chụp → điểm → ghi HR → DONE → lưu → chờ TRIGGER↓
GiamSat.cap_nhat(i, anh, ng)                       EWMA độ sáng, EWMA log độ nét, tỉ lệ NG hai phía

6.3. Lời giải phần lõi

Xem lời giải phần lõi (4 phần, chạy nối tiếp sau khối 6.1)

Phần 1: huấn luyện. Mạng 3 khối conv + BatchNorm, kết thúc bằng max pooling toàn ảnh: một vết lỗi nhỏ ở bất kỳ đâu cũng kích hoạt đầu ra, trong khi average pooling sẽ "pha loãng" nó với phần còn lại của ô. Tiền xử lý trừ trung bình từng ảnh (bỏ khác biệt độ sáng giữa các lô) nhưng chia cho một hằng số thay vì độ lệch chuẩn của ảnh (giữ thông tin tương phản của vết lỗi).

# ===== huan_luyen.py — CNN nhỏ phân loại ô OK/NG, chia train/val/test THEO LÔ =====
import time

import numpy as np
import torch
import torch.nn as nn

torch.manual_seed(0)
torch.set_num_threads(1)        # mô hình rất nhỏ: 1 luồng thường nhanh hơn nhiều luồng (xem chú thích bên dưới)

d = np.load("du_lieu.npz")
X, y, lo = d["X"], d["y"], d["lo"]
TAP = {"train": [1, 2, 3, 4, 5, 6], "val": [7, 8], "test": [9, 10]}
chia = {k: np.isin(lo, v) for k, v in TAP.items()}


class MangNho(nn.Module):
    """3 khối conv; gộp bằng MAX toàn ảnh: lỗi nhỏ, nằm ở đâu cũng được "nhìn thấy"."""

    def __init__(self):
        super().__init__()
        def khoi(vao, ra):
            return [nn.Conv2d(vao, ra, 3, padding=1), nn.BatchNorm2d(ra), nn.ReLU()]
        self.dac_trung = nn.Sequential(*khoi(1, 8), nn.MaxPool2d(2), *khoi(8, 16), nn.MaxPool2d(2),
                                       *khoi(16, 32), nn.AdaptiveMaxPool2d(1))
        self.dau_ra = nn.Linear(32, 1)

    def forward(self, x):
        return self.dau_ra(self.dac_trung(x).flatten(1)).squeeze(1)       # logit; sigmoid → điểm NG


def tien_xu_ly(X):
    """uint8 (N, 48, 48) → float32 (N, 1, 48, 48): trừ trung bình từng ảnh, chia hằng số 32.
    Hằng số (không chia độ lệch chuẩn) để giữ thông tin tương phản; ghi vào metadata khi triển khai."""
    x = X.astype(np.float32)[:, None]
    return (x - x.mean(axis=(2, 3), keepdims=True)) / 32.0


def diem_ng(mo_hinh, X):
    mo_hinh.eval()
    with torch.no_grad():
        return torch.sigmoid(mo_hinh(torch.from_numpy(tien_xu_ly(X)))).numpy()


mo_hinh = MangNho()
SO_EPOCH = 25
opt = torch.optim.Adam(mo_hinh.parameters(), lr=3e-3)
lich = torch.optim.lr_scheduler.CosineAnnealingLR(opt, SO_EPOCH)
ham_mat = nn.BCEWithLogitsLoss()
xt, yt = torch.from_numpy(tien_xu_ly(X[chia["train"]])), torch.from_numpy(y[chia["train"]]).float()
t0 = time.time()
for ep in range(1, SO_EPOCH + 1):
    mo_hinh.train()
    hoan_vi, tong = torch.randperm(len(xt)), 0.0
    for i in range(0, len(xt), 64):
        idx = hoan_vi[i:i + 64]
        xb = torch.rot90(xt[idx], int(torch.randint(0, 4, (1,))), (2, 3))   # tăng cường: xoay 90°, lật
        if torch.rand(1) < 0.5:
            xb = torch.flip(xb, (3,))
        mat = ham_mat(mo_hinh(xb), yt[idx])
        opt.zero_grad()
        mat.backward()
        opt.step()
        tong += mat.item() * len(idx)
    lich.step()
    if ep % 5 == 0:
        s = diem_ng(mo_hinh, X[chia["val"]])
        dung = ((s > 0.5) == y[chia["val"]]).mean()
        print(f"epoch {ep:2d}: loss train {tong / len(xt):.3f} | val đúng {dung:.3f} (ngưỡng tạm 0,5)")
torch.save(mo_hinh.state_dict(), "mo_hinh.pt")
print(f"Đã lưu mo_hinh.pt, {sum(p.numel() for p in mo_hinh.parameters())} tham số, "
      f"huấn luyện {time.time() - t0:.0f} s (tùy máy)")
epoch  5: loss train 0.042 | val đúng 0.993 (ngưỡng tạm 0,5)
epoch 10: loss train 0.017 | val đúng 0.990 (ngưỡng tạm 0,5)
epoch 15: loss train 0.008 | val đúng 0.997 (ngưỡng tạm 0,5)
epoch 20: loss train 0.011 | val đúng 0.993 (ngưỡng tạm 0,5)
epoch 25: loss train 0.009 | val đúng 0.993 (ngưỡng tạm 0,5)
Đã lưu mo_hinh.pt, 6033 tham số, huấn luyện 22 s (tùy máy)

Con số loss và độ đúng có thể lệch chút ít giữa các CPU và phiên bản thư viện; thời gian tùy máy. Về torch.set_num_threads(1): với mô hình vài nghìn tham số và ảnh 48 × 48, khi viết bài này, chạy 4 luồng trên máy đang có tải chậm hơn 1 luồng khoảng 10 lần, vì chi phí đồng bộ luồng lớn hơn phần tính toán. Luôn đo trước khi tin "nhiều luồng nhanh hơn"; trên máy trạm, giữ lõi CPU cho camera và giao tiếp cũng quan trọng.

Phần 2: chọn ngưỡng trên val, đánh giá một lần trên test. Quy tắc được viết ra trước khi nhìn kết quả: escape trên val phải bằng 0, overkill trên val không quá ngân sách 2%, và trong các ngưỡng thỏa mãn, chọn ngưỡng thấp nhất để cách xa các ô NG nhất.

# ===== chon_nguong.py — chọn ngưỡng trên VAL theo ràng buộc escape, rồi đánh giá một lần trên TEST =====
import numpy as np
import torch

mo_hinh = MangNho()
mo_hinh.load_state_dict(torch.load("mo_hinh.pt", weights_only=True))
s_val, y_val = diem_ng(mo_hinh, X[chia["val"]]), y[chia["val"]]
OVERKILL_TOI_DA = 0.02          # ngân sách loại nhầm do QC/sản xuất chấp nhận (từ phân tích chi phí)


def escape_overkill(s, yy, t):
    ng, ok = s[yy == 1], s[yy == 0]
    return (ng < t).mean(), (ok >= t).mean()          # escape: NG có điểm dưới ngưỡng → bị cho qua


print("ngưỡng | escape val | overkill val")
ung_vien = []
for t in np.arange(0.05, 0.96, 0.05):
    e, o = escape_overkill(s_val, y_val, t)
    print(f"  {t:.2f} | {e:9.1%} | {o:9.1%}")
    if e == 0 and o <= OVERKILL_TOI_DA:
        ung_vien.append(t)
# Ưu tiên không lọt lỗi: trong các ngưỡng đạt cả hai ràng buộc, chọn ngưỡng THẤP nhất (xa các ô NG nhất)
NGUONG = round(float(min(ung_vien)), 2)
print(f"Điểm NG thấp nhất trên val: {s_val[y_val == 1].min():.3f} | điểm OK cao nhất: {s_val[y_val == 0].max():.3f}")
print(f"→ Chọn ngưỡng {NGUONG}")

# Đánh giá MỘT lần trên test (2 lô chưa từng thấy) — không chỉnh ngưỡng theo kết quả này
s_test, y_test, lo_test = diem_ng(mo_hinh, X[chia["test"]]), y[chia["test"]], lo[chia["test"]]
e, o = escape_overkill(s_test, y_test, NGUONG)
so_ng, so_ok = int((y_test == 1).sum()), int((y_test == 0).sum())
print(f"TEST: escape {int(round(e * so_ng))}/{so_ng} | overkill {int(round(o * so_ok))}/{so_ok}")
if e == 0:
    print(f"  0 escape trên {so_ng} ô NG → escape thật có lẽ dưới ~{3 / so_ng:.1%} (quy tắc 3/n)")
for k in TAP["test"]:
    m = lo_test == k
    print(f"  lô {k}: overkill {((s_test[m] >= NGUONG) & (y_test[m] == 0)).sum()}/{((y_test[m] == 0)).sum()}, "
          f"escape {((s_test[m] < NGUONG) & (y_test[m] == 1)).sum()}/{(y_test[m] == 1).sum()}")
ngưỡng | escape val | overkill val
  0.05 |      0.0% |      5.3%
  0.10 |      0.0% |      2.4%
  0.15 |      0.0% |      1.9%
  0.20 |      0.0% |      1.0%
  0.25 |      0.0% |      1.0%
  0.30 |      0.0% |      0.5%
  0.35 |      0.0% |      0.5%
  0.40 |      0.0% |      0.5%
  0.45 |      2.2% |      0.0%
  0.50 |      2.2% |      0.0%
  0.55 |      2.2% |      0.0%
  0.60 |      2.2% |      0.0%
  0.65 |      2.2% |      0.0%
  0.70 |      2.2% |      0.0%
  0.75 |      2.2% |      0.0%
  0.80 |      4.3% |      0.0%
  0.85 |      6.5% |      0.0%
  0.90 |      6.5% |      0.0%
  0.95 |      9.8% |      0.0%
Điểm NG thấp nhất trên val: 0.434 | điểm OK cao nhất: 0.415
→ Chọn ngưỡng 0.15
TEST: escape 0/89 | overkill 0/211
  0 escape trên 89 ô NG → escape thật có lẽ dưới ~3.4% (quy tắc 3/n)
  lô 9: overkill 0/104, escape 0/46
  lô 10: overkill 0/107, escape 0/43

Đọc bảng: ở ngưỡng mặc định 0,5 đã có 2,2% ô NG trên val bị lọt; với nhà máy đó là lỗi nghiêm trọng, dù "độ chính xác" vẫn trên 99%. Điểm NG thấp nhất (0,434) và điểm OK cao nhất (0,415) sát nhau: hai lớp chỉ tách được với biên rất mỏng. Ngưỡng 0,15 đổi lấy 1,9% overkill trên val để có biên an toàn lớn hơn cho escape. Trên hai lô test chưa từng thấy, không có escape và không có overkill; nhưng 0 trên 89 ô NG chỉ cho phép nói escape thật có lẽ dưới khoảng 3,4%. Muốn chứng minh tốt hơn cần nhiều mẫu NG hơn (bài tập 2).

Phần 3: xuất ONNX và kiểm tra sai số. Ở torch 2.14 (bản dùng trong bài), torch.onnx.export mặc định dùng exporter dựa trên torch.export (dynamo=True, cần gói onnxscript); exporter TorchScript cũ (dynamo=False) không còn được khuyến nghị. Chiều batch động khai báo bằng dynamic_shapes, với khóa là tên tham số của forward(). external_data=False giữ trọng số trong cùng một file; verbose=False tắt các dòng tiến trình mà exporter in ra.

# ===== xuat_onnx.py — xuất ONNX (torch 2.14: exporter dynamo mặc định), kiểm tra sai số, ghi metadata =====
import json
import os
from datetime import date

import numpy as np
import onnxruntime as ort
import torch

mo_hinh.eval()
mau_vao = torch.from_numpy(tien_xu_ly(X[:1]))
torch.onnx.export(mo_hinh, (mau_vao,), "mo_hinh.onnx", input_names=["anh"], output_names=["logit"],
                  dynamic_shapes={"x": {0: torch.export.Dim("lo_anh")}},   # "x": tên tham số của forward()
                  external_data=False, verbose=False)
METADATA = {"phien_ban": 3, "ngay": date.today().isoformat(), "kich_thuoc": [S, S],
            "tien_xu_ly": "anh_float - mean(anh); / 32.0", "nguong": NGUONG, "lo_train": TAP["train"],
            "lo_val": TAP["val"], "torch": torch.__version__.split("+")[0], "onnxruntime": ort.__version__}
with open("mo_hinh.json", "w", encoding="utf-8") as f:
    json.dump(METADATA, f, ensure_ascii=False, indent=2)

# Kiểm tra: ONNX Runtime cho cùng kết quả với PyTorch trên toàn bộ tập val + test (600 ô)
tuy_chon = ort.SessionOptions()
tuy_chon.intra_op_num_threads = 1
phien = ort.InferenceSession("mo_hinh.onnx", tuy_chon, providers=["CPUExecutionProvider"])
X_kt = X[chia["val"] | chia["test"]]
logit_onnx = phien.run(None, {"anh": tien_xu_ly(X_kt)})[0]
diem_onnx = 1 / (1 + np.exp(-logit_onnx))
diem_torch = diem_ng(mo_hinh, X_kt)
print("Vào:", [(i.name, i.shape) for i in phien.get_inputs()], "| ra:", [(o.name, o.shape) for o in phien.get_outputs()])
print(f"Kích thước file: {os.path.getsize('mo_hinh.onnx') / 1024:.0f} KB | sai lệch điểm lớn nhất: "
      f"{np.abs(diem_onnx - diem_torch).max():.1e}")
print("Quyết định OK/NG khác nhau:", int(((diem_onnx >= NGUONG) != (diem_torch >= NGUONG)).sum()), "/", len(X_kt))
print("Gần ngưỡng nhất:", f"{np.abs(diem_torch - NGUONG).min():.3f}", "(nếu rất nhỏ, sai số số học có thể lật quyết định)")
Vào: [('anh', ['lo_anh', 1, 48, 48])] | ra: [('logit', ['lo_anh'])]
Kích thước file: 40 KB | sai lệch điểm lớn nhất: 4.2e-07
Quyết định OK/NG khác nhau: 0 / 600
Gần ngưỡng nhất: 0.028 (nếu rất nhỏ, sai số số học có thể lật quyết định)

Sai lệch cỡ 1e-7 là sai số làm tròn số thực 32 bit, không lật quyết định nào. Dòng cuối là kiểm tra hay bị quên: nếu có ô nằm cách ngưỡng chỉ 1e-6, một bản ONNX Runtime khác hay một CPU khác có thể cho quyết định khác. Khi tối ưu thêm (FP16, INT8 như bài tăng tốc suy luận), sai lệch lớn hơn nhiều và kiểm tra này trở thành bắt buộc.

Phần 4: trạm chạy thật với PLC giả. BoSuyLuan chỉ cần file ONNX và JSON, chạy "mồi" một lần khi khởi động (lần suy luận đầu tiên luôn chậm). Vòng lặp giống Capstone 1, thêm đo thời gian từng bước và lưu ảnh NG với điểm trong tên file. PLC giả chạy trên cổng 5035.

# ===== tram_ai.py — trạm chạy ONNX Runtime: Modbus với PLC giả, lưu ảnh NG, đo thời gian từng bước =====
import asyncio
import json
import sqlite3
import threading
import time
from datetime import datetime
from pathlib import Path

import cv2
import numpy as np
import onnxruntime as ort
from pymodbus.client import ModbusTcpClient
from pymodbus.server import ModbusTcpServer
from pymodbus.simulator import DataType, SimData, SimDevice

TRIGGER, READY, BUSY, DONE = 0, 1, 2, 3                 # coil
HR_MA_SP, HR_CHU_KY, HR_KQ = 0, 1, 10                   # HR 10 KQ, 11 chu kỳ, 12 mã lỗi, 13 phiên bản, 14–15 điểm
KQ_OK, KQ_NG, KQ_LOI = 1, 2, 3


class PlcGia:
    """PLC giả (pymodbus 3.15, SimData/SimDevice) như bài Modbus TCP và Capstone 1."""

    def __init__(self, port):
        self.port = port

    def bat(self):
        self._da_nghe = threading.Event()
        self._thread = threading.Thread(target=lambda: asyncio.run(self._main()), daemon=True)
        self._thread.start()
        if not self._da_nghe.wait(timeout=3):
            raise RuntimeError("PLC giả không khởi động được")

    async def _main(self):
        self.loop = asyncio.get_running_loop()
        bo_nho = SimDevice(id=1, simdata=(
            [SimData(0, count=16, values=False, datatype=DataType.BITS)],
            [SimData(0, count=16, values=False, datatype=DataType.BITS)],
            [SimData(0, count=32, values=0, datatype=DataType.REGISTERS)],
            [SimData(0, count=16, values=0, datatype=DataType.REGISTERS)],
        ))
        self.server = ModbusTcpServer(bo_nho, address=("127.0.0.1", self.port))
        await self.server.serve_forever(background=True)
        self._da_nghe.set()
        await self.server.serving

    def tat(self):
        asyncio.run_coroutine_threadsafe(self.server.shutdown(), self.loop).result(timeout=3)
        self._thread.join(timeout=3)

    def _goi(self, coro):
        return asyncio.run_coroutine_threadsafe(coro, self.loop).result(timeout=2)

    def doc_coil(self, dc):
        return self._goi(self.server.async_getValues(1, 1, dc, 1))[0]

    def ghi_coil(self, dc, v):
        self._goi(self.server.async_setValues(1, 5, dc, [bool(v)]))

    def doc_hr(self, dc, n=1):
        return self._goi(self.server.async_getValues(1, 3, dc, n))

    def ghi_hr(self, dc, gia_tri):
        self._goi(self.server.async_setValues(1, 16, dc, list(gia_tri)))


def cho(dk, timeout):
    han = time.monotonic() + timeout
    while time.monotonic() < han:
        if dk():
            return True
        time.sleep(0.002)
    return False


def chuong_trinh_plc(plc, so_sp, nhan):
    for k in range(1, so_sp + 1):
        if not cho(lambda: plc.doc_coil(READY), 3.0):
            nhan.append((k, "KHONG_READY", None))
            continue
        plc.ghi_hr(HR_MA_SP, [301, k])
        plc.ghi_coil(TRIGGER, 1)
        if cho(lambda: plc.doc_coil(DONE), 1.0):
            r = plc.doc_hr(HR_KQ, 6)
            diem = ModbusTcpClient.convert_from_registers(r[4:6], ModbusTcpClient.DATATYPE.FLOAT32)
            nhan.append((k, r[0] if r[1] == k else "LECH", round(diem, 3)))
        else:
            nhan.append((k, "TIMEOUT", None))
        plc.ghi_coil(TRIGGER, 0)
        cho(lambda: not plc.doc_coil(DONE), 1.0)


class BoSuyLuan:
    """Chỉ phụ thuộc file .onnx + .json: KHÔNG import torch trên máy trạm."""

    def __init__(self, file_onnx, file_json):
        with open(file_json, encoding="utf-8") as f:
            self.meta = json.load(f)
        tc = ort.SessionOptions()
        tc.intra_op_num_threads = 1
        self.phien = ort.InferenceSession(file_onnx, tc, providers=["CPUExecutionProvider"])
        self.phien.run(None, {"anh": np.zeros((1, 1, S, S), np.float32)})   # chạy mồi: lần đầu luôn chậm

    def diem(self, cac_o):
        x = cac_o.astype(np.float32)[:, None]
        x = (x - x.mean(axis=(2, 3), keepdims=True)) / 32.0              # ĐÚNG như lúc huấn luyện
        return 1 / (1 + np.exp(-self.phien.run(None, {"anh": x})[0]))


class CameraGia:
    """Lô 9; trạm không biết kịch bản: lỗi ở chu kỳ 7, 19, 33, 48; chu kỳ 25 camera không trả ảnh."""
    KICH_BAN = {7: "xuoc", 19: "dom", 33: "xuoc", 48: "dom", 25: "camera_loi"}

    def __init__(self):
        self.rng, self.p = np.random.default_rng(77), thong_so_lo(9)

    def chup(self, k):
        loi = self.KICH_BAN.get(k)
        return None if loi == "camera_loi" else ve_o(self.rng, self.p, loi)


def chay_tram(so_sp, bo, cam, port, thu_muc="anh_ng"):
    cli = ModbusTcpClient("127.0.0.1", port=port, timeout=0.5, retries=1)
    if not cli.connect():
        raise ConnectionError("Không kết nối được PLC")
    con = sqlite3.connect("ket_qua_ai.db")
    con.execute("CREATE TABLE IF NOT EXISTS kq (thoi_gian TEXT, chu_ky INT, ket_qua INT, diem REAL, "
                "phien_ban INT, ms REAL, anh TEXT)")
    nguong, pb = bo.meta["nguong"], bo.meta["phien_ban"]
    tg = {k: [] for k in ("chup", "suy_luan", "plc", "luu")}
    cli.write_coils(READY, [True, False, False], device_id=1)
    truoc, xong, han = False, 0, time.monotonic() + 30
    try:
        while xong < so_sp and time.monotonic() < han:
            trig = cli.read_coils(TRIGGER, count=1, device_id=1).bits[0]
            if trig and not truoc:
                t0 = time.perf_counter()
                cli.write_coils(READY, [False, True, False], device_id=1)
                k = cli.read_holding_registers(HR_CHU_KY, count=1, device_id=1).registers[0]
                anh = cam.chup(k)
                t1 = time.perf_counter()
                if anh is None:
                    kq, diem = KQ_LOI, -1.0                     # không có ảnh: LỖI, PLC xử lý như NG
                else:
                    diem = float(bo.diem(anh[None])[0])
                    kq = KQ_NG if diem >= nguong else KQ_OK
                t2 = time.perf_counter()
                regs = [kq, k, 256 if kq == KQ_LOI else int(kq == KQ_NG), pb]
                regs += cli.convert_to_registers(diem, cli.DATATYPE.FLOAT32)
                cli.write_registers(HR_KQ, regs, device_id=1)            # kết quả trước, một lệnh
                cli.write_coils(READY, [False, False, True], device_id=1)  # rồi DONE
                t3 = time.perf_counter()
                duong_dan = None
                if kq != KQ_OK:
                    ts = datetime.now()
                    duong_dan = f"{ts:%Y-%m-%d}/C{k:05d}_d{max(diem, 0):.2f}.png"
                    (Path(thu_muc) / duong_dan).parent.mkdir(parents=True, exist_ok=True)
                    if anh is not None:
                        cv2.imwrite(str(Path(thu_muc) / duong_dan), anh)
                    else:
                        duong_dan = None
                con.execute("INSERT INTO kq VALUES (?,?,?,?,?,?,?)", (datetime.now().isoformat(timespec="milliseconds"),
                            k, kq, round(diem, 4), pb, round((t3 - t0) * 1000, 2), duong_dan))
                con.commit()
                t4 = time.perf_counter()
                for ten, a, b in (("chup", t0, t1), ("suy_luan", t1, t2), ("plc", t2, t3), ("luu", t3, t4)):
                    tg[ten].append((b - a) * 1000)
                cho(lambda: not cli.read_coils(TRIGGER, count=1, device_id=1).bits[0], 1.0)
                cli.write_coils(READY, [True, False, False], device_id=1)
                trig, xong = False, xong + 1
            truoc = trig
            time.sleep(0.002)
    finally:
        cli.write_coils(READY, [False, False, False], device_id=1)
        cli.close()
        con.close()
    return tg


Path("ket_qua_ai.db").unlink(missing_ok=True)
bo = BoSuyLuan("mo_hinh.onnx", "mo_hinh.json")
plc = PlcGia(5035)
plc.bat()
nhan = []
t_plc = threading.Thread(target=chuong_trinh_plc, args=(plc, 60, nhan), daemon=True)
t_plc.start()
try:
    tg = chay_tram(60, bo, CameraGia(), 5035)
finally:
    t_plc.join(timeout=5)
    plc.tat()

ten_kq = {KQ_OK: "OK", KQ_NG: "NG", KQ_LOI: "LỖI"}
print("PLC nhận:", len(nhan), "kết quả |", {v: sum(1 for _, kq, _ in nhan if kq == k) for k, v in ten_kq.items()})
print("Chu kỳ không OK:", [(k, ten_kq.get(kq, kq), d) for k, kq, d in nhan if kq != KQ_OK])
print("Ảnh NG đã lưu:", sorted(p.name for p in Path("anh_ng").rglob("*.png")))
print("Thời gian từng bước, ms (tùy máy):")
for ten, v in tg.items():
    print(f"  {ten:8s} p50 {np.percentile(v, 50):6.2f} | p99 {np.percentile(v, 99):6.2f}")
# Một sản phẩm thật được chia thành nhiều ô: đo suy luận theo LÔ 64 ô (batch)
lo64 = np.stack([ve_o(np.random.default_rng(i), thong_so_lo(9)) for i in range(64)])
t_lo = []
for _ in range(30):
    a = time.perf_counter()
    bo.diem(lo64)
    t_lo.append((time.perf_counter() - a) * 1000)
print(f"Suy luận lô 64 ô: p50 {np.percentile(t_lo, 50):.1f} ms, p99 {np.percentile(t_lo, 99):.1f} ms (tùy máy)")
PLC nhận: 60 kết quả | {'OK': 55, 'NG': 4, 'LỖI': 1}
Chu kỳ không OK: [(7, 'NG', 0.998), (19, 'NG', 1.0), (25, 'LỖI', -1.0), (33, 'NG', 0.999), (48, 'NG', 0.934)]
Ảnh NG đã lưu: ['C00007_d1.00.png', 'C00019_d1.00.png', 'C00033_d1.00.png', 'C00048_d0.93.png']
Thời gian từng bước, ms (tùy máy):
  chup     p50   1.41 | p99   2.64
  suy_luan p50   0.42 | p99   0.85
  plc      p50   0.76 | p99   2.98
  luu      p50   6.05 | p99  84.14
Suy luận lô 64 ô: p50 7.3 ms, p99 8.5 ms (tùy máy)

PLC nhận đủ 60 kết quả: 4 ô lỗi đều NG với điểm từ 0,93 tới 1,0; chu kỳ 25 (camera không trả ảnh) là LỖI với điểm −1 và không có ảnh để lưu. Các con số thời gian tùy máy; bước suy luận một ô chưa tới 1 ms (p50) ở máy viết bài, còn một nắp thật 64 ô suy luận theo lô cũng chỉ mất cỡ 10 ms, rất xa ngân sách 500 ms. Ở trạm thật, phần tốn thời gian thường là chụp, truyền ảnh và ghi đĩa, không phải mô hình nhỏ.

7. Kiểm thử và nghiệm thu

7.1. Giám sát sau triển khai: kịch bản camera mất nét

Mô hình đạt mọi tiêu chí hôm nghiệm thu chưa chắc đạt sau ba tháng. Kịch bản thử: vòng chỉnh nét của ống kính bị lỏng do rung, ảnh mờ dần từ sản phẩm 1001 tới 2000. Trạm không có lỗi phần mềm nào, PLC vẫn nhận kết quả đều đặn. Ba bộ giám sát chạy song song:

  • Độ sáng trung bình ảnh (EWMA, giới hạn 4σ học từ 200 sản phẩm đầu).
  • Độ nét: log của phương sai Laplacian, cũng theo dõi bằng EWMA. Chỉ tính trên ảnh được chấm OK, vì vết lỗi làm tăng độ "nét" giả (thử nghiệm ban đầu tính trên mọi ảnh đã gây một cảnh báo nhầm ở giai đoạn bình thường).
  • Tỉ lệ NG trong cửa sổ 300 sản phẩm, so với tỉ lệ bình thường 3%, kiểm cả hai phía: NG giảm bất thường cũng đáng ngờ như NG tăng.
# ===== giam_sat.py — giám sát trôi dữ liệu đầu vào (drift) và tỉ lệ NG trong sản xuất =====
from collections import deque

import cv2
import numpy as np


class EWMA:
    """Biểu đồ EWMA: học trung bình và độ lệch chuẩn ở giai đoạn tham chiếu, rồi báo khi trôi quá k·σ_EWMA."""

    def __init__(self, so_tham_chieu=200, lam=0.1, k=4.0):
        self.ref, self.n, self.lam, self.k, self.z = [], so_tham_chieu, lam, k, None

    def cap_nhat(self, x):
        if len(self.ref) < self.n:
            self.ref.append(x)
            if len(self.ref) == self.n:
                self.mu, sd = np.mean(self.ref), np.std(self.ref, ddof=1)
                self.gioi_han = self.k * sd * np.sqrt(self.lam / (2 - self.lam))
                self.z = self.mu
            return None
        self.z = self.lam * x + (1 - self.lam) * self.z
        return abs(self.z - self.mu) > self.gioi_han


class GiamSat:
    def __init__(self, p0=0.03, cua_so=300):
        self.sang, self.net = EWMA(), EWMA()
        self.kq, self.p0 = deque(maxlen=cua_so), p0      # p0: tỉ lệ NG bình thường của dây chuyền
        self.lan_dau = {}

    def cap_nhat(self, i, anh, ng):
        self.kq.append(ng)
        if ng:
            return                         # thống kê đầu vào chỉ lấy trên ảnh OK: vết lỗi làm tăng độ "nét" giả
        if self.sang.cap_nhat(float(anh.mean())):
            self.lan_dau.setdefault("độ sáng", i)
        do_net = float(cv2.Laplacian(anh, cv2.CV_32F).var())   # phương sai Laplacian: thước đo độ nét
        if self.net.cap_nhat(np.log(do_net)):
            self.lan_dau.setdefault("độ nét", i)
        if len(self.kq) == self.kq.maxlen:                     # tỉ lệ NG, kiểm HAI phía
            p, sai = np.mean(self.kq), 3 * np.sqrt(self.p0 * (1 - self.p0) / len(self.kq))
            if p > self.p0 + sai or p < self.p0 - sai:
                self.lan_dau.setdefault("tỉ lệ NG", i)


# 2000 sản phẩm lô 10, tỉ lệ lỗi 3%. Từ sản phẩm 1001, ống kính mất nét dần (vòng chỉnh nét bị lỏng do rung):
# độ mờ tăng tuyến tính tới σ = 1,2 px ở sản phẩm 2000. Trạm vẫn chạy, không báo lỗi gì.
rng = np.random.default_rng(5)
p10 = thong_so_lo(10)
gs = GiamSat()
that, du_doan, mo = [], [], []
for i in range(1, 2001):
    sigma = 0.0 if i <= 1000 else 1.2 * (i - 1000) / 1000
    loi = ("xuoc" if rng.random() < 0.6 else "dom") if rng.random() < 0.03 else None
    anh = ve_o(rng, p10, loi)
    if sigma > 0.05:
        anh = cv2.GaussianBlur(anh, (0, 0), sigma)
    ng = bool(bo.diem(anh[None])[0] >= bo.meta["nguong"])
    gs.cap_nhat(i, anh, ng)
    that.append(loi is not None), du_doan.append(ng), mo.append(sigma)
that, du_doan, mo = map(np.array, (that, du_doan, mo))

for ten in ("độ sáng", "độ nét", "tỉ lệ NG"):
    i = gs.lan_dau.get(ten)
    print(f"Giám sát {ten:8s}: " + (f"CẢNH BÁO tại sản phẩm {i} (độ mờ σ = {mo[i - 1]:.2f} px)" if i
                                    else "không cảnh báo"))
print("sản phẩm   | độ mờ σ (px) | lỗi thật | escape | overkill")
for a, b in ((1, 1000), (1001, 1400), (1401, 1700), (1701, 2000)):
    m = slice(a - 1, b)
    print(f"{a:4d}–{b:<4d} | {mo[m].min():.2f}–{mo[m].max():.2f}    | {that[m].sum():8d} | "
          f"{(that[m] & ~du_doan[m]).sum():6d} | {(~that[m] & du_doan[m]).sum():8d}")
Giám sát độ sáng : không cảnh báo
Giám sát độ nét  : CẢNH BÁO tại sản phẩm 1298 (độ mờ σ = 0.36 px)
Giám sát tỉ lệ NG: không cảnh báo
sản phẩm   | độ mờ σ (px) | lỗi thật | escape | overkill
   1–1000 | 0.00–0.00    |       30 |      0 |        1
1001–1400 | 0.00–0.48    |       13 |      0 |        0
1401–1700 | 0.48–0.84    |       10 |      3 |        0
1701–2000 | 0.84–1.20    |       11 |      6 |        0

Ba bài học từ bảng này:

  1. Mô hình hỏng một cách im lặng. Khi ảnh mờ, vết xước mảnh nhòe vào vân, điểm NG giảm, và trạm bắt đầu cho qua lỗi (escape tăng) chứ không loại nhầm. Không có exception, không có overkill để công nhân phàn nàn.
  2. Giám sát đầu ra đến quá muộn. Escape làm tỉ lệ NG giảm, nhưng với 3% lỗi và cửa sổ 300 sản phẩm, giới hạn dưới gần bằng 0: phải gần như không còn NG nào thì mới báo. Bộ giám sát tỉ lệ NG không báo gì trong cả 2000 sản phẩm.
  3. Giám sát đầu vào đến sớm. Độ nét báo ở độ mờ khoảng 0,36 px, trước escape đầu tiên. Độ sáng không đổi khi mất nét nên bộ giám sát độ sáng im lặng: mỗi loại trôi cần một thước đo riêng (độ sáng cho đèn yếu, độ nét cho lấy nét và kính bẩn, phân bố điểm cho thay đổi vật liệu).
Mẫu kiểm tra định kỳ (golden sample)Bổ sung cho giám sát thống kê: đầu mỗi ca, cho 3–5 mẫu NG chuẩn và vài mẫu OK chuẩn chạy qua trạm; PLC chỉ cho chạy sản xuất khi trạm phân loại đúng hết. Đây là thực hành phổ biến trong nhà máy, rẻ và rất hiệu quả.

7.2. Biên bản nghiệm thu

# ===== bao_cao.py — bảng nghiệm thu trạm AI =====
from pathlib import Path

import numpy as np

e_test, o_test = escape_overkill(s_test, y_test, NGUONG)
lech_onnx = float(np.abs(diem_onnx - diem_torch).max())
lat_onnx = int(((diem_onnx >= NGUONG) != (diem_torch >= NGUONG)).sum())
p99_lo = float(np.percentile(t_lo, 99))
dem_plc = {v: sum(1 for _, kq, _ in nhan if kq == k) for k, v in ((KQ_OK, "OK"), (KQ_NG, "NG"), (KQ_LOI, "LOI"))}
escape_dau = int(np.argmax(that & ~du_doan)) + 1 if (that & ~du_doan).any() else None
canh_bao_net = gs.lan_dau.get("độ nét")
roi_nhau = len(set(TAP["train"]) | set(TAP["val"]) | set(TAP["test"])) == sum(len(v) for v in TAP.values())
hang = [
    ("R1", "Train/val/test không chung lô", "rời nhau", "rời nhau" if roi_nhau else "TRÙNG", roi_nhau),
    ("R2", "Escape trên test (lô 9, 10)", "0", f"{int(round(e_test * so_ng))}/{so_ng}", e_test == 0),
    ("R3", "Overkill trên test", "≤ 2 %", f"{o_test:.1%}", o_test <= OVERKILL_TOI_DA),
    ("R4", "ONNX = PyTorch (điểm / quyết định)", "≤ 1e-4 / 0", f"{lech_onnx:.0e} / {lat_onnx}",
     lech_onnx <= 1e-4 and lat_onnx == 0),
    ("R5", "Suy luận lô 64 ô, p99 (*)", "≤ 50 ms", f"{p99_lo:.1f} ms", p99_lo <= 50),
    ("R6", "PLC nhận đúng kịch bản", "55/4/1", f"{dem_plc['OK']}/{dem_plc['NG']}/{dem_plc['LOI']}",
     (dem_plc["OK"], dem_plc["NG"], dem_plc["LOI"]) == (55, 4, 1)),
    ("R7", "Ảnh NG được lưu", "4", str(len(list(Path("anh_ng").rglob("*.png")))),
     len(list(Path("anh_ng").rglob("*.png"))) == 4),
    ("R8", "Cảnh báo drift trước escape đầu tiên", "có", f"SP {canh_bao_net} < SP {escape_dau}",
     canh_bao_net is not None and (escape_dau is None or canh_bao_net < escape_dau)),
    ("R9", "Metadata mô hình (phiên bản, ngưỡng)", "có", f"v{bo.meta['phien_ban']}, {bo.meta['nguong']}",
     bo.meta["nguong"] == NGUONG),
]
for ma, muc, yc, kq, dat in hang:
    print(f"{ma} {muc:37s} {yc:12s} {kq:18s} {'ĐẠT' if dat else 'KHÔNG ĐẠT'}")
print(f"Kết luận: {'ĐẠT' if all(h[-1] for h in hang) else 'KHÔNG ĐẠT'} ({sum(h[-1] for h in hang)}/{len(hang)}),"
      " (*) tùy máy")
R1 Train/val/test không chung lô         rời nhau     rời nhau           ĐẠT
R2 Escape trên test (lô 9, 10)           0            0/89               ĐẠT
R3 Overkill trên test                    ≤ 2 %        0.0%               ĐẠT
R4 ONNX = PyTorch (điểm / quyết định)    ≤ 1e-4 / 0   4e-07 / 0          ĐẠT
R5 Suy luận lô 64 ô, p99 (*)             ≤ 50 ms      8.5 ms             ĐẠT
R6 PLC nhận đúng kịch bản                55/4/1       55/4/1             ĐẠT
R7 Ảnh NG được lưu                       4            4                  ĐẠT
R8 Cảnh báo drift trước escape đầu tiên  có           SP 1298 < SP 1481  ĐẠT
R9 Metadata mô hình (phiên bản, ngưỡng)  có           v3, 0.15           ĐẠT
Kết luận: ĐẠT (9/9), (*) tùy máy

R5 tùy máy (đo trên CPU, 1 luồng). Như hai dự án trước, biên bản được sinh lại mỗi lần huấn luyện mô hình mới; một mô hình chỉ được đưa xuống trạm khi biên bản của nó đạt hết.

7.3. Lỗi thường gặp

Triệu chứngNguyên nhânCách sửa
Đánh giá rất đẹp, xuống dây chuyền overkill caoChia ngẫu nhiên, cùng lô ở cả train và test; hoặc chỉnh ngưỡng theo testChia theo lô/ngày; chọn ngưỡng trên val; test chỉ dùng một lần
ONNX cho kết quả khác PyTorchTiền xử lý ở trạm khác lúc huấn luyện (chia 255, đổi kênh, resize khác), quên eval()Ghi tiền xử lý vào metadata, một hàm dùng chung, kiểm tra sai số trên dữ liệu thật
Chu kỳ đầu tiên sau khởi động bị timeoutLần suy luận đầu khởi tạo bộ nhớ, tối ưu đồ thịChạy mồi khi khởi động, trước khi bật READY
Camera lỗi mà trạm báo OKẢnh rỗng hoặc ảnh cũ vẫn được đưa vào mô hìnhKhông có ảnh mới → LỖI, không bao giờ OK
Escape tăng dần mà không ai biếtChỉ giám sát tỉ lệ NG một phía hoặc không giám sátGiám sát đầu vào (sáng, nét), tỉ lệ NG hai phía, mẫu chuẩn mỗi ca
Không biết sản phẩm cũ được kiểm bằng mô hình nàoKhông quản lý phiên bảnPhiên bản trong metadata, gửi PLC, lưu trong mỗi bản ghi

8. Sản phẩm bàn giao cho portfolio

  • README: bài toán ("Kiểm tra xước/đốm trên nắp nhôm xước sợi, 30 nắp/phút, escape 0 trên lô mới"), sơ đồ hai thế giới huấn luyện/trạm, bảng nghiệm thu R1–R9, cách chạy từ đầu.
  • Model card (MODEL_CARD.md): dữ liệu từ lô nào, cách chia, kiến trúc, tiền xử lý, ngưỡng và quy tắc chọn ngưỡng, kết quả val/test, giới hạn đã biết (vết xước song song vân khó thấy, chưa thử trên lô có màu khác), điều kiện phải huấn luyện lại.
  • Hình: lưới ảnh các ô NG có điểm cao nhất và các ô OK có điểm cao nhất (gần ngưỡng); biểu đồ phân bố điểm val với vạch ngưỡng; biểu đồ EWMA độ nét có đường giới hạn và điểm báo động.
  • Video 3 phút: chạy trạm với PLC giả, mở thư mục ảnh NG; chạy kịch bản mất nét, chỉ vào thời điểm cảnh báo và bảng escape.
  • Phỏng vấn: nhà tuyển dụng nghe "mô hình đạt 99%" rất nhiều lần. Hãy nói "ngưỡng mặc định 0,5 lọt 2,2% lỗi nên tôi chọn lại ngưỡng theo escape, chấp nhận 1,9% overkill, và tôi phát hiện được camera mất nét trước khi trạm lọt lỗi". Đó là ngôn ngữ của người đã làm trạm thật.

9. Tự đánh giá

Tiêu chíChưa đạtĐạtTốt
Dữ liệuChia ngẫu nhiênChia theo lô, test là lô mới nhấtThống kê theo lô, loại lỗi; có kế hoạch thu thập thêm mẫu NG
NgưỡngDùng 0,5Chọn trên val theo escape và ngân sách overkillQuy tắc viết trước, có biên an toàn, giải thích bằng giới hạn thống kê
Triển khaiTrạm chạy PyTorch từ notebookONNX + metadata, kiểm tra sai số, chạy mồiĐo p50/p99 từng bước, quản lý phiên bản, chạy như dịch vụ 24/7
Tích hợpIn kết quả ra màn hìnhPLC đúng bắt tay, LỖI khi không có ảnh, ảnh NGTự kết nối lại, ghi nền, log xoay vòng (như dự án trạm vision mô phỏng)
Giám sátKhông cóTỉ lệ NG theo cửa sổGiám sát đầu vào + tỉ lệ NG hai phía + mẫu chuẩn mỗi ca, đã thử bằng kịch bản drift
Tài liệuKhông READMEREADME + biên bảnModel card, video, nêu rõ giới hạn

10. Mở rộng

  • Anomaly detection: khi gần như không có ảnh lỗi, thay bộ phân loại bằng mô hình học từ ảnh OK (PatchCore, như dự án anomaly detection của chặng 12); giữ nguyên quy trình ngưỡng trên val, ONNX, PLC, giám sát.
  • Backbone đã huấn luyện sẵn: trên máy bạn (có mạng), thay MangNho bằng resnet18(weights=ResNet18_Weights.DEFAULT) với ảnh 3 kênh như bài transfer learning, so sánh escape/overkill và thời gian.
  • Bản đồ vị trí lỗi: trả về ô có điểm cao nhất và vẽ khung lên ảnh nắp cho HMI; hoặc chuyển sang segmentation để đo kích thước vết lỗi.
  • Tăng tốc: thử OpenVINO hoặc lượng tử hóa INT8, kiểm tra lại sai số và R2–R4 trước khi dùng.
  • Vòng đời mô hình: thư mục ảnh NG và ảnh gần ngưỡng là dữ liệu cho lần huấn luyện sau; viết quy trình duyệt nhãn, huấn luyện lại, sinh biên bản, đổi phiên bản.

Bài 1. Mỗi nắp có 64 ô. Overkill theo ô là 0,1%, các ô độc lập. Overkill theo nắp là bao nhiêu? Muốn overkill theo nắp ≤ 1% thì overkill theo ô phải nhỏ hơn bao nhiêu?

Xem đáp án

P(nắp OK bị loại) = 1 − (1 − 0,001)64 ≈ 6,2%. Muốn ≤ 1%: 1 − (1 − p)64 ≤ 0,01 → p ≤ 1 − 0,991/64 ≈ 0,016% theo ô. Đây là lý do tiêu chí phải viết theo sản phẩm, và đánh giá phải tính lại từ ô sang sản phẩm.

Bài 2. Khách hàng yêu cầu chứng minh escape dưới 1% với độ tin cậy khoảng 95%. Theo quy tắc 3/n, cần bao nhiêu mẫu NG (độc lập) chạy qua trạm mà không lọt cái nào?

Xem đáp án

3/n ≤ 0,01 → n ≥ 300 mẫu NG không lọt mẫu nào. Tập test của bài chỉ có 89 ô NG, nên chỉ chứng minh được dưới khoảng 3,4%. Cần thu thập thêm mẫu NG thật (hoặc nhiều lần chạy của các mẫu khác nhau, cẩn thận với giả định độc lập).

Bài 3. Sau khi nhà cung cấp nhôm đổi, ảnh lô mới sáng hơn 15% và vân thô hơn. Bộ giám sát độ sáng báo ngay. Bạn làm gì, theo thứ tự?

Xem đáp án

(1) Không tắt cảnh báo; chạy mẫu chuẩn OK/NG của ca để biết trạm còn phân loại đúng không. (2) Lấy mẫu vài trăm sản phẩm lô mới, QC gán nhãn, đánh giá escape/overkill của mô hình hiện tại trên đó. (3) Nếu không đạt: thêm dữ liệu lô mới vào train (giữ một phần làm test), huấn luyện lại, chọn ngưỡng, sinh biên bản, tăng phiên bản. (4) Cập nhật giai đoạn tham chiếu của bộ giám sát cho vật liệu mới. Trong lúc chờ, tăng tần suất kiểm tra thủ công lô đó.

Tóm tắt

  • Chia dữ liệu theo lô; chọn ngưỡng trên val bằng quy tắc viết trước (escape trước, overkill trong ngân sách); test chỉ dùng một lần.
  • Ngưỡng 0,5 và "độ chính xác" không phải tiêu chí nhà máy: hãy báo escape, overkill và giới hạn thống kê của chúng.
  • Triển khai bằng ONNX + metadata (phiên bản, ngưỡng, tiền xử lý), kiểm tra sai số và khoảng cách tới ngưỡng, chạy mồi, đo p50/p99 từng bước.
  • Tích hợp như mọi trạm: ghi kết quả trước DONE, không có ảnh thì LỖI, lưu ảnh NG có điểm và phiên bản.
  • Mô hình hỏng im lặng: giám sát đầu vào (độ sáng, độ nét) phát hiện drift sớm hơn tỉ lệ NG; kiểm hai phía và dùng mẫu chuẩn mỗi ca.
  • Ba capstone cùng một khung: đặc tả bằng con số, lõi kỹ thuật có đáp án, tích hợp PLC, biên bản tự sinh.

Bài tiếp theo: Viết tài liệu cho dự án Vision: README, đặc tả, hướng dẫn vận hành, nơi bạn biến ba dự án này thành tài liệu mà khách hàng và đồng nghiệp đọc được.

Bài tiếp theo
Ban biên tập HọcFree

Chúng tôi viết những bài hướng dẫn dễ hiểu, thực tế để ai cũng có thể tự học lập trình và ngoại ngữ, hoàn toàn miễn phí.