Detailed Explanation of the Provided Code

The provided code is for a Django project that involves real-time face detection
and recognition using WebSockets and OpenCV. Below is a detailed line-by-line
explanation of the code along with Postman test cases for API endpoints.

WebSocket Consumer: camera/consumers.py

python
Copy code
import base64
import cv2
import json
import numpy as np
import threading
from channels.generic.websocket import WebsocketConsumer
from .face_recognition_module import process_frame
from django.contrib.auth import get_user_model
from notifications.signals import notify

User = get_user_model()
Imports necessary libraries and modules for image processing, threading, WebSocket
handling, and notifications.
python
Copy code
class CameraConsumer(WebsocketConsumer):
def connect(self):
self.accept()
self.camera_url = self.scope['url_route']['kwargs']['camera_url']
self.capture_thread = threading.Thread(target=self.stream_camera)
self.capture_thread.start()
Defines CameraConsumer which extends WebsocketConsumer.
connect method is called when a WebSocket connection is made. It accepts the
connection, retrieves the camera URL from the scope, and starts a new thread to
handle the camera streaming.
python
Copy code
def disconnect(self, close_code):
if self.capture_thread.is_alive():
self.capture_thread.join()
disconnect method is called when the WebSocket connection is closed. It ensures the
capture thread is properly joined.
python
Copy code
def stream_camera(self):
cap = cv2.VideoCapture(self.camera_url)
if not cap.isOpened():
self.send(text_data=json.dumps({"error": "Failed to open camera
stream"}))
return

while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
stream_camera method captures frames from the camera URL using OpenCV.
If the camera fails to open, an error message is sent via WebSocket.
Continuously reads frames from the camera stream.
python
Copy code
try:
 frame, face_names, detection_time = process_frame(frame)
_, buffer = cv2.imencode('.jpg', frame)
frame_base64 = base64.b64encode(buffer).decode('utf-8')
response = {
"frame": frame_base64,
"face_names": face_names,
"detection_time": detection_time
}
self.send(text_data=json.dumps(response))

for name in face_names:

if name.startswith("Unknown"):
notification_message = f"{name} detected in
{self.camera_url} at {detection_time}, view in database"
else:
notification_message = f"{name} detected in
{self.camera_url} at {detection_time}, view in database"

notify.send(self.scope["user"], recipient=self.scope["user"],
verb="Face Detected", description=notification_message)

except Exception as e:
self.send(text_data=json.dumps({"error": str(e)}))
break

cap.release()
Calls process_frame to detect and recognize faces in the frame.
Encodes the frame as a JPEG and converts it to a base64 string.
Sends the processed frame, detected face names, and detection time via WebSocket.
Sends notifications to the user if faces are detected.
Handles exceptions and releases the camera resource if an error occurs.
Face Recognition Module: camera/face_recognition_module.py
python
Copy code
import numpy as np
import cv2
from mtcnn.mtcnn import MTCNN
import face_recognition
from .models import Face
import time

detector = MTCNN()
Imports necessary libraries and modules for face detection and recognition.
Initializes the MTCNN detector.
python
Copy code
def detect_faces(frame):
return detector.detect_faces(frame)
Defines a function to detect faces using MTCNN.
python
Copy code
def recognize_faces(frame, known_face_encodings, known_face_names):
rgb_frame = frame[:, :, ::-1]
face_locations = face_recognition.face_locations(rgb_frame)
face_encodings = face_recognition.face_encodings(rgb_frame, face_locations)

face_names = []
for face_encoding in face_encodings:
matches = face_recognition.compare_faces(known_face_encodings,
face_encoding)
name = "Unknown"
if True in matches:
first_match_index = matches.index(True)
name = known_face_names[first_match_index]
face_names.append(name)

return face_locations, face_names, face_encodings

Converts the frame from BGR to RGB format.
Detects face locations and encodings using the face_recognition library.
Compares the detected face encodings with known face encodings to recognize faces.
python
Copy code
def process_frame(frame):
start_time = time.time()

known_faces = Face.objects.all()
known_face_encodings = [np.frombuffer(face.embedding, dtype=np.float64) for
face in known_faces]
known_face_names = [face.name for face in known_faces]

face_locations, face_names, face_encodings = recognize_faces(frame,

known_face_encodings, known_face_names)

for (top, right, bottom, left), name, face_encoding in zip(face_locations,

face_names, face_encodings):
if name == "Unknown":
new_face = Face(user_id=1, embedding=face_encoding.tobytes())
new_face.save()
name = f"Unknown {new_face.id}"

cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 255), 2)

cv2.rectangle(frame, (left, bottom - 35), (right, bottom), (0, 0, 255),
cv2.FILLED)
font = cv2.FONT_HERSHEY_DUPLEX
cv2.putText(frame, name, (left + 6, bottom - 6), font, 1.0, (255, 255,
255), 1)

end_time = time.time()
detection_time = end_time - start_time

return frame, face_names, detection_time

Fetches known faces from the database.
Detects and recognizes faces in the frame.
Draws rectangles and labels around detected faces.
Returns the processed frame, detected face names, and detection time.
Models: camera/models.py
python
Copy code
from django.db import models
from django.conf import settings
import urllib.parse
from django.utils import timezone

class Face(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
name = models.CharField(max_length=255, default="Unknown")
embedding = models.BinaryField()
created_at = models.DateTimeField(auto_now_add=True)
 updated_at = models.DateTimeField(auto_now=True)

def save(self, *args, **kwargs):

if self.name.startswith("Unknown"):
max_unknown = Face.objects.filter(name__startswith="Unknown").count()
self.name = f"Unknown {max_unknown + 1}"
super().save(*args, **kwargs)
Defines Face model to store face data, including user, name, embedding, and
timestamps.
Overrides the save method to handle naming for unknown faces.
python
Copy code
class StaticCamera(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
ip_address = models.CharField(max_length=255)
username = models.CharField(max_length=255)
password = models.CharField(max_length=255)
name = models.CharField(max_length=255, default="Static Camera")

def rtsp_url(self):
username = urllib.parse.quote(self.username)
password = urllib.parse.quote(self.password)
return
f"rtsp://{username}:{password}@{self.ip_address}:554/Streaming/Channels/101"
Defines StaticCamera model to store static camera details and generate RTSP URLs.
python
Copy code
class DDNSCamera(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
ddns_hostname = models.CharField(max_length=255)
username = models.CharField(max_length=255)
password = models.CharField(max_length=255)
name = models.CharField(max_length=255, default="DDNS Camera")

def rtsp_url(self):
username = urllib.parse.quote(self.username)
password = urllib.parse.quote(self.password)
return f"rtsp://{username}:{password}@{self.ddns_hostname}:554/Streaming/
Channels/101"
Defines DDNSCamera model to store DDNS camera details and generate RTSP URLs.
python
Copy code
class CameraStream(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
camera = models.ForeignKey(StaticCamera, null=True, blank=True,
on_delete=models.CASCADE)
ddns_camera = models.ForeignKey(DDNSCamera, null=True, blank=True,
on_delete=models.CASCADE)
stream_url = models.CharField(max_length=255)
created_at = models.DateTimeField(default=timezone.now)
Defines CameraStream model to store camera stream details, linking to either
StaticCamera or DDNSCamera.
WebSocket URL Patterns: camera/routing.py
python
Copy code
from django.urls import re_path
from .consumers import CameraConsumer

websocket_urlpatterns = [
 re_path(r'ws/camera/(?P<camera_url>[^/]+)/$', CameraConsumer.as_asgi()),
]
Defines URL patterns for WebSocket connections, mapping to CameraConsumer.
Serializers: camera/serializers.py
python
Copy code
continue

2/2

ChatGPT
Serializers: camera/serializers.py
python
Copy code
from rest_framework import serializers
from .models import StaticCamera, DDNSCamera, CameraStream, Face

class StaticCameraSerializer(serializers.ModelSerializer):
class Meta:
model = StaticCamera
fields = ['ip_address', 'username', 'password']
Defines StaticCameraSerializer to serialize and deserialize StaticCamera objects.
python
Copy code
class DDNSCameraSerializer(serializers.ModelSerializer):
class Meta:
model = DDNSCamera
fields = ['ddns_hostname', 'username', 'password']
Defines DDNSCameraSerializer to serialize and deserialize DDNSCamera objects.
python
Copy code
class CameraStreamSerializer(serializers.ModelSerializer):
class Meta:
model = CameraStream
fields = ['stream_url']
Defines CameraStreamSerializer to serialize and deserialize CameraStream objects.
python
Copy code
class FaceSerializer(serializers.ModelSerializer):
class Meta:
model = Face
fields = ['id', 'name', 'embedding', 'created_at']
Defines FaceSerializer to serialize and deserialize Face objects.
python
Copy code
class RenameFaceSerializer(serializers.ModelSerializer):
class Meta:
model = Face
fields = ['name']
Defines RenameFaceSerializer to serialize and deserialize only the name field of
Face objects.
python
Copy code
class RenameCameraSerializer(serializers.Serializer):
name = serializers.CharField(max_length=255)
Defines RenameCameraSerializer to handle renaming cameras.
URL Patterns: camera/urls.py
python
Copy code
from django.urls import path
from .views import StaticCameraView, DDNSCameraView, GetStreamURLView, FaceView,
RenameFaceView, DetectedFacesView, RenameCameraView, NotificationView

urlpatterns = [
path('static-camera/', StaticCameraView.as_view(), name='static_camera'),
path('ddns-camera/', DDNSCameraView.as_view(), name='ddns_camera'),
path('get-stream-url/<str:camera_type>/', GetStreamURLView.as_view(),
name='get_stream_url'),
path('faces/', FaceView.as_view(), name='faces'),
path('rename-face/<int:pk>/', RenameFaceView.as_view(), name='rename_face'),
path('detected-faces/', DetectedFacesView.as_view(), name='detected_faces'),
path('rename-camera/<str:camera_type>/<int:pk>/', RenameCameraView.as_view(),
name='rename_camera'),
path('notifications/', NotificationView.as_view(), name='notifications'),
]
Defines URL patterns for various API endpoints, mapping to their respective views.
Views: camera/views.py
python
Copy code
from django.shortcuts import render
from notifications.models import Notification
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from .models import StaticCamera, DDNSCamera, CameraStream, Face
from .serializers import StaticCameraSerializer, DDNSCameraSerializer,
CameraStreamSerializer, FaceSerializer, RenameFaceSerializer,
RenameCameraSerializer
from rest_framework import status, generics
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from .pagination import DynamicPageSizePagination

class NotificationView(APIView):
permission_classes = [IsAuthenticated]

def get(self, request):

notifications = Notification.objects.filter(recipient=request.user)
notification_data = [{"id": n.id, "verb": n.verb, "description":
n.description, "timestamp": n.timestamp} for n in notifications]
return Response(notification_data, status=200)
Defines NotificationView to handle fetching notifications for the authenticated
user.
python
Copy code
class StaticCameraView(generics.GenericAPIView):
serializer_class = StaticCameraSerializer
permission_classes = [IsAuthenticated]

@swagger_auto_schema(request_body=StaticCameraSerializer)
def post(self, request):
serializer = self.serializer_class(data=request.data)
if serializer.is_valid():
static_camera = StaticCamera.objects.create(user=request.user,
**serializer.validated_data)
CameraStream.objects.create(user=request.user, camera=static_camera,
stream_url=static_camera.rtsp_url())
return Response({"message": "Static camera details saved
successfully"}, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
Defines StaticCameraView to handle creating new static camera entries.
python
Copy code
class DDNSCameraView(generics.GenericAPIView):
serializer_class = DDNSCameraSerializer
permission_classes = [IsAuthenticated]

@swagger_auto_schema(request_body=DDNSCameraSerializer)
def post(self, request):
serializer = self.serializer_class(data=request.data)
if serializer.is_valid():
ddns_camera = DDNSCamera.objects.create(user=request.user,
**serializer.validated_data)
CameraStream.objects.create(user=request.user, ddns_camera=ddns_camera,
stream_url=ddns_camera.rtsp_url())
return Response({"message": "DDNS camera details saved successfully"},
status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
Defines DDNSCameraView to handle creating new DDNS camera entries.
python
Copy code
class GetStreamURLView(APIView):
permission_classes = [IsAuthenticated]

def get(self, request, camera_type):

try:
if camera_type == 'static':
cameras = StaticCamera.objects.filter(user=request.user)
else:
cameras = DDNSCamera.objects.filter(user=request.user)

stream_urls = []
for camera in cameras:
streams = CameraStream.objects.filter(user=request.user,
camera=camera if camera_type == 'static' else None, ddns_camera=camera if
camera_type != 'static' else None)
for stream in streams:
stream_urls.append(stream.stream_url)

return Response({"stream_urls": stream_urls},

status=status.HTTP_200_OK)
except (StaticCamera.DoesNotExist, DDNSCamera.DoesNotExist):
return Response({"error": "Camera not found"},
status=status.HTTP_404_NOT_FOUND)
Defines GetStreamURLView to fetch stream URLs for the user's cameras.
python
Copy code
class FaceView(generics.GenericAPIView):
serializer_class = FaceSerializer
permission_classes = [IsAuthenticated]

def post(self, request):

serializer = self.serializer_class(data=request.data)
if serializer.is_valid():
Face.objects.create(user=request.user, **serializer.validated_data)
return Response({"message": "Face details saved successfully"},
status=status.HTTP_201_CREATED)
 return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
Defines FaceView to handle creating new face entries.
python
Copy code
class RenameFaceView(generics.GenericAPIView):
serializer_class = RenameFaceSerializer
permission_classes = [IsAuthenticated]

def get_queryset(self):
return Face.objects.all()

def patch(self, request, pk):

try:
face = Face.objects.get(pk=pk)
except Face.DoesNotExist:
return Response({"error": "Face not found"},
status=status.HTTP_404_NOT_FOUND)

serializer = self.get_serializer(face, data=request.data, partial=True)

if serializer.is_valid():
serializer.save()
return Response({"message": "Face renamed successfully"},
status=status.HTTP_200_OK)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
Defines RenameFaceView to handle renaming faces.
python
Copy code
class DetectedFacesView(generics.ListAPIView):
serializer_class = FaceSerializer
permission_classes = [IsAuthenticated]
pagination_class = DynamicPageSizePagination

def get_queryset(self):
queryset = Face.objects.filter(user=self.request.user).order_by('-
created_at')
date = self.request.query_params.get('date', None)
if date:
queryset = queryset.filter(created_at__date=date)
return queryset
Defines DetectedFacesView to list detected faces for the authenticated user with
optional date filtering.
python
Copy code
class RenameCameraView(APIView):

@swagger_auto_schema(
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
required=['name'],
properties={
'name': openapi.Schema(type=openapi.TYPE_STRING, description='New
name of the camera'),
},
),
responses={200: 'OK', 400: 'Bad Request', 404: 'Not Found'},
)
def patch(self, request, camera_type, pk):
name = request.data.get('name')
if not name:
 return Response({"name": ["This field is required."]},
status=status.HTTP_400_BAD_REQUEST)

if camera_type == 'static':
try:
camera = StaticCamera.objects.get(pk=pk)
except StaticCamera.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
elif camera_type == 'ddns':
try:
camera = DDNSCamera.objects.get(pk=pk)
except DDNSCamera.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)

camera.name = name
camera.save()

return Response(status=status.HTTP_200_OK)
Defines RenameCameraView to handle renaming static or DDNS cameras.
Documentation for Provided Code
WebSocket Consumer: CameraConsumer
This consumer handles the WebSocket connection for streaming camera frames,
processing them for face detection and recognition, and sending the processed
frames back to the client. It also sends notifications for detected faces.

Face Recognition Module: face_recognition_module

This module includes functions to detect and recognize faces in a frame using MTCNN
and the face_recognition library. It processes each frame, draws bounding boxes
around detected faces, and updates the database with unknown faces.

Models
Face: Stores face data including embeddings, names, and timestamps.
StaticCamera: Stores details for static cameras and generates RTSP URLs.
DDNSCamera: Stores details for DDNS cameras and generates RTSP URLs.
CameraStream: Links camera streams to users and cameras.
URL Patterns and Views
URL patterns map to views handling CRUD operations for cameras and faces, fetching
stream URLs, and handling notifications.
Views use Django REST framework to implement API endpoints for the above
operations.
Postman Test Cases
1. Add Static Camera
URL: /static-camera/
Method: POST
Request Body:

json
Copy code
{
"ip_address": "192.168.1.100",
"username": "admin",
"password": "password"
}
Expected Response:

json
Copy code
{
"message": "Static camera details saved successfully"
}
2. Add DDNS Camera
URL: /ddns-camera/
Method: POST
Request Body:

json
Copy code
{
"ddns_hostname": "example.ddns.net",
"username": "admin",
"password": "password"
}
Expected Response:

json
Copy code
{
"message": "DDNS camera details saved successfully"
}
3. Get Stream URLs
URL: /get-stream-url/static/
Method: GET
Expected Response:

json
Copy code
{
"stream_urls": ["rtsp://admin:[email protected]:554/Streaming/Channels/101"]
}
4. Save Face Details
URL: /faces/
Method: POST
Request Body:

json
Copy code
{
"name": "John Doe",
"embedding": "base64encodedstring"
}
Expected Response:

json
Copy code
{
"message": "Face details saved successfully"
}
5. Rename Face
URL: /rename-face/1/
Method: PATCH
Request Body:

json
Copy code
{
"name": "Jane Doe"
}
Expected Response:

json
Copy code
{
"message": "Face renamed successfully"
}
6. List Detected Faces
URL: /detected-faces/
Method: GET
Expected Response:

json
Copy code
[
{
"id": 1,
"name": "Unknown 1",
"embedding": "base64encodedstring",
"created_at": "2024-07-01T00:00:00Z"
},
{
"id": 2,
"name": "John Doe",
"embedding": "base64encodedstring",
"created_at": "2024-07-01T00:00:00Z"
}
]
7. Rename Camera
URL: /rename-camera/static/1/
Method: PATCH
Request Body:

json
Copy code
{
"name": "New Camera Name"
}
Expected Response:

json
Copy code
{
"message": "Camera renamed successfully"
}
8. Get Notifications
URL: /notifications/
Method: GET
Expected Response:

json
Copy code
[
{
"id": 1,
"verb": "Face Detected",
"description": "John Doe detected in camera_url at detection_time, view in
database",
 "timestamp": "2024-07-01T00:00:00Z"
}
]
These test cases cover the basic functionalities of the provided code. Adjust the
request bodies and expected responses based on the actual data in your database and
test environment.