Add security alarm system as ultralytics solution (#18281)

Co-authored-by: UltralyticsAssistant <web@ultralytics.com>
Co-authored-by: Glenn Jocher <glenn.jocher@ultralytics.com>
This commit is contained in:
Muhammad Rizwan Munawar 2024-12-17 23:42:37 +05:00 committed by GitHub
parent 5c2ecb7669
commit fd8159339c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 216 additions and 128 deletions

View file

@ -27,149 +27,54 @@ The Security Alarm System Project utilizing Ultralytics YOLO11 integrates advanc
### Code
#### Set up the parameters of the message
???+ note
App Password Generation is necessary
- Navigate to [App Password Generator](https://myaccount.google.com/apppasswords), designate an app name such as "security project," and obtain a 16-digit password. Copy this password and paste it into the designated password field as instructed.
- Navigate to [App Password Generator](https://myaccount.google.com/apppasswords), designate an app name such as "security project," and obtain a 16-digit password. Copy this password and paste it into the designated `password` field in the code below.
```python
password = ""
from_email = "" # must match the email used to generate the password
to_email = "" # receiver email
```
!!! example "Security Alarm System using YOLO11 Example"
#### Server creation and authentication
=== "Python"
```python
import smtplib
```python
import cv2
server = smtplib.SMTP("smtp.gmail.com: 587")
server.starttls()
server.login(from_email, password)
```
from ultralytics import solutions
#### Email Send Function
cap = cv2.VideoCapture("Path/to/video/file.mp4")
assert cap.isOpened(), "Error reading video file"
```python
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
# Video writer
w, h, fps = (int(cap.get(x)) for x in (cv2.CAP_PROP_FRAME_WIDTH, cv2.CAP_PROP_FRAME_HEIGHT, cv2.CAP_PROP_FPS))
video_writer = cv2.VideoWriter("security_alarm_output.avi", cv2.VideoWriter_fourcc(*"mp4v"), fps, (w, h))
from_email = "abc@gmail.com" # The sender email address
password = "---- ---- ---- ----" # 16-digits password generated via: https://myaccount.google.com/apppasswords
to_email = "xyz@gmail.com" # The receiver email address
def send_email(to_email, from_email, object_detected=1):
"""Sends an email notification indicating the number of objects detected; defaults to 1 object."""
message = MIMEMultipart()
message["From"] = from_email
message["To"] = to_email
message["Subject"] = "Security Alert"
# Add in the message body
message_body = f"ALERT - {object_detected} objects has been detected!!"
message.attach(MIMEText(message_body, "plain"))
server.sendmail(from_email, to_email, message.as_string())
```
#### Object Detection and Alert Sender
```python
from time import time
import cv2
import torch
from ultralytics import YOLO
from ultralytics.utils.plotting import Annotator, colors
class ObjectDetection:
def __init__(self, capture_index):
"""Initializes an ObjectDetection instance with a given camera index."""
self.capture_index = capture_index
self.email_sent = False
# model information
self.model = YOLO("yolo11n.pt")
# visual information
self.annotator = None
self.start_time = 0
self.end_time = 0
# device information
self.device = "cuda" if torch.cuda.is_available() else "cpu"
def predict(self, im0):
"""Run prediction using a YOLO model for the input image `im0`."""
results = self.model(im0)
return results
def display_fps(self, im0):
"""Displays the FPS on an image `im0` by calculating and overlaying as white text on a black rectangle."""
self.end_time = time()
fps = 1 / round(self.end_time - self.start_time, 2)
text = f"FPS: {int(fps)}"
text_size = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, 1.0, 2)[0]
gap = 10
cv2.rectangle(
im0,
(20 - gap, 70 - text_size[1] - gap),
(20 + text_size[0] + gap, 70 + gap),
(255, 255, 255),
-1,
# Init SecurityAlarm
security = solutions.SecurityAlarm(
show=True, # Display the output
model="yolo11n.pt", # i.e. YOLO11s.pt
records=1, # Total detections count to send an email about security
)
cv2.putText(im0, text, (20, 70), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 0, 0), 2)
def plot_bboxes(self, results, im0):
"""Plots bounding boxes on an image given detection results; returns annotated image and class IDs."""
class_ids = []
self.annotator = Annotator(im0, 3, results[0].names)
boxes = results[0].boxes.xyxy.cpu()
clss = results[0].boxes.cls.cpu().tolist()
names = results[0].names
for box, cls in zip(boxes, clss):
class_ids.append(cls)
self.annotator.box_label(box, label=names[int(cls)], color=colors(int(cls), True))
return im0, class_ids
security.authenticate(from_email, password, to_email) # Authenticate the email server
def __call__(self):
"""Run object detection on video frames from a camera stream, plotting and showing the results."""
cap = cv2.VideoCapture(self.capture_index)
assert cap.isOpened()
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
frame_count = 0
while True:
self.start_time = time()
ret, im0 = cap.read()
assert ret
results = self.predict(im0)
im0, class_ids = self.plot_bboxes(results, im0)
if len(class_ids) > 0: # Only send email If not sent before
if not self.email_sent:
send_email(to_email, from_email, len(class_ids))
self.email_sent = True
else:
self.email_sent = False
self.display_fps(im0)
cv2.imshow("YOLO11 Detection", im0)
frame_count += 1
if cv2.waitKey(5) & 0xFF == 27:
# Process video
while cap.isOpened():
success, im0 = cap.read()
if not success:
print("Video frame is empty or video processing has been successfully completed.")
break
im0 = security.monitor(im0)
video_writer.write(im0)
cap.release()
video_writer.release()
cv2.destroyAllWindows()
server.quit()
```
#### Call the Object Detection class and Run the Inference
```python
detector = ObjectDetection(capture_index=0)
detector()
```
```
That's it! When you execute the code, you'll receive a single notification on your email if any object is detected. The notification is sent immediately, not repeatedly. However, feel free to customize the code to suit your project requirements.
@ -177,6 +82,21 @@ That's it! When you execute the code, you'll receive a single notification on yo
<img width="256" src="https://github.com/ultralytics/docs/releases/download/0/email-received-sample.avif" alt="Email Received Sample">
### Arguments `SecurityAlarm`
Here's a table with the `SecurityAlarm` arguments:
| Name | Type | Default | Description |
| ------------ | ------ | ------- | ------------------------------------------------------- |
| `model` | `str` | `None` | Path to Ultralytics YOLO Model File |
| `line_width` | `int` | `2` | Line thickness for bounding boxes. |
| `show` | `bool` | `False` | Flag to control whether to display the video stream. |
| `records` | `int` | `5` | Total detections count to send an email about security. |
### Arguments `model.track`
{% include "macros/track-args.md" %}
## FAQ
### How does Ultralytics YOLO11 improve the accuracy of a security alarm system?

View file

@ -9,3 +9,4 @@
| `kpts` | `list` | `[6, 8, 10]` | List of keypoints used for monitoring workouts. These keypoints correspond to body joints or parts, such as shoulders, elbows, and wrists, for exercises like push-ups, pull-ups, squats, ab-workouts. |
| `analytics_type` | `str` | `line` | Specifies the type of analytics visualization to generate. Options include `"line"`, `"pie"`, `"bar"`, or `"area"`. The default is `"line"` for trend visualization. |
| `json_file` | `str` | `None` | Path to the JSON file defining regions for parking systems or similar applications. Enables flexible configuration of analysis areas. |
| `records` | `int` | `5` | Total detections count that triggers an automated email notification about unusual activity. |

View file

@ -0,0 +1,16 @@
---
description: Discover how Ultralytics' Security Alarm System enhances real-time surveillance with intelligent object detection and tracking. Learn about setup, monitoring, and threat detection.
keywords: Ultralytics, Security Alarm System, Real-time Surveillance, Object Detection, Video Monitoring, Python, Threat Detection
---
# Reference for `ultralytics/solutions/security_alarm.py`
!!! note
This file is available at [https://github.com/ultralytics/ultralytics/blob/main/ultralytics/solutions/security_alarm.py](https://github.com/ultralytics/ultralytics/blob/main/ultralytics/solutions/security_alarm.py). If you spot a problem please help fix it by [contributing](https://docs.ultralytics.com/help/contributing/) a [Pull Request](https://github.com/ultralytics/ultralytics/edit/main/ultralytics/solutions/security_alarm.py) 🛠️. Thank you 🙏!
<br>
## ::: ultralytics.solutions.security_alarm.SecurityAlarm
<br><br>

View file

@ -49,6 +49,10 @@ Here's our curated list of Ultralytics solutions that can be used to create awes
yolo solutions source="path/to/video/file.mp4" # specify video file path
```
## Arguments
{% include "macros/solutions-args.md" %}
## Contribute to Our Solutions
We welcome contributions from the community! If you've mastered a particular aspect of Ultralytics YOLO that's not yet covered in our solutions, we encourage you to share your expertise. Writing a guide is a great way to give back to the community and help us make our documentation more comprehensive and user-friendly.

View file

@ -572,6 +572,7 @@ nav:
- parking_management: reference/solutions/parking_management.md
- queue_management: reference/solutions/queue_management.md
- region_counter: reference/solutions/region_counter.md
- security_alarm: reference/solutions/security_alarm.md
- solutions: reference/solutions/solutions.md
- speed_estimation: reference/solutions/speed_estimation.md
- streamlit_inference: reference/solutions/streamlit_inference.md

View file

@ -17,3 +17,6 @@ kpts: [6, 8, 10] # Keypoints for workouts monitoring, i.e. If you want to consid
# Analytics settings
analytics_type: "line" # Analytics type i.e "line", "pie", "bar" or "area" charts. By default, "line" analytics will be used for processing.
json_file: # parking system regions file path.
# Security alarm system
records: 5 # Total detections count to send an email about security

View file

@ -8,6 +8,7 @@ from .object_counter import ObjectCounter
from .parking_management import ParkingManagement, ParkingPtsSelection
from .queue_management import QueueManager
from .region_counter import RegionCounter
from .security_alarm import SecurityAlarm
from .speed_estimation import SpeedEstimator
from .streamlit_inference import inference
from .trackzone import TrackZone
@ -25,4 +26,5 @@ __all__ = (
"inference",
"RegionCounter",
"TrackZone",
"SecurityAlarm",
)

View file

@ -0,0 +1,141 @@
# Ultralytics YOLO 🚀, AGPL-3.0 license
from ultralytics.solutions.solutions import LOGGER, BaseSolution
from ultralytics.utils.plotting import Annotator, colors
class SecurityAlarm(BaseSolution):
"""
A class to manage security alarm functionalities for real-time monitoring.
This class extends the BaseSolution class and provides features to monitor
objects in a frame, send email notifications when specific thresholds are
exceeded for total detections, and annotate the output frame for visualization.
Attributes:
email_sent (bool): Flag to track if an email has already been sent for the current event.
records (int): Threshold for the number of detected objects to trigger an alert.
Methods:
authenticate: Sets up email server authentication for sending alerts.
send_email: Sends an email notification with details and an image attachment.
monitor: Monitors the frame, processes detections, and triggers alerts if thresholds are crossed.
Examples:
>>> security = SecurityAlarm()
>>> security.authenticate("abc@gmail.com", "1111222233334444", "xyz@gmail.com")
>>> frame = cv2.imread("frame.jpg")
>>> processed_frame = security.monitor(frame)
"""
def __init__(self, **kwargs):
"""Initializes the SecurityAlarm class with parameters for real-time object monitoring."""
super().__init__(**kwargs)
self.email_sent = False
self.records = self.CFG["records"]
def authenticate(self, from_email, password, to_email):
"""
Authenticates the email server for sending alert notifications.
Args:
from_email (str): Sender's email address.
password (str): Password for the sender's email account.
to_email (str): Recipient's email address.
This method initializes a secure connection with the SMTP server
and logs in using the provided credentials.
Examples:
>>> alarm = SecurityAlarm()
>>> alarm.authenticate("sender@example.com", "password123", "recipient@example.com")
"""
import smtplib
self.server = smtplib.SMTP("smtp.gmail.com: 587")
self.server.starttls()
self.server.login(from_email, password)
self.to_email = to_email
self.from_email = from_email
def send_email(self, im0, records=5):
"""
Sends an email notification with an image attachment indicating the number of objects detected.
Args:
im0 (numpy.ndarray): The input image or frame to be attached to the email.
records (int): The number of detected objects to be included in the email message.
This method encodes the input image, composes the email message with
details about the detection, and sends it to the specified recipient.
Examples:
>>> alarm = SecurityAlarm()
>>> frame = cv2.imread("path/to/image.jpg")
>>> alarm.send_email(frame, records=10)
"""
from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import cv2
img_bytes = cv2.imencode(".jpg", im0)[1].tobytes() # Encode the image as JPEG
# Create the email
message = MIMEMultipart()
message["From"] = self.from_email
message["To"] = self.to_email
message["Subject"] = "Security Alert"
# Add the text message body
message_body = f"Ultralytics ALERT!!! " f"{records} objects have been detected!!"
message.attach(MIMEText(message_body, "plain"))
# Attach the image
image_attachment = MIMEImage(img_bytes, name="ultralytics.jpg")
message.attach(image_attachment)
# Send the email
try:
self.server.send_message(message)
LOGGER.info("✅ Email sent successfully!")
except Exception as e:
print(f"❌ Failed to send email: {e}")
def monitor(self, im0):
"""
Monitors the frame, processes object detections, and triggers alerts if thresholds are exceeded.
Args:
im0 (numpy.ndarray): The input image or frame to be processed and annotated.
This method processes the input frame, extracts detections, annotates the frame
with bounding boxes, and sends an email notification if the number of detected objects
surpasses the specified threshold and an alert has not already been sent.
Returns:
(numpy.ndarray): The processed frame with annotations.
Examples:
>>> alarm = SecurityAlarm()
>>> frame = cv2.imread("path/to/image.jpg")
>>> processed_frame = alarm.monitor(frame)
"""
self.annotator = Annotator(im0, line_width=self.line_width) # Initialize annotator
self.extract_tracks(im0) # Extract tracks
# Iterate over bounding boxes, track ids and classes index
for box, cls in zip(self.boxes, self.clss):
# Draw bounding box
self.annotator.box_label(box, label=self.names[cls], color=colors(cls, True))
total_det = len(self.clss)
if total_det > self.records: # Only send email If not sent before
if not self.email_sent:
self.send_email(im0, total_det)
self.email_sent = True
self.display_output(im0) # display output with base class function
return im0 # return output image for more usage