kruthika198/abnormal-behavior-detection

1import gradio as gr2import cv23import numpy as np4import os5from ultralytics import YOLO6from collections import defaultdict, deque7 8# ========== 📧 Email Alert Function ==========9import smtplib10from email.mime.text import MIMEText11 12def send_email(subject, body, sender_email, receiver_email, app_password):13    try:14        msg = MIMEText(body)15        msg["Subject"] = subject16        msg["From"] = sender_email17        msg["To"] = receiver_email18 19        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:20            server.login(sender_email, app_password)21            server.send_message(msg)22        return True23    except Exception as e:24        print(f"❌ Email Error: {e}")25        return False26 27# ========== 🔍 Main Processing ==========28def process_video(video_file, crowd_thresh, run_speed_thresh, run_frames, kick_thresh, kick_frames, crawl_ratio, sender_email, receiver_email, app_pass):29    try:30        video_path = video_file if isinstance(video_file, str) else video_file.name31        cap = cv2.VideoCapture(video_path)32        fps = cap.get(cv2.CAP_PROP_FPS)33        width = int(cap.get(3))34        height = int(cap.get(4))35 36        if fps == 0 or width == 0 or height == 0:37            raise ValueError("📼 Video metadata unreadable or invalid file.")38 39        model = YOLO("yolov8n.pt")40        out_path = "output.avi"41        out = cv2.VideoWriter(out_path, cv2.VideoWriter_fourcc(*"XVID"), fps, (width, height))42 43        track_history = defaultdict(lambda: deque(maxlen=10))44        consec_run = defaultdict(int)45        consec_kick = defaultdict(int)46        run_alerted = set()47        kick_alerted = set()48        crawl_alerted = set()49        crowd_alerted_frames = set()50        frame_num = 051        font = cv2.FONT_HERSHEY_SIMPLEX52        prev_gray = None53 54        while cap.isOpened():55            ret, frame = cap.read()56            if not ret:57                break58 59            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)60            results = model.track(frame, persist=True, verbose=False)[0]61 62            people_count = 063            run_ids, kick_ids, crawl_ids = set(), set(), set()64 65            if results.boxes is not None:66                for box in results.boxes:67                    if int(box.cls[0]) != 0 or box.id is None:68                        continue69                    track_id = int(box.id[0])70                    people_count += 171                    x1, y1, x2, y2 = map(int, box.xyxy[0])72                    cx, cy = (x1 + x2) // 2, (y1 + y2) // 273                    track_history[track_id].append((frame_num, cx, cy))74 75                    # 🧎 Crawling76                    if (y2 - y1) < height * crawl_ratio:77                        crawl_ids.add(track_id)78                        if track_id not in crawl_alerted:79                            send_email("🧎 Crawling Detected", f"ID: {track_id}", sender_email, receiver_email, app_pass)80                            crawl_alerted.add(track_id)81 82                    # 🏃 Running83                    if len(track_history[track_id]) >= 5:84                        f0, x0, y0 = track_history[track_id][0]85                        f1, x1_, y1_ = track_history[track_id][-1]86                        dt = (f1 - f0) / fps87                        dist = np.linalg.norm([x1_ - x0, y1_ - y0])88                        speed = dist / dt if dt > 0 else 089 90                        if speed > run_speed_thresh:91                            consec_run[track_id] += 192                        else:93                            consec_run[track_id] = 094 95                        if consec_run[track_id] >= run_frames:96                            run_ids.add(track_id)97                            if track_id not in run_alerted:98                                send_email("🏃 Running Detected", f"ID: {track_id}", sender_email, receiver_email, app_pass)99                                run_alerted.add(track_id)100 101                    # 🦵 Kicking with Optical Flow102                    if prev_gray is not None:103                        leg_top = y2 - (y2 - y1) // 3104                        leg_roi = gray[leg_top:y2, x1:x2]105                        prev_leg_roi = prev_gray[leg_top:y2, x1:x2]106 107                        if leg_roi.size > 0 and prev_leg_roi.size > 0:108                            flow = cv2.calcOpticalFlowFarneback(prev_leg_roi, leg_roi, None, 0.5, 3, 15, 3, 5, 1.2, 0)109                            mag, _ = cv2.cartToPolar(flow[..., 0], flow[..., 1])110                            motion_mag = np.mean(mag)111 112                            if motion_mag > kick_thresh:113                                consec_kick[track_id] += 1114                            else:115                                consec_kick[track_id] = 0116 117                            if consec_kick[track_id] >= kick_frames:118                                kick_ids.add(track_id)119                                if track_id not in kick_alerted:120                                    send_email("🦵 Kicking Detected", f"ID: {track_id}", sender_email, receiver_email, app_pass)121                                    kick_alerted.add(track_id)122 123            # 👥 Crowd Alert124            if people_count > crowd_thresh and frame_num not in crowd_alerted_frames:125                send_email("⚠️ Crowd Alert", f"{people_count} people detected", sender_email, receiver_email, app_pass)126                crowd_alerted_frames.add(frame_num)127 128            out.write(frame)129            prev_gray = gray.copy()130            frame_num += 1131 132        cap.release()133        out.release()134 135        if frame_num == 0:136            raise ValueError("No frames processed. Empty video?")137 138        return out_path139    except Exception as e:140        return f"❌ Error: {e}"141 142# ========== 🖼️ UI Setup ==========143inputs = [144    gr.Video(label="📹 Upload CCTV Footage", format="mp4"),145    gr.Number(label="👥 Crowd Threshold", value=5),146    gr.Number(label="🏃 Running Speed Threshold", value=90.0),147    gr.Number(label="🏃 Running Frame Count Threshold", value=4),148    gr.Number(label="🦵 Kicking Motion Threshold", value=2.5),149    gr.Number(label="🦵 Kicking Frame Count Threshold", value=2),150    gr.Number(label="🧎 Crawling Height Ratio", value=0.3),151    gr.Textbox(label="📧 Sender Email"),152    gr.Textbox(label="📨 Receiver Email"),153    gr.Textbox(label="🔐 Gmail App Password", type="password")154]155 156outputs = gr.Video(label="🎬 Output Video")157 158interface = gr.Interface(159    fn=process_video,160    inputs=inputs,161    outputs=outputs,162    title="🚨 Abnormal Activity Detection",163    description="Upload a video and enter thresholds to detect crowding, running, kicking, and crawling. Email alerts will be sent for detected activities."164)165 166interface.launch()167