upload all for testing

AI_Model/testModel.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8f6cdb9e17e9bf1494749a6d249a51d9418c5ab4cf9e5dc55f028274e603b385
+size 195232657

data/model.py

@@ -0,0 +1,3 @@
+from ultralytics import YOLO
+
+testModel = YOLO('AI_Model/testModel.pt')

test_api.py

@@ -0,0 +1,59 @@
+from fastapi import FastAPI
+from pydantic import BaseModel
+import asyncio
+from typing import List, Union
+from test_main import mainDet
+import uvicorn
+import logging
+from datetime import datetime
+import pytz
+import torch
+import json
+
+app = FastAPI()
+
+class Item(BaseModel):
+    url: str
+
+def get_bd_time():
+    bd_timezone = pytz.timezone("Asia/Dhaka")
+    time_now = datetime.now(bd_timezone)
+    current_time = time_now.strftime("%I:%M:%S %p")
+    return current_time
+
+
+async def process_item(item: Item):
+    try:
+        result = await mainDet(item.url)
+        result = json.loads(result)
+        return result
+    finally:
+        torch.cuda.empty_cache()
+        pass
+
+
+@app.get("/status")
+async def status():
+    return "AI Server in running"
+
+@app.post("/Test_bkash_sticker_count")
+async def create_items(items: Union[Item, List[Item]]):
+    try:
+        results = await process_item(items)
+        print("Result Sent to User:", results)
+        print("###################################################################################################")
+        print(items)
+        print("Last Execution Time : ", get_bd_time())
+        return results
+    except Exception as e:
+        return {"AI": f"Error: {str(e)}"}
+    finally:
+        torch.cuda.empty_cache()
+        pass
+
+if __name__ == "__main__":
+    try:
+        uvicorn.run(app, host="127.0.0.1", port=8000)
+    finally:
+        torch.cuda.empty_cache()
+

test_main.py

@@ -0,0 +1,41 @@
+import json
+import pandas as pd
+from PIL import Image
+from aiohttp import ClientSession
+from io import BytesIO
+import asyncio
+from data.model import testModel
+
+async def getImage(img_url, session):
+    async with session.get(img_url) as response:
+        img_data = await response.read()
+        return BytesIO(img_data)
+    
+
+async def detection(model,img_content,confidence):
+    img = Image.open(img_content)
+    # result = model(img)
+    result = model(source=img,device=0,conf=confidence)
+    detection = {}
+    data = json.loads(result[0].tojson())
+    if len(data) == 0:
+        res = {"AI": "No Detection"}
+        detection.update(res)
+    else:
+        df = pd.DataFrame(data)
+        name_counts = df['name'].value_counts().sort_index()
+        
+        for name, count in name_counts.items():
+            res = {name: count}
+            detection.update(res)
+    return detection
+
+async def mainDet(url):
+    async with ClientSession() as session:
+        image = await asyncio.create_task(getImage(url, session))
+        nbrtuDict = await asyncio.create_task(detection(testModel, image,0.6))
+
+        nagad_result = json.dumps({"sticker count test":nbrtuDict})
+
+        
+        return nagad_result