# import cv2 # import numpy as np # import cv2 # import mediapipe as mp # import cv2 # import pandas as pd # import datetime # from django.core.mail import EmailMessage # from django.core.files.base import ContentFile # from django.core.management.base import BaseCommand # from multiprocessing import Process # from myprofile.models import * # from ultralytics import YOLO # from django.conf import settings # def process_device(camera_id,start_time_str, end_time_str, current_time_str): # camera = Camera.objects.get(id=camera_id) # width = 1000 # height = 700 # start_time = datetime.datetime.strptime(start_time_str, "%H:%M").time() # end_time = datetime.datetime.strptime(end_time_str, "%H:%M").time() # current_time_user = datetime.datetime.strptime(current_time_str, "%H:%M").time() # cap = cv2.VideoCapture(camera.rtsp_url) # # fourcc = cv2.VideoWriter_fourcc(*'mp4v') # Use appropriate codec based on system support # # output_video = cv2.VideoWriter('output_video.mp4', fourcc, 20.0, (width, height)) # fgbg = cv2.createBackgroundSubtractorKNN() # fixed_roi = [(50, 150), (50, 300), (200, 300), (200, 150)] # while True: # ret, frame = cap.read() # if not ret: # break # resized_frame = cv2.resize(frame, (width, height)) # fgmask = fgbg.apply(resized_frame) # kernel = np.ones((3, 3), np.uint8) # Adjust kernel size as needed # opening = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel, iterations=2) # closing = cv2.morphologyEx(opening, cv2.MORPH_CLOSE, kernel, iterations=2) # contours, _ = cv2.findContours(closing, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) # for cnt in contours: # area = cv2.contourArea(cnt) # if area > 500: # x, y, w, h = cv2.boundingRect(cnt) # if fixed_roi[0][0] <= x <= fixed_roi[2][0] and fixed_roi[0][1] <= y <= fixed_roi[2][1]: # if start_time <= current_time_user <= end_time: # cv2.rectangle(resized_frame, (x, y), (x + w, y + h), (0, 255, 0), 2) # cv2.putText(resized_frame, "Person Going Inside the Door (Green ROI)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) # else: # cv2.rectangle(resized_frame, (x, y), (x + w, y + h), (0, 0, 255), 2) # cv2.putText(resized_frame, "Intrusion Detected (Red ROI)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) # cv2.polylines(resized_frame, [np.array(fixed_roi)], isClosed=True, color=(0, 0, 255), thickness=2) # # output_video.write(resized_frame) # cv2.imshow("Intrusion Detection", resized_frame) # key = cv2.waitKey(1) & 0xFF # if key == ord('q'): # break # # output_video.release() # cap.release() # cv2.destroyAllWindows() # class Command(BaseCommand): # help = 'Triggers alerts based on video analysis' # def handle(self, *args, **options): # camera = Camera.objects.filter(is_enabled=True) # processes = [] # for cam in camera: # start_time_str = "08:00" # Example start time # end_time_str = "17:00" # Example end time # current_time_str = "12:00" # Example current time # process = Process(target=process_device, args=(cam.id, start_time_str, end_time_str, current_time_str)) # process.start() # processes.append(process) # for process in processes: # process.join() #========================================= Illustraion code with save in Database =========================================== import cv2 import numpy as np import cv2 import mediapipe as mp import cv2 import pandas as pd import datetime from django.core.mail import EmailMessage from django.core.files.base import ContentFile from django.core.management.base import BaseCommand from multiprocessing import Process from myprofile.models import * from ultralytics import YOLO from django.conf import settings def process_device(camera_id, start_time_str, end_time_str, current_time_str): camera = Camera.objects.get(id=camera_id) roi_data = Roi.objects.filter(camera=camera_id) width = 1000 height = 700 start_time = datetime.datetime.strptime(start_time_str, "%H:%M").time() end_time = datetime.datetime.strptime(end_time_str, "%H:%M").time() current_time_user = datetime.datetime.strptime(current_time_str, "%H:%M").time() cap = cv2.VideoCapture(camera.rtsp_url) fgbg = cv2.createBackgroundSubtractorKNN() while True: ret, frame = cap.read() if not ret: break resized_frame = cv2.resize(frame, (width, height)) fgmask = fgbg.apply(resized_frame) kernel = np.ones((3, 3), np.uint8) # Adjust kernel size as needed opening = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel, iterations=2) closing = cv2.morphologyEx(opening, cv2.MORPH_CLOSE, kernel, iterations=2) # Draw ROIs on the frame for roi in roi_data: roi_x1 = roi.x1 roi_y1 = roi.y1 roi_width = roi.width roi_height = roi.height cv2.rectangle(resized_frame, (roi_x1, roi_y1), (roi_x1 + roi_width, roi_y1 + roi_height), (0, 255, 0), 2) cv2.putText(resized_frame, roi.roi_name, (roi_x1, roi_y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 0), 2) contours, _ = cv2.findContours(closing, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) for cnt in contours: area = cv2.contourArea(cnt) if area > 500: x, y, w, h = cv2.boundingRect(cnt) # Check for intrusion in each ROI for roi in roi_data: roi_x1 = roi.x1 roi_y1 = roi.y1 roi_width = roi.width roi_height = roi.height if roi_x1 <= x <= roi_x1 + roi_width and roi_y1 <= y <= roi_y1 + roi_height: if start_time <= current_time_user <= end_time: cv2.rectangle(resized_frame, (x, y), (x + w, y + h), (0, 255, 0), 2) cv2.putText(resized_frame, "Person Going Inside the Door (Green ROI)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) else: cv2.rectangle(resized_frame, (x, y), (x + w, y + h), (0, 0, 255), 2) cv2.putText(resized_frame, "Intrusion Detected (Red ROI)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) # Save the frame in the database in Alert model alert = Alert(camera=camera, alert_message=f'Intrusion Detected in {camera.camera_name}') # Create a CameraEvent object for Intrusion Detection intrusion_event = CameraEvent.objects.get_or_create(camera_event="IntrusionDetection")[0] alert.camera_events = intrusion_event alert.save() # Save camera Incharge in alert alert.camera_incharge.set(camera.camera_user.all()) # Save alert image to the database _, encoded_frame = cv2.imencode('.jpg', resized_frame) image_file = ContentFile(encoded_frame.tobytes()) alert.frame.save('alert.jpg', image_file) # Send an email notification to the camera incharge # subject = f'Alert: Person detected in {camera.camera_name}' # message = f'A person has been detected in {camera.camera_name} at {datetime.datetime.now()}' # email = EmailMessage(subject, message, settings.EMAIL_HOST_USER, [camera.camera_user.email]) # email.attach_file(alert.frame.path) # email.send() cv2.imshow("Intrusion Detection", resized_frame) key = cv2.waitKey(1) & 0xFF if key == ord('q'): break cap.release() cv2.destroyAllWindows() class Command(BaseCommand): help = 'Triggers alerts based on video analysis' def handle(self, *args, **options): camera = Camera.objects.filter(is_enabled=True) processes = [] for cam in camera: start_time_str = "08:00" # Example start time end_time_str = "17:00" # Example end time current_time_str = "12:00" # Example current time process = Process(target=process_device, args=(cam.id, start_time_str, end_time_str, current_time_str)) process.start() processes.append(process) for process in processes: process.join()