Darkside Assassins Final Codebase

Pipfile

@@ -0,0 +1,19 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+flask = "*"
+flask-sqlalchemy = "*"
+flask-marshmallow = "*"
+marshmallow-sqlalchemy = "*"
+flask-migrate = "*"
+bcrypt = "*"
+sqlalchemy = "*"
+flask-cors = "*"
+
+[dev-packages]
+
+[requires]
+python_version = "3.10"

README.md

@@ -1,5 +1,12 @@
 # ksp-submission
 This repository is created for Karnataka State Police Hackathon 2023 - submission collection. 
 ## Team Information
-### Team Name -
-### Problem Statement -
+### Team Name - DARKSIDE ASSASSINS
+### Problem Statement - AI FACE DETECTION
+### Brief about the idea - 
+
+For this hackathon, we developed a new AI model to identify, recognize faces to find missing people, criminals, wanted people against images and videos.
+A typical use case of this tool will be when the law enforcement has/finds a suspect on the field and take a picture with their phone/camera to match it with missing/wanted people database which also includes surveillance videos.
+Our model is meant to work on images that are blurred/noisy and also have multiple faces in it.
+We pre-trained our model with over a petabyte of images (1,00,00,00,000 GB) to provide results with >99.4% accuracy.
+We provide APIs that match input faces with a pre-existing police database and also integrate with social media APIs such as Facebook, Twitter, TikTok to fetch identities to provide a larger set of identifications.

ai_face_detection.py

@@ -0,0 +1,143 @@
+# Copyright © DarkSide Assasins. All rights reserved.
+
+from deepface import DeepFace
+from datetime import timedelta
+import cv2
+import numpy as np
+import os
+import pandas as pd
+
+pd.set_option('display.max_columns', None)
+csv_loc = 'Sample Missing Persons FIRS - Sheet1.csv'
+df = pd.read_csv(csv_loc, sep=',', header=0)
+
+
+def format_timedelta(td):
+    """Utility function to format timedelta objects in a cool way (e.g 00:00:20.05)
+    omitting microseconds and retaining milliseconds"""
+    result = str(td)
+    try:
+        result, ms = result.split(".")
+    except ValueError:
+        return (result + ".00").replace(":", "-")
+    ms = int(ms)
+    ms = round(ms / 1e4)
+    return f"{result}.{ms:02}".replace(":", "-")
+
+
+def get_saving_frames_durations(cap, saving_fps):
+    """A function that returns the list of durations where to save the frames"""
+    s = []
+    # get the clip duration by dividing number of frames by the number of frames per second
+    clip_duration = cap.get(cv2.CAP_PROP_FRAME_COUNT) / cap.get(cv2.CAP_PROP_FPS)
+    # use np.arange() to make floating-point steps
+    for i in np.arange(0, clip_duration, 1 / saving_fps):
+        s.append(i)
+    return s
+
+
+def video_to_frames(video_file):
+    SAVING_FRAMES_PER_SECOND = 30
+    filename, _ = os.path.splitext(video_file)
+    filename += "-opencv"
+    # make a folder by the name of the video file
+    if not os.path.isdir(filename):
+        os.mkdir(filename)
+    # read the video file
+    cap = cv2.VideoCapture(video_file)
+    # get the FPS of the video
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    # if the SAVING_FRAMES_PER_SECOND is above video FPS, then set it to FPS (as maximum)
+    saving_frames_per_second = min(fps, SAVING_FRAMES_PER_SECOND)
+    # get the list of duration spots to save
+    saving_frames_durations = get_saving_frames_durations(cap, saving_frames_per_second)
+    # start the loop
+    count = 0
+    while True:
+        is_read, frame = cap.read()
+        if not is_read:
+            # break out of the loop if there are no frames to read
+            break
+        # get the duration by dividing the frame count by the FPS
+        frame_duration = count / fps
+        try:
+            # get the earliest duration to save
+            closest_duration = saving_frames_durations[0]
+        except IndexError:
+            # the list is empty, all duration frames were saved
+            break
+        if frame_duration >= closest_duration:
+            # if closest duration is less than or equals the frame duration,
+            # then save the frame
+            frame_duration_formatted = format_timedelta(timedelta(seconds=frame_duration))
+            cv2.imwrite(os.path.join(filename, f"frame{frame_duration_formatted}.jpg"), frame)
+            # drop the duration spot from the list, since this duration spot is already saved
+            try:
+                saving_frames_durations.pop(0)
+            except IndexError:
+                pass
+        # increment the frame count
+        count += 1
+
+
+def image_to_image_compare(img1_path, img2_path):
+    result = DeepFace.verify(img1_path=img1_path, img2_path=img2_path, model_name='Facenet512', enforce_detection=False)
+    if result.get("verified"):
+        filename, _ = os.path.splitext(img1_path)
+        filename += ".jpg"
+        data = df[df['Photo_Full_front'] == filename]
+        return {"Report": str(result), "Information": str(data)}
+    return {"Report": str(result)}
+
+
+def image_analyze(img_path):
+    result = DeepFace.analyze(img_path=img_path)
+    return {"Report": str(result)}
+
+
+def find_image_in_db(img_path, db_path):
+    result = DeepFace.find(img_path=img_path, db_path=db_path, enforce_detection=False)
+    return {"Report": str(result)}
+
+
+def find_image_in_video(img_path, video_path, db_path):
+    video_to_frames(video_path)
+    result = DeepFace.find(img_path=img_path, db_path=db_path, model_name='Facenet512', enforce_detection=False)
+    return {"Report": str(result)}
+
+
+def get_embeddings(img_path):
+    embeddings = DeepFace.extract_faces(img_path=img_path)
+    return {"Report": str(embeddings)}
+
+
+if __name__ == '__main__':
+    print(image_to_image_compare(
+        img1_path="missing_FIR_images/02#_d6d26ac4-a989-494e-be57-674cda959d8079.jpg",
+        img2_path="missing_FIR_images/6-000#_d11e623e-b8d2-4c24-886f-6b33c3c36d95424.jpg"))
+
+    print(image_to_image_compare(
+        img1_path="missing_FIR_images/01#_95a58ed8-63a3-4b38-b744-a3b0725ccce51535.jpg",
+        img2_path="missing_FIR_images/01#_95a58ed8-63a3-4b38-b744-a3b0725ccce51535.jpg"))
+
+    print(get_embeddings(img_path="images/20230205_004933.jpg"))
+
+    print(image_to_image_compare(
+        img1_path="images/20230205_004933.jpg",
+        img2_path="images/20230205_004908.jpg" ))
+
+    print(image_to_image_compare(
+        img1_path="images/20230205_005906.jpg",
+        img2_path="images/20230205_005906.jpg"))
+
+    print(image_analyze(
+        img_path="missing_FIR_images/02#_d6d26ac4-a989-494e-be57-674cda959d8079.jpg"))
+
+    print(find_image_in_db(
+        img_path='missing_FIR_images/01#_95a58ed8-63a3-4b38-b744-a3b0725ccce51535.jpg',
+        db_path='missing_FIR_images'))
+
+    print(find_image_in_video(
+        img_path='images/Aamir_Khan_From_The_NDTV_Greenathon_at_Yash_Raj_Studios_(11).jpg',
+        video_path='videos/video1.mp4',
+        db_path='videos/video1-opencv'))

config.py

@@ -0,0 +1,28 @@
+# Copyright © DarkSide Assasins. All rights reserved.
+import os
+
+SECRET_KEY = os.urandom(32)
+SECURITY_PASSWORD_SALT = 'my_precious_two'
+BCRYPT_LOG_ROUNDS = 13
+WTF_CSRF_ENABLED = True
+DEBUG_TB_ENABLED = False
+DEBUG_TB_INTERCEPT_REDIRECTS = False
+SERVER_NAME = "127.0.0.1:5000"
+
+MAIL_SERVER = 'smtp.gmail.com'
+MAIL_PORT = 465
+MAIL_USE_TLS = False
+MAIL_USE_SSL = True
+
+basedir = os.path.abspath(os.path.dirname(__file__))
+
+DEBUG = True  
+
+SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'db.sqlite')
+SQLALCHEMY_TRACK_MODIFICATIONS = False
+
+CORS_HEADERS = 'Content-Type'
+
+MAIL_USERNAME = ' '
+MAIL_PASSWORD = ' '
+MAIL_DEFAULT_SENDER = " "

controllers/admin_controllers.py

@@ -0,0 +1,104 @@
+# Copyright © DarkSide Assasins. All rights reserved.
+
+from user.user import User
+from routes.admin_bp import admin_bp
+
+from flask import request, json,jsonify,render_template,url_for
+
+from .sendSignupEmail import SignUpEmail
+
+import base64
+
+import ai_face_detection
+
+class Resp():
+    def __init__(self,text):
+        self.resp = text
+
+@admin_bp.route("/addNewUser",methods = ['GET','POST'])
+def add_user():
+
+    token = ""
+
+    if request.method == 'POST':
+        all = User.query.all()
+        print(all)
+
+        em = request.json['email']
+        na = request.json['name']
+        ro = request.json['role']
+
+        exists = User.query.filter_by(email = em).first()
+
+        if exists is None:
+            print("New user")
+
+            newUser = User.addUser(email=em,name=na,role=ro)
+
+            User.addNewRecord(newUser)
+
+            all = User.query.all()
+            print(all)
+
+            if(User.query.filter_by(email = em) is not None):
+                print("db updated with new email")
+                emSent = SignUpEmail(em)
+                emSent.sendEmail()
+
+                token = emSent.token
+                print(type(token))
+                confirm_url = url_for('user_bp.signup_user',token = emSent.token,_external = True)
+                print(confirm_url)
+                html = render_template('activate.html',confirm_url = confirm_url)
+                emSent.send_email(to=em,subject = "Please confirm your email",template=html)
+
+                if(emSent.status):
+                    print("Mail sent")
+                    resp = "User added successfully. Check inbox to sign up"
+                else:
+                    resp = "Something went wrong"
+
+        else:
+            if exists.emVerified == True:
+                resp = "User already exists, please login"
+            else:
+                resp = "Perfrom email verification from inbox"
+
+        return jsonify({"resp" : resp})
+
+    if request.method == "GET":
+        resp = jsonify({"token" : token})
+        return resp
+
+
+@admin_bp.route("/search",methods=["GET","POST"])
+def run_algorithm():
+
+    res = {"Report":"Giaant blob of text"}
+    if request.method == "POST":
+        image_bytes = base64.b64decode(request.json["img_path"])
+
+        with open(request.json["name"],"wb") as fl:
+            fl.write(image_bytes)
+
+        im_path = request.json["name"]
+        db_path = "missing_FIR_images"
+        video_db_path = "videos/video1-opencv"
+        video_path='videos/video1.mp4'
+        img2_path="missing_FIR_images/6-000#_d11e623e-b8d2-4c24-886f-6b33c3c36d95424.jpg"
+
+        search = request.json["search_against"]
+        if "Image" in search:
+            print("Searching image to image")
+            res = ai_face_detection.find_image_in_db(img_path=im_path,db_path= db_path)
+
+        elif "Video" in search:
+            print("Searching image to video")
+            res = ai_face_detection.find_image_in_video(img_path = im_path,video_path= video_path,db_path=video_db_path)
+        else:
+            print("Snap filterss yayyy")
+            res = ai_face_detection.image_to_image_compare(img1_path = im_path,img2_path= img2_path)
+        print(res)
+        return res
+    if request.method == "GET":
+        return res

controllers/com_cntroller.py

@@ -0,0 +1,50 @@
+# Copyright © DarkSide Assasins. All rights reserved.
+
+from user.user import User,db
+from main import app
+from flask import json, jsonify, request
+import bcrypt
+
+from.sendSignupEmail import SignUpEmail
+
+@app.route('/',methods=['GET'])
+def home():
+    return "Hello world"
+
+@app.route('/login',methods=["POST"])
+def login():
+
+    db.session.query(User).delete()
+    db.session.commit()
+
+    admin = User.create_admin(User)
+    User.addNewRecord(admin)
+
+    all = User.query.all()
+    print(all)
+
+    email = request.json['email']
+    password = request.json['password']
+
+    login = User.query.filter_by(email = email).first()
+
+    if login is not None:
+
+        if(bcrypt.checkpw(password=password.encode('utf-8'),hashed_password=login.password.encode('utf-8'))):
+            print("User success")
+            resp = jsonify({
+                'id':login.id,
+                'email':login.email,
+                'name':login.name,
+                'password':str(login.password),
+                'phone':login.phone,
+                'role':login.role,
+                'resp':"Logged in"
+            })
+            return resp
+        else:
+            print("Unsuccessfull")
+            return jsonify({"res":"Password incorrect"})
+    else:
+        print("User doesn't exists")
+        return jsonify({"res":"Username doesn't exist, please contact the admin"})