import time
from time import perf_counter
import cv2
import torch
import [Link] as nn
import [Link] as models
import [Link] as transforms
from PIL import Image
import numpy as np
import dalsa_cam, os
# ===== Config =====
checkpoint_path = r"C:\Users\Raiso\Train_pytorch\snack_classifier.pth"
SERVER_INDEX = 1
BAYER_PATTERN = "RGGB"
NUM_BUFFERS = 4 # Nếu .pyd cũ không hỗ trợ sẽ tự bỏ qua
print("dalsa_cam from:", [Link](dalsa_cam.__file__))
# ===== Load checkpoint =====
ckpt = [Link](checkpoint_path, map_location="cpu")
class_names = [Link]("class_names", ["Bad Bag", "Good Bag"])
print("📂 Class names:", class_names)
print("🔢 Number of classes:", len(class_names))
# ===== Build model =====
model = models.resnet18(weights=None)
[Link] = [Link]([Link].in_features, len(class_names))
model.load_state_dict(ckpt["model_state_dict"])
[Link]()
print("✅ Model loaded successfully!")
# ===== Transform =====
transform = [Link]([
[Link]((224, 224)),
[Link](),
])
# ===== Helpers =====
def predict_frame_rgb(rgb_np: [Link]):
img = [Link](rgb_np)
img = transform(img).unsqueeze(0)
with torch.no_grad():
probs = [Link](model(img), dim=1)
conf, pred = [Link](probs, 1)
return class_names[[Link]()], float([Link]())
def put_lines(img, lines, org=(20, 30), step=26, scale=0.7):
x, y = org
for line in lines:
[Link](img, line, (x, y), cv2.FONT_HERSHEY_SIMPLEX, scale, (255, 255, 0), 2, cv2.LINE_AA)
y += step
def main():
# ---- Open camera (try new signature; fallback to old) ----
try:
dalsa_cam.open(server_index=SERVER_INDEX, bayer=BAYER_PATTERN,
num_buffers=NUM_BUFFERS)
except TypeError:
dalsa_cam.open(server_index=SERVER_INDEX, bayer=BAYER_PATTERN, wbR=1.0, wbG=1.0,
wbB=1.0, gamma=1.0)
info = dalsa_cam.info()
print("INFO:", info)
HAS_GRAB = hasattr(dalsa_cam, "start_grab") and hasattr(dalsa_cam, "pop_frame")
if HAS_GRAB:
dalsa_cam.start_grab()
print("▶ GRAB mode (trigger). Waiting frames...")
else:
print("▶ SNAP fallback (no grab API in this .pyd).")
# ---- Counters ----
capture_count = 0 # số lần chụp hợp lệ (có frame)
infer_count = 0 # số lần infer đã chạy
per_class = {name: 0 for name in class_names}
t0_run = perf_counter()
ema_fps = None
ema_alpha = 0.2
# ---- UI ----
win = "DALSA - Prediction"
[Link](win, cv2.WINDOW_NORMAL)
[Link](win, 960, 540)
i=0
try:
while True:
loop_t0 = perf_counter()
# ---- Capture ----
t_cap0 = perf_counter()
if HAS_GRAB:
arr = dalsa_cam.pop_frame(timeout_ms=300) # None nếu chưa có trigger
if arr is None:
# để cửa sổ không "Not Responding", vẫn xử lý phím
if [Link](1) & 0xFF == ord('q'):
break
continue
rgb = [Link](arr)
else:
rgb = [Link](dalsa_cam.snap_and_get(save=False))
t_cap1 = perf_counter()
capture_count += 1
# ---- Inference ----
t_inf0 = perf_counter()
label, conf = predict_frame_rgb(rgb)
t_inf1 = perf_counter()
infer_count += 1
per_class[label] = per_class.get(label, 0) + 1
# ---- Overlay & show ----
bgr = [Link](rgb, cv2.COLOR_RGB2BGR)
[Link](bgr, f"{label} ({conf*100:.1f}%)", (20, 40),
cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2, cv2.LINE_AA)
cap_ms = (t_cap1 - t_cap0) * 1000.0
inf_ms = (t_inf1 - t_inf0) * 1000.0
tot_ms = (perf_counter() - loop_t0) * 1000.0
fps = 1000.0 / tot_ms if tot_ms > 0 else 0.0
ema_fps = fps if ema_fps is None else (1 - ema_alpha) * ema_fps + ema_alpha * fps
elapsed = max(perf_counter() - t0_run, 1e-6)
rate_per_min = infer_count / elapsed * 60.0
lines = [
f"Capture: {cap_ms:.1f} ms",
f"Infer: {inf_ms:.1f} ms",
f"Total: {tot_ms:.1f} ms | FPS~ {ema_fps:.1f}",
f"Captures: {capture_count}",
f"Inferences: {infer_count} (~{rate_per_min:.1f}/min)",
# thêm thống kê từng lớp (nếu muốn xem)
for cname in class_names:
[Link](f"{cname}: {per_class.get(cname,0)}")
put_lines(bgr, lines, org=(20, 80), step=24, scale=0.7)
[Link](win, bgr)
print(f"[{i}] {tuple([Link])} => {label} ({conf*100:.1f}%) | "
f"cap={cap_ms:.1f}ms inf={inf_ms:.1f}ms total={tot_ms:.1f}ms "
f"captures={capture_count} infers={infer_count} rpm~{rate_per_min:.1f}")
i += 1
key = [Link](1) & 0xFF
if key == ord('q'):
break
elif key == ord('r'):
capture_count = 0
infer_count = 0
per_class = {name: 0 for name in class_names}
t0_run = perf_counter()
print("↺ Counters reset.")
except KeyboardInterrupt:
print("Stop")
finally:
if HAS_GRAB:
dalsa_cam.stop_grab()
dalsa_cam.close()
[Link]()
if __name__ == "__main__":
main()