The lastest ARC Model for my Final Year Thesis [v1]

action_recognition.py

from transformers import AutoProcessor, AutoModelForImageClassification
from PIL import Image
import cv2
import numpy as np
import winsound
import time
import json, os
import torch
# Temporal smoothing buffer
from collections import deque
prediction_buffer = deque(maxlen=10)  # last 5 frames

# -------------------------------------------------
# MODEL LOADING
# -------------------------------------------------
model_path = r'C:\Users\SURFACE\action_recognition_model'
processor = AutoProcessor.from_pretrained(model_path)
model = AutoModelForImageClassification.from_pretrained(model_path)

# Class labels
id2label = model.config.id2label

# -------------------------------------------------
# ALERT CONFIGURATION
# -------------------------------------------------
SUSPICIOUS_ACTIONS = {"fighting", "calling", "running"}  # define what is suspicious
ALERT_THRESHOLD = 10          # consecutive frames
ALERT_COOLDOWN_SEC = 3        # seconds between alarms

alert_counter = 0
current_alert_action = None
last_alert_time = 0.0
# -------------------------------------------------

# Open webcam
# -------------------------------------------------
# CAMERA SOURCE SELECTION (short)
# -------------------------------------------------
CONFIG = "camera.json"
# Load last source if exists
last = json.load(open(CONFIG)) if os.path.exists(CONFIG) else None

print("\n1) Use last camera" if last else "\n1) Use last camera (none saved)")
print("2) Enter stream URL")
print("3) Use built-in camera")

choice = input("Select 1/2/3: ").strip()

if choice == "1" and last:
    source = int(last["source"]) if str(last["source"]).isdigit() else last["source"]

elif choice == "2":
    source = input("Enter stream URL: ").strip()

elif choice == "3":
    source = int(input("Camera index (default 0): ").strip() or "0")

else:
    raise RuntimeError("Invalid choice")

cap = cv2.VideoCapture(source)

if not cap.isOpened():
    raise RuntimeError("Failed to open camera")

# Save for next session
json.dump({"source": str(source)}, open(CONFIG, "w"))

# -------------------------------------------------

prev_gray = None
motion_threshold = 50000  # adjust if needed

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

    # -------------------------------------------------
    # RECONNECT IF STREAM FAILS
    # -------------------------------------------------
    if not ret:
        print("Reconnecting...")
        cap.release()
        time.sleep(0.5)
        cap = cv2.VideoCapture(source)
        if cap.isOpened():
            print("Camera connected.")
        continue

    # -------------------------------------------------
    # MOTION DYNAMICS (FRAME-DIFFERENCE MOTION SCORE)
    # -------------------------------------------------
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    motion_detected = True

    if prev_gray is not None:
        frame_diff = cv2.absdiff(prev_gray, gray)
        motion_score = frame_diff.sum()
        if motion_score < motion_threshold:
            motion_detected = False

    prev_gray = gray

    # -------------------------------------------------
    # MODEL INFERENCE
    # -------------------------------------------------
    image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
    inputs = processor(images=image, return_tensors="pt")
    outputs = model(**inputs)
    logits = outputs.logits
    probs = torch.softmax(logits, dim=-1)[0]
    conf, idx = torch.max(probs, dim=-1)

    conf = conf.item()
    predicted_class_label = id2label[idx.item()]

    # Reject low confidence predictions
    CONF_THRESHOLD = 0.15

    if conf < CONF_THRESHOLD:
        predicted_class_label = "NO_ACTIVITY"
    
    print(predicted_class_label, conf)
    

    # -------------------------------------------------
    # TEMPORAL SMOOTHING (MAJORITY VOTE)
    # -------------------------------------------------
    # -------------------------------------------------
    # TEMPORAL SMOOTHING (MAJORITY VOTE)
    # -------------------------------------------------
    if predicted_class_label != "NO_ACTIVITY":
        prediction_buffer.append(predicted_class_label)

    if len(prediction_buffer) > 0:
        predicted_class_label = max(set(prediction_buffer), key=prediction_buffer.count)
    else:
        predicted_class_label = "NO_ACTIVITY"

    print(f"Predicted action class: {predicted_class_label}")

    # -------------------------------------------------
    # ALERT LOGIC
    # -------------------------------------------------
    if predicted_class_label in SUSPICIOUS_ACTIONS and motion_detected:
        if current_alert_action == predicted_class_label:
            alert_counter += 1
        else:
            current_alert_action = predicted_class_label
            alert_counter = 1

        if alert_counter >= ALERT_THRESHOLD:
            now = time.time()
            if now - last_alert_time >= ALERT_COOLDOWN_SEC:
                for _ in range(3):
                    winsound.Beep(1000, 250)
                    time.sleep(0.05)

                print(f"ALERT TRIGGERED: {predicted_class_label}")
                last_alert_time = now
    else:
        alert_counter = 0
        current_alert_action = None

    # -------------------------------------------------
    # VISUAL OVERLAY
    # -------------------------------------------------
    if alert_counter >= ALERT_THRESHOLD:
        cv2.putText(
            frame,
            f"ALERT: {predicted_class_label.upper()}",
            (30, 40),
            cv2.FONT_HERSHEY_SIMPLEX,
            1,
            (0, 0, 255),
            2
        )

    cv2.imshow("Webcam Feed", frame)

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()