add roboflow

.gitignore

@@ -1 +1,4 @@
-.venv/
+.venv/
+
+__pycache__/
+*.jpg

app.py

@@ -4,14 +4,11 @@ from fastapi import FastAPI, File, UploadFile, HTTPException
 from fastapi.responses import HTMLResponse, JSONResponse
 import os
 from fastapi.middleware.cors import CORSMiddleware
-from utils import detect_licensePlate, licence_dect, lookup_user, lookup_user_metadata, ner_recog, read_text_img, upload_to_s3, vehicle_dect, write_to_upload
-from pinecone import Pinecone
-from deepface import DeepFace
+from utils import crop_images_from_detections, detect_licensePlate, licence_dect, lookup_user, lookup_user_metadata, ner_recog, read_text_img, upload_to_s3, vehicle_dect, write_to_upload
+# from pinecone import Pinecone
+# from deepface import DeepFace
 
 
-pc = Pinecone(api_key="dc53a991-1d1a-4f03-b718-1ec0df3b0f00")
-index = pc.Index("faces-id")
-
 app = FastAPI()
 
 app.add_middleware(
@@ -197,9 +194,11 @@ async def upload_files(
         with open(back_path, "wb") as back_file:
             back_file.write(await back.read())
         
-       
-
+        front_id = crop_images_from_detections(front_path)
+        back_id = crop_images_from_detections(back_path)
 
+        print(f"[*] ---- Text front ----> {front_id}")
+        print(f"[*] ---- Text back ----> {back_id}")
         
         front_url =  upload_to_s3(front_path)
         back_url =  upload_to_s3(back_path)
@@ -226,29 +225,29 @@ async def upload_files(
             
             
             print(f"[*] --- Serial ---> {serial_number}")
-            embedding =  DeepFace.represent(img_path=face_path, model_name='DeepFace')
-            embedding_vector = embedding[0]['embedding']
-            existing_user =  lookup_user_metadata(index, embedding_vector, serial_number)
-
-            print(f"[*] --- Existing user ---> {existing_user}")
-            if (len(existing_user["matches"]) <= 0):
-                print(f"[*] --- No match found --->")
-                await index.upsert(
-                    vectors=[
-                            {
-                                "id": str(uuid.uuid4()),
-                                "values" : embedding_vector,
-                                "metadata" : {"name": serial_number}
-                            }
-                        ],
-                        namespace="ns1"
-                    )
-            elif(len(existing_user["matches"]) > 0):
-                if (existing_user["matches"][0]["score"] >= 0.79):
-                    pass
-                    # return JSONResponse(content={"message": "This user and id card already exist"}, status_code=200)
-                elif(existing_user["matches"][0]["score"]):
-                    return HTTPException(content={"message" : "This card belongs to someone else"}, status_code=404)
+            # embedding =  DeepFace.represent(img_path=face_path, model_name='DeepFace')
+            # embedding_vector = embedding[0]['embedding']
+            # existing_user =  lookup_user_metadata(index, embedding_vector, serial_number)
+
+            # print(f"[*] --- Existing user ---> {existing_user}")
+            # if (len(existing_user["matches"]) <= 0):
+            #     print(f"[*] --- No match found --->")
+            #     await index.upsert(
+            #         vectors=[
+            #                 {
+            #                     "id": str(uuid.uuid4()),
+            #                     "values" : embedding_vector,
+            #                     "metadata" : {"name": serial_number}
+            #                 }
+            #             ],
+            #             namespace="ns1"
+            #         )
+            # elif(len(existing_user["matches"]) > 0):
+            #     if (existing_user["matches"][0]["score"] >= 0.79):
+            #         pass
+            #         # return JSONResponse(content={"message": "This user and id card already exist"}, status_code=200)
+            #     elif(existing_user["matches"][0]["score"]):
+            #         return HTTPException(content={"message" : "This card belongs to someone else"}, status_code=404)
 
 
 

requirements.txt

@@ -11,4 +11,5 @@ python-multipart
 pinecone-client
 deepface
 tf-keras
-tensorflow==2.12
+tensorflow==2.12
+roboflow

test.py

@@ -0,0 +1,56 @@
+# import tempfile
+# from roboflow import Roboflow
+# from PIL import Image
+# import json
+
+# def set_image_dpi(file_path):
+
+#     im = Image.open(file_path)
+
+#     length_x, width_y = im.size
+
+#     factor = min(1, float(1024.0 / length_x))
+
+#     size = int(factor * length_x), int(factor * width_y)
+
+#     im_resized = im.resize(size, Image.LANCZOS)
+
+#     temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.png')
+#     i = 0
+#     i =+ 1
+#     temp_filename = f'enhanced{i}.jpg'
+
+#     im_resized.save(temp_filename, dpi=(300, 300))
+
+#     return temp_filename
+
+# def crop_images_from_detections(detections):
+#     for detection in detections:
+#         image_path = detection["image_path"]
+#         image = Image.open(image_path)
+        
+#         x1 = detection['x'] - detection['width'] / 2
+#         x2 = detection['x'] + detection['width'] / 2
+#         y1 = detection['y'] - detection['height'] / 2
+#         y2 = detection['y'] + detection['height'] / 2
+#         box = (x1, y1, x2, y2)
+        
+#         cropped_image = image.crop(box)
+#         path= f"{detection['class']}.jpg"
+        
+#         cropped_image.save(path)  # Or save the image using cropped_image.save('path_to_save_image')
+#         set_image_dpi(path)
+
+
+
+# rf = Roboflow(api_key="P4usj8uPwcbnflvyJIAB")
+
+# project = rf.workspace("ntchindagiscard").project("id_card_annotation")
+
+
+# model = project.version(1).model
+
+# model.predict("2ed1bdb5-5c09-40a0-a39f-9ff6c15380bf-front.jpg", confidence=20, overlap=50).save('prediction.jpg')
+# result = model.predict("2ed1bdb5-5c09-40a0-a39f-9ff6c15380bf-front.jpg", confidence=20, overlap=50)
+# crop_images_from_detections(result)
+# print(result)

utils.py

@@ -1,4 +1,5 @@
 import os
+import tempfile
 import uuid
 from fastapi import File
 from paddleocr import PaddleOCR
@@ -11,8 +12,7 @@ from botocore.exceptions import NoCredentialsError
 from dotenv import load_dotenv
 from colorthief import ColorThief
 import logging
-
-import logging
+from roboflow import Roboflow
 
 # Configure logging
 logging.basicConfig(
@@ -32,10 +32,55 @@ nlp_ner = spacy.load("en_pipeline")
 detector = YOLO('best.pt')
 vehicle = YOLO('yolov8x.pt')
 
+rf = Roboflow(api_key="P4usj8uPwcbnflvyJIAB")
+project = rf.workspace("ntchindagiscard").project("id_card_annotation")
+model = project.version(1).model
+
 aws_access_key_id=os.getenv('AWS_ACCESS_KEY')
 aws_secret_access_key = os.getenv('AWS_SECRET_KEY')
 bucket_name='vvims'
 
+
+def read_text_img(img_path:str) -> str:
+    """
+        Read text from images
+
+        Args:
+        - img_path: Path to the images in which the text will be extracted
+
+        Returns:
+        - text: The extracted text
+    """
+
+    result = ocr_model.ocr(img_path)
+    text = ''
+    if result[0]:
+        for res in result[0]:
+            text += res[1][0] + ' '
+    return text
+
+def crop_images_from_detections(image_path):
+    results = {}
+    detections = model.predict( image_path , confidence=20, overlap=50)
+    for detection in detections:
+        image_path = image_path
+        image = Image.open(image_path)
+        
+        x1 = detection['x'] - detection['width'] / 2
+        x2 = detection['x'] + detection['width'] / 2
+        y1 = detection['y'] - detection['height'] / 2
+        y2 = detection['y'] + detection['height'] / 2
+        box = (x1, y1, x2, y2)
+        
+        cropped_image = image.crop(box)
+        path= f"detections.jpg"
+        cropped_image.save(path)
+        text = read_text_img(path)
+
+        results[detection['class']] = text
+    return results
+    
+
 # Function to upload a file to S3
 def upload_to_s3(
         file_path, 
@@ -97,24 +142,6 @@ def ner_recog(text:str) -> dict:
 
     return {"entities": entities}
 
-
-def read_text_img(img_path:str) -> str:
-    """
-        Read text from images
-
-        Args:
-        - img_path: Path to the images in which the text will be extracted
-
-        Returns:
-        - text: The extracted text
-    """
-
-    result = ocr_model.ocr(img_path)
-    text = ''
-    if result[0]:
-        for res in result[0]:
-            text += res[1][0] + ' '
-    return text
 def detect_licensePlate(img: str) -> dict:
     image = Image.open(img)
     results = vehicle(source=img, cls=['car', 'bus', 'truck', 'motorcycle'], conf=0.7)