Lưu kết quả kiểm tra vào SQLite, ảnh NG theo ngày và dọn ổ đĩa

Ba tháng sau khi bàn giao trạm, khách hàng gọi: "Lô hàng tuần trước bị khiếu nại, cho xem kết quả kiểm tra và ảnh của các sản phẩm đó". Hoặc tệ hơn: "Máy vision tự dừng lúc 3 giờ sáng", và nguyên nhân là ổ đĩa đầy ảnh. Bài "MES và truy xuất nguồn gốc" đã nói trạm vision nên lưu gì; bài này là phần làm: lưu kết quả vào SQLite sao cho truy vấn nhanh, lưu ảnh NG theo thư mục ngày, tự dọn ảnh cũ để ổ đĩa không bao giờ đầy, và ghi tất cả những thứ đó trong một thread nền để không làm chậm cycle time. Đây là mảnh ghép cuối trước dự án chặng 9.

1. Vì sao SQLite, không phải CSV?

SQLite hợp với dữ liệu cục bộ của một trạm: hàng chục nghìn đến vài triệu bản ghi, một chương trình ghi. Khi nhiều trạm cần ghi chung một CSDL qua mạng, đó là việc của máy chủ CSDL hoặc MES (đừng đặt file SQLite trên ổ mạng chia sẻ). Ngay cả khi có MES, CSDL cục bộ vẫn là bộ đệm khi mạng gián đoạn.

2. Thiết kế bảng và mở CSDL đúng cách

import sqlite3
from pathlib import Path

GOC = Path("du_lieu")
GOC.mkdir(exist_ok=True)

SCHEMA = """
CREATE TABLE IF NOT EXISTS kiem_tra (
    id            INTEGER PRIMARY KEY,
    thoi_gian     TEXT    NOT NULL,          -- ISO 8601: '2026-10-09T08:15:02.123'
    tram          TEXT    NOT NULL,
    ma_sp         TEXT,                      -- số seri / mã đọc được, nếu có
    ket_qua       TEXT    NOT NULL CHECK (ket_qua IN ('OK', 'NG')),
    ma_loi        TEXT,                      -- NULL khi OK
    gia_tri       REAL,                      -- giá trị đo chính (mm)
    tg_xu_ly_ms   REAL,
    recipe        TEXT,
    duong_dan_anh TEXT                       -- tương đối so với thư mục ảnh; NULL nếu không lưu/đã xóa
);
CREATE INDEX IF NOT EXISTS ix_kt_thoi_gian ON kiem_tra(thoi_gian);
CREATE INDEX IF NOT EXISTS ix_kt_kq_thoi_gian ON kiem_tra(ket_qua, thoi_gian);
"""


def mo_csdl(duong_dan):
    con = sqlite3.connect(duong_dan, timeout=5.0)          # chờ tối đa 5 s nếu file đang bị khóa
    con.execute("PRAGMA journal_mode=WAL")                  # đọc không chặn ghi, ghi không chặn đọc
    con.execute("PRAGMA synchronous=NORMAL")                # an toàn với WAL, nhanh hơn FULL nhiều
    con.executescript(SCHEMA)
    return con


con = mo_csdl(GOC / "ket_qua.db")
print("journal_mode:", con.execute("PRAGMA journal_mode").fetchone()[0])
print("Bảng:", [r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")])
print("Chỉ mục:", [r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='index'")])

journal_mode: wal
Bảng: ['kiem_tra']
Chỉ mục: ['ix_kt_thoi_gian', 'ix_kt_kq_thoi_gian']

• Thời gian dạng chuỗi ISO 8601 (2026-10-09T08:15:02.123): sắp xếp theo chữ cũng là sắp xếp theo thời gian, đọc được bằng mắt, và các hàm date(), strftime() của SQLite hiểu được. Nhớ ghi rõ là giờ địa phương hay UTC, và đồng bộ đồng hồ máy (NTP).
• Chỉ mục trên thoi_gian và (ket_qua, thoi_gian): gần như mọi câu hỏi đều theo khoảng thời gian. Đừng tạo chỉ mục cho mọi cột: mỗi chỉ mục làm chậm việc ghi một chút.
• WAL (write-ahead log): người khác mở CSDL để xem (ví dụ bằng DB Browser for SQLite) không chặn chương trình đang ghi. Với WAL, synchronous=NORMAL vẫn giữ CSDL nguyên vẹn khi mất điện, chỉ có thể mất các giao dịch cuối cùng chưa kịp đồng bộ xuống đĩa.
• Đường dẫn ảnh tương đối so với thư mục ảnh: chuyển cả thư mục sang ổ khác hay máy khác mà CSDL vẫn đúng.

3. Ghi dữ liệu: tham số và gom giao dịch

import time
from datetime import datetime, timedelta

import numpy as np

SQL_THEM = """INSERT INTO kiem_tra (thoi_gian, tram, ma_sp, ket_qua, ma_loi, gia_tri, tg_xu_ly_ms, recipe, duong_dan_anh)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"""

# Một bản ghi: luôn dùng tham số ?, không bao giờ ghép chuỗi SQL bằng f-string
con.execute(SQL_THEM, ("2026-10-09T08:15:02.123", "VIS-01", "SN0001", "OK", None, 10.004, 38.2, "VONG_DEM_10", None))
con.commit()

# Dữ liệu giả 3 ngày, 2 ca/ngày, mỗi 20 s một sản phẩm
rng = np.random.default_rng(1)
LOI = ["thieu_vat", "xuoc_be_mat", "sai_kich_thuoc"]
ban_ghi = []
for ngay in range(3):
    t = datetime(2026, 10, 6 + ngay, 6, 0, 0)
    ty_le_ng = [0.02, 0.05, 0.03][ngay]
    for i in range(1440):                                   # 8 giờ x 180 sản phẩm/giờ
        ng = rng.random() < ty_le_ng
        ma_loi = str(rng.choice(LOI, p=[0.5, 0.3, 0.2])) if ng else None
        d = round(float(rng.normal(10.0, 0.02)), 3)
        ban_ghi.append((t.isoformat(timespec="milliseconds"), "VIS-01", f"SN{ngay}{i:05d}",
                        "NG" if ng else "OK", ma_loi, d, round(float(rng.normal(40, 3)), 1),
                        "VONG_DEM_10", None))
        t += timedelta(seconds=20)

# So sánh: commit từng dòng và gom một giao dịch
t0 = time.perf_counter()
for r in ban_ghi[:500]:
    con.execute(SQL_THEM, r)
    con.commit()
t_tung_dong = time.perf_counter() - t0

t0 = time.perf_counter()
with con:                                                   # một giao dịch: tự commit khi ra khỏi khối
    con.executemany(SQL_THEM, ban_ghi[500:])
t_gom = time.perf_counter() - t0

print(f"Commit từng dòng : {t_tung_dong / 500 * 1e6:7.1f} µs/dòng")
print(f"Một giao dịch    : {t_gom / (len(ban_ghi) - 500) * 1e6:7.1f} µs/dòng")
print("Tổng số bản ghi:", con.execute("SELECT COUNT(*) FROM kiem_tra").fetchone()[0])

Commit từng dòng :    55.6 µs/dòng
Một giao dịch    :     3.2 µs/dòng
Tổng số bản ghi: 4321

Thời gian tùy máy và ổ đĩa, nhưng gom giao dịch luôn nhanh hơn nhiều lần tính trên mỗi dòng. Mỗi commit có một chi phí cố định: khóa file, ghi nhật ký, và ở chế độ mặc định synchronous=FULL còn phải chờ hệ điều hành xác nhận dữ liệu đã xuống đĩa (fsync). Trên HDD hoặc thẻ nhớ công nghiệp, một lần fsync có thể tốn vài ms đến hàng chục ms, đủ ăn mất cycle time. WAL + NORMAL đã bỏ fsync ở mỗi commit (chỉ đồng bộ khi checkpoint) nên demo trên khá nhanh, nhưng vẫn nên gom: ở mục 7 ta commit sau mỗi 50 dòng hoặc mỗi 0,5 s, cái nào đến trước. Khi mất điện, CSDL vẫn nguyên vẹn; cái mất là vài giây dữ liệu cuối (bản ghi còn trong hàng đợi và các giao dịch chưa kịp đồng bộ xuống đĩa). Với hầu hết trạm điều này chấp nhận được, vì PLC đã nhận kết quả của từng sản phẩm.

4. Truy vấn: yield theo ngày, lý do NG, NG theo giờ

print("Ngày        Tổng    OK   NG  Yield")
for ngay, tong, ok, ng, y in con.execute("""
        SELECT date(thoi_gian) AS ngay, COUNT(*), SUM(ket_qua = 'OK'), SUM(ket_qua = 'NG'),
               ROUND(100.0 * SUM(ket_qua = 'OK') / COUNT(*), 2)
        FROM kiem_tra GROUP BY ngay ORDER BY ngay"""):
    print(f"{ngay}  {tong:5d} {ok:5d} {ng:4d}  {y:.2f}%")

print("\nLý do NG ngày 2026-10-07:")
for ma_loi, n in con.execute("""
        SELECT ma_loi, COUNT(*) AS n FROM kiem_tra
        WHERE ket_qua = 'NG' AND thoi_gian >= ? AND thoi_gian < ?
        GROUP BY ma_loi ORDER BY n DESC""", ("2026-10-07", "2026-10-08")):
    print(f"  {ma_loi:<15} {n}")

print("\nNG theo giờ ngày 2026-10-07:")
hang = con.execute("""
        SELECT strftime('%H', thoi_gian) AS gio, SUM(ket_qua = 'NG') FROM kiem_tra
        WHERE thoi_gian >= ? AND thoi_gian < ? GROUP BY gio""", ("2026-10-07", "2026-10-08")).fetchall()
print("  " + "  ".join(f"{g}h:{n}" for g, n in hang))

ke_hoach = con.execute("""EXPLAIN QUERY PLAN SELECT COUNT(*) FROM kiem_tra
                         WHERE thoi_gian >= ? AND thoi_gian < ?""", ("2026-10-07", "2026-10-08")).fetchall()
print("\nKế hoạch truy vấn:", ke_hoach[0][-1])

Ngày        Tổng    OK   NG  Yield
2026-10-06   1440  1410   30  97.92%
2026-10-07   1440  1373   67  95.35%
2026-10-08   1440  1393   47  96.74%
2026-10-09      1     1    0  100.00%

Lý do NG ngày 2026-10-07:
  thieu_vat       34
  xuoc_be_mat     26
  sai_kich_thuoc  7

NG theo giờ ngày 2026-10-07:
  06h:7  07h:8  08h:8  09h:7  10h:7  11h:10  12h:10  13h:10

Kế hoạch truy vấn: SEARCH kiem_tra USING COVERING INDEX ix_kt_thoi_gian (thoi_gian>? AND thoi_gian<?)

Ngày 07 yield giảm còn 95,35%, chủ yếu do thieu_vat: đây là loại báo cáo mà quản lý sản xuất muốn xem mỗi sáng. Dòng cuối là kế hoạch truy vấn: SEARCH ... USING COVERING INDEX ix_kt_thoi_gian nghĩa là SQLite dùng chỉ mục thay vì đọc cả bảng. Lưu ý mẹo viết điều kiện: thoi_gian >= '2026-10-07' AND thoi_gian < '2026-10-08' dùng được chỉ mục, còn date(thoi_gian) = '2026-10-07' thì không (phải tính hàm trên từng dòng). Với vài triệu dòng, khác biệt là vài ms so với vài giây.

5. Lưu ảnh NG theo thư mục ngày

Cấu trúc gợi ý:

du_lieu/
├── ket_qua.db
└── anh/
    ├── 2026-10-08/
    │   ├── 061512_004_SN00421_thieu_vat.jpg
    │   └── ...
    └── 2026-10-09/
        └── 081502_123_SN0042_xuoc_be_mat.jpg   ← giờ phút giây_ms_mã SP_lỗi

• Một thư mục mỗi ngày: dọn dẹp chỉ cần xóa cả thư mục; mỗi thư mục không quá nhiều file (trình quản lý file và lệnh liệt kê chậm hẳn khi một thư mục có hàng trăm nghìn file).
• Tên file tự mô tả và sắp xếp được: nhìn tên biết thời điểm, sản phẩm, lỗi mà không cần mở CSDL. Chỉ dùng chữ không dấu, số, _, -.

from datetime import datetime

import cv2
import numpy as np

THU_MUC_ANH = GOC / "anh"


def duong_dan_anh(thoi_diem, ma_sp, ma_loi, duoi=".jpg"):
    """anh/2026-10-09/081502_123_SN0042_thieu_vat.jpg  (thư mục theo ngày, tên sắp xếp được theo giờ)"""
    ngay = thoi_diem.strftime("%Y-%m-%d")
    ten = f"{thoi_diem.strftime('%H%M%S_%f')[:-3]}_{ma_sp}_{ma_loi}{duoi}"
    return f"{ngay}/{ten}"                          # đường dẫn TƯƠNG ĐỐI, lưu vào CSDL


def luu_anh(anh, tuong_doi, chat_luong=90):
    dich = THU_MUC_ANH / tuong_doi
    dich.parent.mkdir(parents=True, exist_ok=True)
    if dich.suffix == ".jpg":
        ok = cv2.imwrite(str(dich), anh, [cv2.IMWRITE_JPEG_QUALITY, chat_luong])
    else:
        ok = cv2.imwrite(str(dich), anh, [cv2.IMWRITE_PNG_COMPRESSION, 3])
    if not ok:
        raise OSError(f"Không ghi được {dich}")
    return dich.stat().st_size


# Ảnh giả 1280x1024 có vân bề mặt (giống ảnh thật hơn một mảng phẳng)
rng = np.random.default_rng(0)
anh = cv2.GaussianBlur(rng.integers(90, 160, (1024, 1280), dtype=np.uint8), (5, 5), 0)
cv2.circle(anh, (640, 512), 300, 210, -1)
cv2.line(anh, (500, 400), (760, 470), 60, 3)        # vết xước
anh = cv2.cvtColor(anh, cv2.COLOR_GRAY2BGR)

t = datetime(2026, 10, 9, 8, 15, 2, 123000)
print("Đường dẫn:", duong_dan_anh(t, "SN0042", "xuoc_be_mat"))
print(f"Ảnh thô (BGR)  : {anh.nbytes / 1024:8.0f} KB")
for duoi, cl in [(".png", None), (".jpg", 95), (".jpg", 90), (".jpg", 75)]:
    td = duong_dan_anh(t, "SN0042", f"thu_{cl or 'png'}", duoi)
    kb = luu_anh(anh, td, cl or 0) / 1024
    doc_lai = cv2.imread(str(THU_MUC_ANH / td))
    sai_lech = np.abs(doc_lai.astype(int) - anh.astype(int)).max()
    print(f"{duoi[1:].upper():4} {str(cl or ''):>3}       : {kb:8.0f} KB   sai lệch lớn nhất {sai_lech}")

Đường dẫn: 2026-10-09/081502_123_SN0042_xuoc_be_mat.jpg
Ảnh thô (BGR)  :     3840 KB
PNG            :      697 KB   sai lệch lớn nhất 0
JPG   95       :      272 KB   sai lệch lớn nhất 8
JPG   90       :      196 KB   sai lệch lớn nhất 18
JPG   75       :      121 KB   sai lệch lớn nhất 38

Với ảnh này, PNG không mất mát nhưng nặng gấp khoảng 3,5 lần JPEG 90. Lựa chọn thực tế:

JPEG 75 sai lệch tới 38 mức xám (thường ở các cạnh sắc như mép vật, vết xước): nếu bạn chạy lại thuật toán ngưỡng trên ảnh này, kết quả có thể khác lúc kiểm tra thật. Camera đơn sắc thì lưu ảnh xám một kênh (dữ liệu thô nhỏ hơn 3 lần so với chuyển sang BGR). Nhiều trạm lưu cả hai: PNG ảnh gốc cho NG, JPEG ảnh kết quả cho NG và một tỷ lệ nhỏ mẫu OK.

6. Tự dọn ảnh: theo tuổi và theo dung lượng

Hai lớp bảo vệ chạy cùng nhau:

1. Theo tuổi: giữ N ngày (theo yêu cầu truy xuất của khách hàng, ví dụ 30 hay 90 ngày). Dựa vào tên thư mục, không dựa vào ngày sửa file (ngày sửa thay đổi khi chép, khi khôi phục).
2. Theo dung lượng trống: dù tuổi chưa tới, nếu ổ trống dưới ngưỡng (ví dụ 20%) thì xóa thư mục cũ nhất cho tới khi đủ, nhưng không bao giờ xóa vài ngày gần nhất, và phải báo động để người vận hành biết dữ liệu đang bị cắt sớm.

Hàm đo dung lượng được truyền vào làm tham số (mặc định shutil.disk_usage), nhờ vậy demo dùng một "ổ giả" 12 MB để kết quả lặp lại được trên mọi máy.

import re
import shutil
from collections import namedtuple
from datetime import date, timedelta

MAU_NGAY = re.compile(r"^\d{4}-\d{2}-\d{2}$")


def cac_thu_muc_ngay(goc):
    """Các thư mục tên YYYY-MM-DD trong goc, cũ nhất trước. Thư mục tên khác: không đụng tới."""
    kq = [(date.fromisoformat(p.name), p) for p in goc.iterdir() if p.is_dir() and MAU_NGAY.match(p.name)]
    return sorted(kq)


def xoa_mot_ngay(con, p):
    so_file = sum(1 for _ in p.iterdir())
    shutil.rmtree(p)
    with con:                                       # đánh dấu trong CSDL: ảnh không còn
        n = con.execute("UPDATE kiem_tra SET duong_dan_anh = NULL WHERE duong_dan_anh LIKE ?",
                        (p.name + "/%",)).rowcount
    return so_file, n


def don_theo_tuoi(con, goc, giu_ngay, hom_nay):
    han = hom_nay - timedelta(days=giu_ngay)
    da_xoa = []
    for ngay, p in cac_thu_muc_ngay(goc):
        if ngay < han:
            xoa_mot_ngay(con, p)
            da_xoa.append(p.name)
    return da_xoa


def don_theo_dung_luong(con, goc, trong_toi_thieu_pct, giu_it_nhat=3, dung_luong=shutil.disk_usage):
    da_xoa = []
    thu_muc = cac_thu_muc_ngay(goc)
    while True:
        du = dung_luong(goc)
        trong_pct = 100 * du.free / du.total
        if trong_pct >= trong_toi_thieu_pct:
            return da_xoa, trong_pct
        if len(thu_muc) <= giu_it_nhat:              # không xóa sạch: báo động cho người vận hành
            print(f"  CẢNH BÁO: ổ vẫn chỉ trống {trong_pct:.1f}% nhưng chỉ còn {len(thu_muc)} ngày ảnh")
            return da_xoa, trong_pct
        _, p = thu_muc.pop(0)
        xoa_mot_ngay(con, p)
        da_xoa.append(p.name)


# --- Dựng 40 ngày ảnh giả (mỗi ngày 3 ảnh x 100 KB) và bản ghi trỏ tới chúng
rng = np.random.default_rng(2)
hom_nay = date(2026, 10, 9)
for k in range(40):
    ngay = hom_nay - timedelta(days=k)
    for j in range(3):
        td = f"{ngay.isoformat()}/050000_{j:03d}_SN{k:03d}{j}_thieu_vat.jpg"
        f = THU_MUC_ANH / td
        f.parent.mkdir(parents=True, exist_ok=True)
        f.write_bytes(rng.bytes(100 * 1024))
        con.execute(SQL_THEM, (f"{ngay.isoformat()}T05:00:00.000", "VIS-01", f"SN{k:03d}{j}", "NG",
                               "thieu_vat", 9.8, 41.0, "VONG_DEM_10", td))
con.commit()
print("Ban đầu:", len(cac_thu_muc_ngay(THU_MUC_ANH)), "ngày ảnh")

# 1) Theo tuổi: giữ 30 ngày
xoa = don_theo_tuoi(con, THU_MUC_ANH, giu_ngay=30, hom_nay=hom_nay)
print(f"Theo tuổi: xóa {len(xoa)} ngày, từ {xoa[0]} đến {xoa[-1]}")

# 2) Theo dung lượng: "ổ giả" 12 MB, cần trống ít nhất 40%
DungLuong = namedtuple("DungLuong", "total used free")


def o_dia_gia(goc, tong=12 * 1024 * 1024):
    dung = sum(f.stat().st_size for f in goc.rglob("*") if f.is_file())
    return DungLuong(tong, dung, tong - dung)


xoa, trong = don_theo_dung_luong(con, THU_MUC_ANH, trong_toi_thieu_pct=40, dung_luong=o_dia_gia)
print(f"Theo dung lượng: xóa thêm {len(xoa)} ngày ({xoa[0]} … {xoa[-1]}), ổ trống {trong:.1f}%")
print("Còn lại:", len(cac_thu_muc_ngay(THU_MUC_ANH)), "ngày, cũ nhất", cac_thu_muc_ngay(THU_MUC_ANH)[0][0])
con_tro = [p for (p,) in con.execute("SELECT duong_dan_anh FROM kiem_tra WHERE duong_dan_anh IS NOT NULL")]
tro_hong = [p for p in con_tro if not (THU_MUC_ANH / p).exists()]
print("Bản ghi còn trỏ tới ảnh:", len(con_tro), "| trỏ tới file không tồn tại:", len(tro_hong))

# Ổ đĩa thật (kết quả tùy máy)
du = shutil.disk_usage(".")
print(f"Ổ thật: tổng {du.total / 1e9:.0f} GB, trống {100 * du.free / du.total:.0f}%")

Ban đầu: 40 ngày ảnh
Theo tuổi: xóa 9 ngày, từ 2026-08-31 đến 2026-09-08
Theo dung lượng: xóa thêm 11 ngày (2026-09-09 … 2026-09-19), ổ trống 40.7%
Còn lại: 20 ngày, cũ nhất 2026-09-20
Bản ghi còn trỏ tới ảnh: 60 | trỏ tới file không tồn tại: 0
Ổ thật: tổng 271 GB, trống 12%

Dòng cuối (ổ thật) tùy máy. Ba chi tiết an toàn trong code: chỉ xóa thư mục có tên đúng mẫu ngày (một thư mục "backup" ai đó tạo trong anh/ sẽ không bị đụng tới); cập nhật duong_dan_anh = NULL cho bản ghi có ảnh bị xóa để CSDL không trỏ tới file không tồn tại (kiểm tra cuối cho kết quả 0); và giữ tối thiểu vài ngày gần nhất. Nếu cần biết "đã từng có ảnh nhưng đã xóa", thêm cột trạng thái thay vì đặt NULL.

7. Ghi nền: không để ổ đĩa làm chậm cycle time

Ghi một ảnh PNG 5 MP có thể mất hàng chục đến hàng trăm ms tùy ổ đĩa, và thỉnh thoảng lâu hơn nhiều (ổ đang bận, phần mềm diệt virus quét file mới). Nếu ghi ngay trong vòng lặp kiểm tra, một lần chậm như vậy làm trễ kết quả gửi PLC và gây TIMEOUT. Giải pháp: vòng lặp kiểm tra chỉ đẩy bản ghi vào hàng đợi (vài µs) rồi đi tiếp; một thread riêng lấy ra, ghi ảnh, ghi CSDL, gom commit.

import queue
import sqlite3
import threading
import time
from datetime import datetime, timedelta


SQL_THEM_TEN = """INSERT INTO kiem_tra (thoi_gian, tram, ma_sp, ket_qua, ma_loi, gia_tri, tg_xu_ly_ms, recipe,
                  duong_dan_anh) VALUES (:thoi_gian, :tram, :ma_sp, :ket_qua, :ma_loi, :gia_tri, :tg_xu_ly_ms,
                  :recipe, :duong_dan_anh)"""          # tham số theo tên: truyền thẳng một dict


class GhiNen(threading.Thread):
    """Thread ghi đĩa: nhận bản ghi qua hàng đợi, lưu ảnh NG và gom commit SQLite."""

    def __init__(self, duong_dan_db, thu_muc_anh, kich_thuoc_hang_doi=200, gom=50, chu_ky_commit=0.5):
        super().__init__(daemon=True)
        self.hang_doi = queue.Queue(maxsize=kich_thuoc_hang_doi)
        self.duong_dan_db, self.thu_muc_anh = duong_dan_db, thu_muc_anh
        self.gom, self.chu_ky_commit = gom, chu_ky_commit
        self.so_ghi = self.so_anh = self.so_bo = 0

    def gui(self, ban_ghi, anh=None):
        """Gọi từ vòng lặp kiểm tra: KHÔNG BAO GIỜ chặn. Đầy hàng đợi thì bỏ ảnh, giữ kết quả."""
        try:
            self.hang_doi.put_nowait((ban_ghi, anh))
        except queue.Full:
            self.so_bo += 1

    def dung(self):
        self.hang_doi.put((None, None))           # "viên thuốc độc": báo thread kết thúc
        self.join(timeout=10)

    def run(self):
        con = mo_csdl(self.duong_dan_db)          # kết nối SQLite tạo TRONG thread dùng nó
        cho_commit = 0
        t_commit = time.monotonic()
        while True:
            try:
                ban_ghi, anh = self.hang_doi.get(timeout=self.chu_ky_commit)
                if ban_ghi is None:               # tín hiệu dừng
                    break
                self._ghi_mot(con, ban_ghi, anh)
                cho_commit += 1
            except queue.Empty:                   # rảnh: không có gì mới
                pass
            if cho_commit and (cho_commit >= self.gom or time.monotonic() - t_commit >= self.chu_ky_commit):
                con.commit()                      # gom: tối đa 50 dòng hoặc 0,5 s một lần commit
                cho_commit, t_commit = 0, time.monotonic()
        con.commit()
        con.close()

    def _ghi_mot(self, con, ban_ghi, anh):
        if anh is not None:
            try:
                luu_anh(anh, ban_ghi["duong_dan_anh"])
                self.so_anh += 1
            except OSError:
                ban_ghi["duong_dan_anh"] = None   # ổ đầy / lỗi ghi: mất ảnh nhưng KHÔNG mất kết quả
        con.execute(SQL_THEM_TEN, ban_ghi)
        self.so_ghi += 1


# Vòng lặp "kiểm tra" giả: 300 sản phẩm, cứ 50 cái có 1 NG kèm ảnh
ghi = GhiNen(GOC / "ket_qua.db", THU_MUC_ANH)
ghi.start()
anh_ng = anh[::2, ::2].copy()                     # ảnh 640x512 từ phần trước
t_sp = datetime(2026, 10, 9, 14, 0, 0)
thoi_gian_gui = []
for i in range(300):
    ng = i % 50 == 49
    ma_sp = f"SN9{i:04d}"
    ban_ghi = {"thoi_gian": t_sp.isoformat(timespec="milliseconds"), "tram": "VIS-01", "ma_sp": ma_sp,
               "ket_qua": "NG" if ng else "OK", "ma_loi": "thieu_vat" if ng else None, "gia_tri": 10.0,
               "tg_xu_ly_ms": 40.0, "recipe": "VONG_DEM_10",
               "duong_dan_anh": duong_dan_anh(t_sp, ma_sp, "thieu_vat") if ng else None}
    t0 = time.perf_counter()
    ghi.gui(ban_ghi, anh_ng if ng else None)
    thoi_gian_gui.append(time.perf_counter() - t0)
    t_sp += timedelta(seconds=1)
    time.sleep(0.002)                             # giả lập thời gian chụp + xử lý của trạm
ghi.dung()

print(f"Đã ghi {ghi.so_ghi} bản ghi, {ghi.so_anh} ảnh NG, bỏ {ghi.so_bo}")
tg = sorted(thoi_gian_gui)
print(f"Mỗi lần gửi: trung vị {tg[len(tg) // 2] * 1e6:.0f} µs, lớn nhất {tg[-1] * 1e6:.0f} µs")
n = con.execute("SELECT COUNT(*) FROM kiem_tra WHERE thoi_gian >= '2026-10-09T14' AND ket_qua = 'NG'").fetchone()[0]
print("Đọc lại từ kết nối khác: NG chiều 2026-10-09 =", n)

Đã ghi 300 bản ghi, 6 ảnh NG, bỏ 0
Mỗi lần gửi: trung vị 32 µs, lớn nhất 1559 µs
Đọc lại từ kết nối khác: NG chiều 2026-10-09 = 6

Thời gian mỗi lần gửi tùy máy, nhưng cỡ µs đến vài chục µs ở trung vị. Giá trị lớn nhất có thể lên vài ms: thread trong Python dùng chung một khóa GIL, nên khi thread ghi đang chạy code Python, vòng lặp chính phải chờ tới lượt. Vẫn nhỏ hơn rất nhiều so với ghi trực tiếp; nếu cần chặt chẽ hơn nữa thì dùng tiến trình riêng (multiprocessing). Các quyết định thiết kế trong GhiNen:

• Hàng đợi có giới hạn (maxsize) và put_nowait: nếu ổ đĩa chết, hàng đợi đầy và trạm bỏ bản ghi, đếm số lần bỏ thay vì ăn hết RAM rồi treo. Kết quả OK/NG vẫn đã gửi PLC, nên đây là chọn lựa an toàn; hãy báo động khi so_bo tăng.
• Kết nối SQLite tạo trong chính thread ghi: mặc định đối tượng kết nối không được dùng chéo giữa các thread (check_same_thread). Đọc báo cáo từ thread khác thì mở kết nối khác, WAL cho phép điều đó.
• Lỗi ghi ảnh không làm mất kết quả: bắt OSError, đặt đường dẫn ảnh NULL, vẫn ghi bản ghi.
• Dừng sạch: gửi None vào hàng đợi, thread ghi nốt những gì đang chờ, commit, đóng CSDL. Gọi dung() khi tắt chương trình, nếu không các bản ghi cuối chưa commit sẽ mất.
• Tham số theo tên (:thoi_gian) cho phép truyền thẳng một dict, không sợ nhầm thứ tự cột.

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

9. Bài tập

for ngay_gio, tong, ng, pct in con.execute("""
        SELECT strftime('%Y-%m-%d %Hh', thoi_gian) AS gio, COUNT(*) AS tong,
               SUM(ket_qua = 'NG') AS ng, ROUND(100.0 * SUM(ket_qua = 'NG') / COUNT(*), 2) AS pct
        FROM kiem_tra
        WHERE thoi_gian >= '2026-10-06' AND thoi_gian < '2026-10-09'
        GROUP BY gio HAVING tong >= 100
        ORDER BY pct DESC, gio LIMIT 3"""):
    print(ngay_gio, tong, ng, f"{pct}%")

2026-10-07 11h 180 10 5.56%
2026-10-07 12h 180 10 5.56%
2026-10-07 13h 180 10 5.56%

sp_moi_ngay = 16 * 3600 / 2                     # 2 ca x 8 giờ, 2 s/sản phẩm
mb_ng = sp_moi_ngay * 0.03 * 2.5                # 3% NG, PNG khoảng 2,5 MB
mb_ok = sp_moi_ngay * 0.01 * 0.6                # lưu mẫu 1% OK, JPEG 90 khoảng 0,6 MB
gb_ngay = (mb_ng + mb_ok) / 1024
dung_duoc = 500 * 0.8                           # ổ 500 GB, chừa 20% trống
print(f"{sp_moi_ngay:.0f} SP/ngày → {gb_ngay:.2f} GB ảnh/ngày → giữ được khoảng {dung_duoc / gb_ngay:.0f} ngày")

28800 SP/ngày → 2.28 GB ảnh/ngày → giữ được khoảng 176 ngày

Tóm tắt

• SQLite: một file, có sẵn trong Python, giao dịch an toàn; bật WAL + synchronous=NORMAL; chỉ mục theo thời gian.
• Ghi bằng tham số (? hoặc :ten), gom nhiều dòng trong một giao dịch; truy vấn theo khoảng thời gian để dùng được chỉ mục.
• Ảnh NG: thư mục theo ngày, tên file tự mô tả, đường dẫn tương đối trong CSDL; PNG cho ảnh gốc cần phân tích lại, JPEG 85–90 cho ảnh xem.
• Dọn ảnh theo tuổi (tên thư mục) và theo dung lượng trống (shutil.disk_usage), cập nhật CSDL khi xóa, giữ tối thiểu vài ngày, báo động.
• Mọi việc ghi đĩa ra thread nền với hàng đợi có giới hạn; kết nối SQLite tạo trong thread dùng nó; dừng sạch để không mất bản ghi cuối.

Bài tiếp theo: Dự án chặng 9: trạm vision hoàn chỉnh với PLC mô phỏng, ghép Modbus TCP, xử lý ảnh, SQLite và ảnh NG thành một chương trình chạy ổn định.