เซ็นเซอร์หมวกกันน็อค

import cv2
import time
import requests
from datetime import datetime
from ultralytics import YOLO

==========================================

1. ตั้งค่าระบบ (CONFIGURATION)

==========================================

สัญญาณภาพ: ใส่ RTSP URL กล้องวงจรปิด หรือใส่เลข 0 สำหรับ Webcam

CAMERA_SOURCE = "rtsp://admin:123456@192.168.1.100:554/stream1"

โหลดโมเดล YOLO (แนะนำไฟล์ .pt ที่ Train ตรวจจับหมวกกันน็อกโดยเฉพาะ)

MODEL_PATH = "helmet_detection.pt"

LINE Notify Access Token (นำ Token จาก notify-bot.line.me มาใส่)

LINE_TOKEN = "YOUR_LINE_NOTIFY_TOKEN_HERE"

ID ของคลาสที่ไม่สวมหมวกกันน็อก (อิงตาม Label ใน Dataset ของคุณ)

NO_HELMET_CLASS_ID = 1

ตั้งค่า Cooldown (วินาที) ไม่ให้ส่งแจ้งเตือนซ้ำซ้อนกันเกินไป

ALERT_COOLDOWN = 5

==========================================

2. ฟังก์ชันระบบ

==========================================

def send_line_notify(message, image_path=None):
"""ส่งข้อความและรูปภาพเข้า LINE Notify"""
url = "https://notify-api.line.me/api/notify"
headers = {"Authorization": f"Bearer {LINE_TOKEN}"}
data = {"message": message}

files = None
if image_path:
    files = {"imageFile": open(image_path, "rb")}

try:
    response = requests.post(url, headers=headers, data=data, files=files, timeout=10)
    if response.status_code == 200:
        print("LINE Notify: ส่งข้อมูลสำเร็จ")
    else:
        print(f"LINE Notify Error: Code {response.status_code}")
except Exception as e:
    print(f"ไม่สามารถส่ง LINE Notify ได้: {e}")

def connect_camera(source):
"""เชื่อมต่อกล้องวงจรปิดอัตโนมัติพร้อมระบบ Reconnect"""
while True:
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] กำลังเชื่อมต่อกล้อง...")
cap = cv2.VideoCapture(source)
if cap.isOpened():
print("เชื่อมต่อกล้องวงจรปิดสำเร็จ!")
return cap
print("เชื่อมต่อไม่สำเร็จ จะลองใหม่ในอีก 5 วินาที...")
time.sleep(5)

def trigger_hardware_relay():
"""จำลองการส่งสัญญาณไปยังอุปกรณ์ภายนอก (เช่น สั่ง Arduino เปิดไซเรน หรือปิดไม้กั้น)"""
print("HARDWARE TRIGGER: ส่งสัญญาณแจ้งเตือนไปยังบอร์ดควบคุม (Siren/Buzzer ON)")

==========================================

3. ลูปการทำงานหลัก (MAIN LOOP)

==========================================

model = YOLO(MODEL_PATH)
cap = connect_camera(CAMERA_SOURCE)
last_alert_time = 0

print("=== ระบบตรวจจับหมวกกันน็อกอัตโนมัติพร้อมทำงาน ===")

while True:
try:
ret, frame = cap.read()

    # กรณีสัญญาณกล้องดับ/หลุด ให้ทำการ Reconnect
    if not ret:
        print("สัญญาณกล้องขาดหาย กำลังเชื่อมต่อใหม่...")
        cap.release()
        cap = connect_camera(CAMERA_SOURCE)
        continue

    # ส่งภาพเข้าโมเดลตรวจจับ
    results = model(frame, conf=0.5, verbose=False)
    boxes = results[0].boxes

    has_no_helmet = False

    # วนลูปเช็กว่าวัตถุที่เจอเข้าข่าย 'ไม่ใส่หมวกกันน็อก' หรือไม่
    if boxes is not None:
        for box in boxes:
            cls_id = int(box.cls[0])
            if cls_id == NO_HELMET_CLASS_ID:
                has_no_helmet = True
                break

    # Action เมื่อพบผู้ไม่สวมหมวกกันน็อก
    if has_no_helmet:
        current_time = time.time()

        # เช็ก Cooldown ก่อนส่งแจ้งเตือน
        if current_time - last_alert_time > ALERT_COOLDOWN:
            now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            timestamp_file = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"no_helmet_{timestamp_file}.jpg"

            # วาด Bounding Box และเก็บบันทึกไฟล์ภาพ
            annotated_frame = results[0].plot()
            cv2.imwrite(filename, annotated_frame)
            print(f"\n⚠️ [{now_str}] ตรวจพบผู้ไม่สวมหมวกกันน็อก! บันทึกไฟล์: {filename}")

            # 1. สั่งงานฮาร์ดแวร์
            trigger_hardware_relay()

            # 2. ส่งภาพและข้อความแจ้งเตือนผ่าน LINE Notify
            alert_msg = f"\n[แจ้งเตือน AI]\nตรวจพบผู้ไม่สวมหมวกกันน็อก!\nเวลา: {now_str}"
            send_line_notify(alert_msg, filename)

            last_alert_time = current_time

except Exception as e:
    print(f"เกิดข้อผิดพลาดในระบบ: {e}")
    time.sleep(1)