# -*- coding: utf-8 -*- """app.py Automated Fire and Accident Detection for CCTV with Freshdesk Ticket Creation """ import cv2 import os import PIL.Image as Image import gradio as gr import numpy as np from ultralytics import YOLO import requests import json from datetime import datetime import tempfile import torch import uuid from huggingface_hub import upload_file HF_TOKEN = os.getenv("HF_TOKEN") REPO_ID = "Zynaly/Surveillance-Intelligent-Camera" FRESHDESK_DOMAIN = "umtedu-help.freshdesk.com" API_KEY = os.getenv("FRESHDESK_API_KEY", "cGyaT4wMNm4F0FEufpWs") BASE_URL = "https://huggingface.co/spaces/Zynaly/Surveillance-Intelligent-Camera/tree/main" MEDIA_DIR = "media" FIRE_DIR = os.path.join(MEDIA_DIR, "fire") ACCIDENT_DIR = os.path.join(MEDIA_DIR, "accidents") os.makedirs(FIRE_DIR, exist_ok=True) os.makedirs(ACCIDENT_DIR, exist_ok=True) # Fixed thresholds for automated detection FIRE_CONF_THRESHOLD = 0.25 FIRE_IOU_THRESHOLD = 0.45 ACCIDENT_CONF_THRESHOLD = 0.3 ACCIDENT_IOU_THRESHOLD = 0.55 # Load models with explicit task definition fire_model = YOLO("fire.pt", task="detect") # Fire detection model accident_model = YOLO("best.pt", task="detect") # Accident detection model def save_image(image, incident_type): try: # Create filename and local path timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") unique_id = str(uuid.uuid4())[:8] filename = f"{incident_type}_{timestamp}_{unique_id}.jpg" local_path = os.path.join("media", incident_type, filename) # Save image locally first image.save(local_path) # Upload image to Hugging Face Space repo remote_path = f"media/{incident_type}/{filename}" url = upload_file( path_or_fileobj=local_path, path_in_repo=remote_path, repo_id=REPO_ID, token=HF_TOKEN, repo_type="space" # <=== IMPORTANT! ) print(f"✅ Image uploaded to Hugging Face: {url}") return url except Exception as e: print(f"❌ Error uploading image: {str(e)}") return None # Function to create Freshdesk ticket def create_freshdesk_ticket(incident_type, confidence_score, img): # Save the image to the appropriate directory and get its local path image_url = None image_path = None if incident_type.lower() == "fire incident": image_url = save_image(img, "fire") image_path = os.path.join(MEDIA_DIR, "fire", os.path.basename(image_url.split("/")[-1])) elif incident_type.lower() == "accident incident": image_url = save_image(img, "accidents") image_path = os.path.join(MEDIA_DIR, "accidents", os.path.basename(image_url.split("/")[-1])) elif incident_type.lower() == "fire and accident incident": fire_url = save_image(img, "fire") accident_url = save_image(img, "accidents") image_url = fire_url or accident_url image_path = os.path.join(MEDIA_DIR, "fire" if fire_url else "accidents", os.path.basename(image_url.split("/")[-1])) # Shortened subject subject = f"""{incident_type} Detected - Confidence: {confidence_score*100:.1f}% Details: 1. Address: Hafeez Centre, Gulberg, Lahore 2. Image URL: {image_url or 'https://example.com/roboi.jpg'} """ # Detailed description description = f""" {incident_type} is critical. Details: 1. Address: 123 Main Street, Lahore 2. Phone: 923013225853 3. Confidence Score: {confidence_score*100:.1f}% 4. Image URL: {image_url or 'https://example.com/roboi.jpg'} 5. Incident Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} """ ticket_data = { "email": "safe.city@example.com", "subject": subject, "description": description, "priority": 4, # Urgent "status": 2 # Open } # Create ticket url = f"https://{FRESHDESK_DOMAIN}/api/v2/tickets" headers = {"Content-Type": "application/json"} response = requests.post( url, auth=(API_KEY, "X"), headers=headers, data=json.dumps(ticket_data) ) if response.status_code == 201: ticket = response.json() ticket_id = ticket.get('id') print(f"✅ Ticket created successfully: Ticket ID {ticket_id}") print(json.dumps(ticket, indent=2)) # Attach image to ticket if image_path exists if image_path and os.path.exists(image_path): attachment_url = f"https://{FRESHDESK_DOMAIN}/api/v2/tickets/{ticket_id}/attachments" try: with open(image_path, 'rb') as f: files = {'attachments[]': (os.path.basename(image_path), f, 'image/jpeg')} # Do not set Content-Type header; let requests handle it attachment_response = requests.post( attachment_url, auth=(API_KEY, "X"), files=files ) if attachment_response.status_code == 201: print(f"✅ Image attached to ticket {ticket_id}") else: print(f"❌ Failed to attach image: {attachment_response.status_code} - {attachment_response.text}") except Exception as e: print(f"❌ Error accessing image file {image_path}: {str(e)}") else: print(f"❌ Image file not found: {image_path}") return f"Ticket created for {incident_type} with ID {ticket_id}" else: print(f"❌ Failed to create ticket: {response.status_code} - {response.text}") return f"Failed to create ticket for {incident_type}: {response.status_code} - {response.text}" # Image inference function def detect_image(image): try: pil_img = image # Fire detection fire_results = fire_model.predict( source=pil_img, conf=FIRE_CONF_THRESHOLD, iou=FIRE_IOU_THRESHOLD, show_labels=True, show_conf=True, imgsz=640, verbose=False ) fire_detected = False fire_confidence = 0.0 fire_annotated_img = fire_results[0].plot() fire_confidences = [] fire_classes = [] for r in fire_results: if r.boxes: for box in r.boxes: confidence = box.conf[0].item() class_id = int(box.cls[0].item()) fire_confidences.append(confidence) fire_classes.append(class_id) if confidence >= FIRE_CONF_THRESHOLD: fire_detected = True fire_confidence = max(fire_confidence, confidence) print(f"Fire model raw confidences: {fire_confidences}, classes: {fire_classes}") # Accident detection accident_results = accident_model.predict( source=pil_img, conf=ACCIDENT_CONF_THRESHOLD, iou=ACCIDENT_IOU_THRESHOLD, show_labels=True, show_conf=True, imgsz=640, verbose=False ) accident_detected = False accident_confidence = 0.0 accident_annotated_img = accident_results[0].plot() accident_confidences = [] accident_classes = [] accident_boxes = accident_results[0].boxes if accident_boxes: for box in accident_boxes: confidence = box.conf[0].item() class_id = int(box.cls[0].item()) accident_confidences.append(confidence) accident_classes.append(class_id) if confidence >= ACCIDENT_CONF_THRESHOLD: accident_detected = True accident_confidence = max(accident_confidence, confidence) print(f"Accident model raw confidences: {accident_confidences}, classes: {accident_classes}") # Combine annotated images fire_annotated_img = np.array(fire_annotated_img) accident_annotated_img = np.array(accident_annotated_img) combined_img = Image.fromarray(np.maximum(fire_annotated_img, accident_annotated_img)) # Detection info detection_info = "Detection Results:\n" if fire_detected: detection_info += f"Fire detected with confidence: {fire_confidence*100:.1f}%\n" else: detection_info += f"No fire detected. Raw confidences: {fire_confidences}, Classes: {fire_classes}\n" if accident_detected: detection_info += f"Accident detected with confidence: {accident_confidence*100:.1f}%\n" else: detection_info += f"No accident detected. Raw confidences: {accident_confidences}, Classes: {accident_classes}\n" # Create a single Freshdesk ticket ticket_info = "" if fire_detected and not accident_detected: ticket_info = create_freshdesk_ticket("Fire Incident", fire_confidence, pil_img) elif accident_detected and not fire_detected: ticket_info = create_freshdesk_ticket("Accident Incident", accident_confidence, pil_img) elif fire_detected and accident_detected: ticket_info = create_freshdesk_ticket("Fire and Accident Incident", max(fire_confidence, accident_confidence), pil_img) else: ticket_info = "No ticket created: No incidents detected" return combined_img, detection_info, ticket_info except Exception as e: return image, f"Error during detection: {str(e)}\nRaw confidences: Fire {fire_confidences}, Accident {accident_confidences}", "No ticket created due to error" # Video processing function def detect_video(video_path): try: cap = cv2.VideoCapture(video_path) frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = int(cap.get(cv2.CAP_PROP_FPS)) output_path = tempfile.mktemp(suffix='.mp4') fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(output_path, fourcc, fps, (frame_width, frame_height)) frame_count = 0 fire_detected_once = False accident_detected_once = False fire_detection_frames = [] accident_detection_frames = [] fire_confidences_all = [] accident_confidences_all = [] fire_classes_all = [] accident_classes_all = [] while cap.isOpened(): ret, frame = cap.read() if not ret: break frame_count += 1 pil_img = Image.fromarray(frame[..., ::-1]) # Convert BGR to RGB # Fire detection fire_results = fire_model.predict( source=pil_img, conf=FIRE_CONF_THRESHOLD, iou=FIRE_IOU_THRESHOLD, show_labels=True, show_conf=True, imgsz=640, verbose=False ) fire_detected = False fire_confidence = 0.0 fire_annotated_frame = fire_results[0].plot() fire_confidences = [] fire_classes = [] for r in fire_results: if r.boxes: for box in r.boxes: confidence = box.conf[0].item() class_id = int(box.cls[0].item()) fire_confidences.append(confidence) fire_classes.append(class_id) if confidence >= FIRE_CONF_THRESHOLD: fire_detected = True fire_confidence = max(fire_confidence, confidence) fire_detection_frames.append(frame_count) fire_confidences_all.extend(fire_confidences) fire_classes_all.extend(fire_classes) # Accident detection accident_results = accident_model.predict( source=pil_img, conf=ACCIDENT_CONF_THRESHOLD, iou=ACCIDENT_IOU_THRESHOLD, show_labels=True, show_conf=True, imgsz=640, verbose=False ) accident_detected = False accident_confidence = 0.0 accident_annotated_frame = accident_results[0].plot() accident_confidences = [] accident_classes = [] accident_boxes = accident_results[0].boxes if accident_boxes: for box in accident_boxes: confidence = box.conf[0].item() class_id = int(box.cls[0].item()) accident_confidences.append(confidence) accident_classes.append(class_id) if confidence >= ACCIDENT_CONF_THRESHOLD: accident_detected = True accident_confidence = max(accident_confidence, confidence) accident_detection_frames.append(frame_count) accident_confidences_all.extend(accident_confidences) accident_classes_all.extend(accident_classes) # Combine annotated frames fire_annotated_frame = np.array(fire_annotated_frame) accident_annotated_frame = np.array(accident_annotated_frame) combined_frame = np.maximum(fire_annotated_frame, accident_annotated_frame) out.write(combined_frame) # Create a single ticket for the first detection of each incident type if fire_detected and not fire_detected_once: create_freshdesk_ticket("Fire Incident", fire_confidence, pil_img) fire_detected_once = True if accident_detected and not accident_detected_once: create_freshdesk_ticket("Accident Incident", accident_confidence, pil_img) accident_detected_once = True cap.release() out.release() detection_info = f"Video processed successfully!\n" detection_info += f"Total frames: {frame_count}\n" detection_info += f"Frames with fire detections: {len(set(fire_detection_frames))}\n" detection_info += f"Frames with accident detections: {len(set(accident_detection_frames))}\n" if fire_detection_frames: detection_info += f"Fire detection frames: {sorted(set(fire_detection_frames))[:10]}...\n" else: detection_info += f"No fire detections. Raw confidences (sample): {fire_confidences_all[:10]}, Classes: {fire_classes_all[:10]}...\n" if accident_detection_frames: detection_info += f"Accident detection frames: {sorted(set(accident_detection_frames))[:10]}...\n" else: detection_info += f"No accident detections. Raw confidences (sample): {accident_confidences_all[:10]}, Classes: {accident_classes_all[:10]}...\n" ticket_info = f"Tickets created: {'Fire' if fire_detected_once else ''}{' and ' if fire_detected_once and accident_detected_once else ''}{'Accident' if accident_detected_once else ''}." if fire_detected_once or accident_detected_once else "No tickets created: No incidents detected" return output_path, detection_info, ticket_info except Exception as e: return None, f"Error processing video: {str(e)}\nRaw confidences: Fire {fire_confidences_all[:10]}, Accident {accident_confidences_all[:10]}", "No ticket created due to error" # Create Gradio interface for CCTV automation with gr.Blocks(title="Rapid Rescue - Automated CCTV Fire and Accident Detection") as iface: gr.Markdown(""" # 🚨 Rapid Rescue - Automated CCTV Fire and Accident Detection System This AI system automatically detects fires and accidents in images and videos from CCTV feeds using two YOLO models: - YOLOv8n for fire detection (Confidence: 0.25, IoU: 0.45) - YOLOv8m for accident detection (Confidence: 0.3, IoU: 0.55) **Features:** - Fully automated detection with fixed thresholds - Creates Freshdesk tickets for detected incidents with saved image URLs - Supports both images and videos from CCTV feeds - Images saved in media/fire and media/accidents directories - Optimized for deployment on Hugging Face Spaces **Usage:** 1. Upload an image or video from a CCTV feed 2. Click process to run detection 3. View results with bounding boxes, confidence scores, class labels, and ticket creation status """) with gr.Tabs(): with gr.Tab("Image Detection"): with gr.Row(): with gr.Column(): image_input = gr.Image(type="pil", label="Upload CCTV Image") image_button = gr.Button("Detect Fire and Accidents", variant="primary") with gr.Column(): image_output = gr.Image(label="Detection Results") image_info = gr.Textbox(label="Detection Information", lines=8) ticket_info = gr.Textbox(label="Ticket Creation Status", lines=2) image_button.click( fn=detect_image, inputs=[image_input], outputs=[image_output, image_info, ticket_info] ) with gr.Tab("Video Detection"): with gr.Row(): with gr.Column(): video_input = gr.Video(label="Upload Video") video_button = gr.Button("Process Video", variant="primary") with gr.Column(): video_output = gr.Video(label="Processed Video") video_info = gr.Textbox(label="Processing Information", lines=8) ticket_info = gr.Textbox(label="Ticket Creation Status", lines=2) video_button.click( fn=detect_video, inputs=[video_input], outputs=[video_output, video_info, ticket_info] ) gr.Markdown(""" ### Notes - Freshdesk tickets are created automatically when fire or accident is detected (one per incident type). - Images are saved in media/fire or media/accidents directories with unique filenames. - Ticket includes URL to the saved image. - For videos, one ticket is created per incident type with the first detected frame saved. - Deploy on Hugging Face Spaces with `requirements.txt` and model files (`fire.pt`, `best.pt`). - Debug info includes raw confidence scores and class labels to verify detection performance. """) # Launch the app if __name__ == "__main__": iface.launch()