add: image preprocessing

app.py

@@ -1,26 +1,18 @@
-import secrets
-import cv2
-import numpy as np
-from PIL.ImageOps import grayscale
-from fastapi.encoders import jsonable_encoder
-from src.utils import supabaseGetPublicURL, deductAndTrackCredit, returnBytesData
-from fastapi import FastAPI, File, UploadFile, Header, HTTPException, Form, Depends
-from src.pipelines.completePipeline import Pipeline
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse
-from supabase import create_client, Client
-from typing import Dict, Union, List
-from io import BytesIO
-from PIL import Image
-import pandas as pd
-import base64
-import os
-from pydantic import BaseModel
-import replicate
+"""
+project @ NTO-TCP-HF
+created @ 2024-10-28
+author  @ github.com/ishworrsubedii
+"""
 
-pipeline = Pipeline()
+from fastapi import FastAPI
+from starlette.middleware.cors import CORSMiddleware
 
-app = FastAPI(title="Magical Mirror Web Plugin", description="API for Magical Mirror Application ", version="1.0")
+from src.api.image_prep_api import preprocessing_router
+from src.api.nto_api import nto_cto_router
+
+app = FastAPI()
+app.include_router(nto_cto_router, tags=["NTO-CTO"])
+app.include_router(preprocessing_router, tags=["Image-Preprocessing"])
 
 app.add_middleware(
     CORSMiddleware,
@@ -29,434 +21,3 @@ app.add_middleware(
     allow_methods=["*"],
     allow_headers=["*"],
 )
-url: str = os.getenv("SUPABASE_URL")
-key: str = os.getenv("SUPABASE_KEY")
-supabase: Client = create_client(supabase_key=key, supabase_url=url)
-bucket = supabase.storage.from_("JewelMirrorOutputs")
-
-
-def replicate_run(input):
-    output = replicate.run(
-        "stability-ai/stable-diffusion-inpainting:95b7223104132402a9ae91cc677285bc5eb997834bd2349fa486f53910fd68b3",
-        input=input)
-    return output
-
-
-class NecklaceTryOnIDEntity(BaseModel):
-    necklaceImageId: str
-    necklaceCategory: str
-    storename: str
-    api_token: str
-
-
-@app.post("/clothingTryOnV2")
-async def clothing_try_on_v2(image: UploadFile = File(...), clothing_type: str = Form(...)):
-    image_bytes = await image.read()
-    image = Image.open(BytesIO(image_bytes)).convert("RGB")
-
-    mask = await pipeline.shoulderPointMaskGeneration_(image=image)
-
-    mask_img_base_64, act_img_base_64 = BytesIO(), BytesIO()
-    mask.save(mask_img_base_64, format="WEBP")
-    image.save(act_img_base_64, format="WEBP")
-    mask_bytes_ = base64.b64encode(mask_img_base_64.getvalue()).decode("utf-8")
-    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
-
-    mask_data_uri = f"data:image/webp;base64,{mask_bytes_}"
-    image_data_uri = f"data:image/webp;base64,{image_bytes_}"
-
-    input = {
-        "mask": mask_data_uri,
-        "image": image_data_uri,
-        "prompt": f"Dull {clothing_type}, non-reflective clothing, properly worn, natural setting, elegant, natural look, neckline without jewellery, simple, perfect eyes, perfect face, perfect body, high quality, realistic, photorealistic, high resolution,traditional full sleeve blouse",
-        "negative_prompt": "necklaces, jewellery, jewelry, necklace, neckpiece, garland, chain, neck wear, jewelled neck, jeweled neck, necklace on neck, jewellery on neck, accessories, watermark, text, changed background, wider body, narrower body, bad proportions, extra limbs, mutated hands, changed sizes, altered proportions, unnatural body proportions, blury, ugly",
-        "num_inference_steps": 25
-    }
-
-    output = replicate_run(input)
-
-    response = {
-        "output": f"{output[0]}",
-        'code': 200
-    }
-
-    return JSONResponse(content=response, status_code=200)
-
-
-@app.post("/clothingTryOn")
-async def clothing_try_on(image: UploadFile = File(...),
-                          mask: UploadFile = File(...), clothing_type: str = Form(...)):
-    image_bytes = await image.read()
-    mask_bytes = await mask.read()
-    image, mask = Image.open(BytesIO(image_bytes)).convert("RGB"), Image.open(
-        BytesIO(mask_bytes)).convert("RGB")
-
-    actual_image = image.copy()
-
-    jewellery_mask = Image.fromarray(
-        np.bitwise_and(np.array(mask), np.array(image))
-    )
-    arr_orig = np.array(grayscale(mask))
-
-    image = cv2.inpaint(np.array(image), arr_orig, 15, cv2.INPAINT_TELEA)
-    image = Image.fromarray(image).resize((512, 512))
-
-    arr = arr_orig.copy()
-    mask_y = np.where(arr == arr[arr != 0][0])[0][0]
-    arr[mask_y:, :] = 255
-
-    mask = Image.fromarray(arr).resize((512, 512))
-
-    mask_img_base_64, act_img_base_64 = BytesIO(), BytesIO()
-    mask.save(mask_img_base_64, format="WEBP")
-    image.save(act_img_base_64, format="WEBP")
-    mask_bytes_ = base64.b64encode(mask_img_base_64.getvalue()).decode("utf-8")
-    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
-
-    mask_data_uri = f"data:image/webp;base64,{mask_bytes_}"
-    image_data_uri = f"data:image/webp;base64,{image_bytes_}"
-
-    input = {
-        "mask": mask_data_uri,
-        "image": image_data_uri,
-        "prompt": f"Dull {clothing_type}, non-reflective clothing, properly worn, natural setting, elegant, natural look, neckline without jewellery, simple, perfect eyes, perfect face, perfect body, high quality, realistic, photorealistic, high resolution,traditional full sleeve blouse",
-        "negative_prompt": "necklaces, jewellery, jewelry, necklace, neckpiece, garland, chain, neck wear, jewelled neck, jeweled neck, necklace on neck, jewellery on neck, accessories, watermark, text, changed background, wider body, narrower body, bad proportions, extra limbs, mutated hands, changed sizes, altered proportions, unnatural body proportions, blury, ugly",
-        "num_inference_steps": 25
-    }
-
-    output = replicate_run(input)
-    image_base = str(output[0])
-    image_base_64 = image_base.split(",")[1]
-    image_data = base64.b64decode(image_base_64)
-
-    output_image = Image.open(BytesIO(image_data)).resize(actual_image.size)
-    output_image = np.bitwise_and(
-        np.array(output_image),
-        np.bitwise_not(np.array(Image.fromarray(arr_orig).convert("RGB"))),
-    )
-
-    result = Image.fromarray(np.bitwise_or(np.array(output_image), np.array(jewellery_mask)))
-
-    in_mem_file = BytesIO()
-    result.save(in_mem_file, format="WEBP", quality=85)
-    base_64_output = base64.b64encode(in_mem_file.getvalue()).decode('utf-8')
-
-    response = {
-        "output": f"data:image/WEBP;base64,{base_64_output}",
-        'code': 200
-    }
-
-    return JSONResponse(content=response, status_code=200)
-
-
-@app.post("/productData/{storeId}")
-async def product_data(
-        storeId: str,
-        filterattributes: List[Dict[str, Union[str, int, float]]],
-        storename: str = Header(default="default")
-):
-    """Filters product data based on the provided attributes and store ID."""
-
-    try:
-        response = supabase.table('MagicMirror').select("*").execute()
-        df = pd.DataFrame(response.dict()["data"])
-
-        df = df[df["StoreName"] == storeId]
-
-        # Preprocess filterattributes to handle multiple or duplicated attributes
-        attribute_dict = {}
-        for attr in filterattributes:
-            key, value = list(attr.items())[
-                0]  # This will convert the dictionary into a list and get the key and value.
-            if key in attribute_dict:  # This will check if the key is already present in the dictionary.
-                if isinstance(attribute_dict[key],
-                              list):  # This will create a list if there are multiple values for the same key and we are doing or operation.
-                    attribute_dict[key].append(value)  # This will append the value to the list.
-                else:
-                    attribute_dict[key] = [attribute_dict[key], value]
-            else:
-                attribute_dict[key] = [value]  # This will create a list if there is only one value for the key.
-
-        priceFrom = None
-        priceTo = None
-        weightFrom = None
-        weightTo = None
-        weightAscending = None
-        priceAscending = None
-        idAscending = None
-        dateAscending = None
-
-        for key, value in attribute_dict.items():
-            if key == 'priceFrom':
-                priceFrom = value[0]
-
-            elif key == "priceTo":
-                priceTo = value[0]
-
-            elif key == "priceAscending":
-                priceAscending = value[0]
-
-            elif key == "weightFrom":
-                weightFrom = value[0]
-
-            elif key == "weightTo":
-                weightTo = value[0]
-
-            elif key == "weightAscending":
-                weightAscending = value[0]
-
-            elif key == "idAscending":
-                idAscending = value[0]
-
-            elif key == "dateAscending":
-                dateAscending = value[0]
-
-        df["image_url"] = df.apply(
-            lambda row: supabaseGetPublicURL(f"{row['StoreName']}/{row['Category']}/image/{row['Id']}.png"),
-            axis=1)
-        df["thumbnail_url"] = df.apply(
-            lambda row: supabaseGetPublicURL(f"{row['StoreName']}/{row['Category']}/thumbnail/{row['Id']}.png"),
-            axis=1)
-
-        df.reset_index(drop=True, inplace=True)
-        for key, values in attribute_dict.items():
-            try:
-                df = df[df[key].isin(values)]
-
-            except:
-                pass
-
-        # applying filter for price and weight
-        if priceFrom is not None:
-            df = df[df["Price"] >= priceFrom]
-        if priceTo is not None:
-            df = df[df["Price"] <= priceTo]
-        if weightFrom is not None:
-            df = df[df["Weight"] >= weightFrom]
-        if weightTo is not None:
-            df = df[df["Weight"] <= weightTo]
-
-        if priceAscending is not None:
-            if priceAscending == 1:
-                value = True
-
-            else:
-                value = False
-            df = df.sort_values(by="Price", ascending=value)
-        if weightAscending is not None:
-            if weightAscending == 1:
-                value = True
-
-            else:
-                value = False
-            df = df.sort_values(by="Weight", ascending=value)
-
-        if idAscending is not None:
-            if idAscending == 1:
-                value = True
-            else:
-                value = False
-            df = df.sort_values(by="Id", ascending=value)
-
-        if dateAscending is not None:
-            if dateAscending == 1:
-                value = True
-            else:
-                value = False
-            df = df.sort_values(by="UpdatedAt", ascending=value)
-
-        df = df.drop(["CreatedAt", "EstimatedPrice"], axis=1)
-
-        result = {}
-        for _, row in df.iterrows():
-            category = row["Category"]
-            if category not in result:  # this is for checking duplicate category
-                result[category] = []
-            result[category].append(row.to_dict())
-
-        return JSONResponse(content=jsonable_encoder(result))  # this will convert the result into json format.
-
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Failed to fetch or process data: {e}")
-
-
-async def parse_necklace_try_on_id(necklaceImageId: str = Form(...),
-                                   necklaceCategory: str = Form(...),
-                                   storename: str = Form(...),
-                                   api_token: str = Form(...)) -> NecklaceTryOnIDEntity:
-    return NecklaceTryOnIDEntity(
-        necklaceImageId=necklaceImageId,
-        necklaceCategory=necklaceCategory,
-        storename=storename,
-        api_token=api_token
-    )
-
-
-@app.post("/necklaceTryOnID")
-async def necklace_try_on_id(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
-                             image: UploadFile = File(...)):
-    data, _ = supabase.table("APIKeyList").select("*").filter("API_KEY", "eq",
-                                                              necklace_try_on_id.api_token).execute()
-
-    api_key_actual = data[1][0]['API_KEY']
-    if api_key_actual != necklace_try_on_id.api_token:
-        return JSONResponse(content={"error": "Invalid API Key"}, status_code=401)
-
-    else:
-        imageBytes = await image.read()
-
-        jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
-
-        try:
-            image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
-
-        except:
-            error_message = {
-                "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
-            }
-
-            return JSONResponse(content=error_message, status_code=404)
-
-        result, headetText, mask = await pipeline.necklaceTryOn_(image=image, jewellery=jewellery,
-                                                                 storename=necklace_try_on_id.storename)
-
-        inMemFile = BytesIO()
-        inMemFileMask = BytesIO()
-        result.save(inMemFile, format="WEBP", quality=85)
-        mask.save(inMemFileMask, format="WEBP", quality=85)
-        outputBytes = inMemFile.getvalue()
-        maskBytes = inMemFileMask.getvalue()
-        response = {
-            "output": f"data:image/WEBP;base64,{base64.b64encode(outputBytes).decode('utf-8')}",
-            "mask": f"data:image/WEBP;base64,{base64.b64encode(maskBytes).decode('utf-8')}"
-        }
-        creditResponse = deductAndTrackCredit(storename=necklace_try_on_id.storename, endpoint="/necklaceTryOnID")
-        if creditResponse == "No Credits Available":
-            response = {
-                "error": "No Credits Remaining"
-            }
-
-            return JSONResponse(content=response)
-
-        else:
-            return JSONResponse(content=response)
-
-
-@app.post("/necklace_try_on_id_url")
-async def necklace_try_on_id_url(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
-                                 image: UploadFile = File(...)):
-    imageBytes = await image.read()
-
-    jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
-
-    try:
-        image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
-
-    except:
-        error_message = {
-            "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
-        }
-
-        return JSONResponse(content=error_message, status_code=404)
-
-    result, headerText, mask = await pipeline.necklaceTryOn_(
-        image=image, jewellery=jewellery, storename=necklace_try_on_id.storename
-    )
-
-    response = {}
-    for image_obj, prefix in [(result, "result"), (mask, "mask")]:
-        file_name = f"{prefix}{secrets.token_hex(24)}.webp"
-
-        inMemFile = BytesIO()
-        image_obj.save(inMemFile, format="WEBP", quality=85)
-        in_mem_file = inMemFile.getvalue()
-
-        bucket.upload(file_name, in_mem_file)
-
-        public_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/JewelMirrorOutputs/{file_name}"
-
-        response[f"{prefix}_url"] = public_url
-
-    return JSONResponse(content=response)
-
-
-@app.post("/canvasPoints")
-async def canvas_points(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
-                        image: UploadFile = File(...)):
-    imageBytes = await image.read()
-
-    jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
-
-    try:
-        image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
-
-    except:
-        error_message = {
-            "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
-        }
-
-        return JSONResponse(content=error_message, status_code=404)
-
-    response = await pipeline.canvasPoint(
-        image=image, jewellery=jewellery, storename=necklace_try_on_id.storename
-    )
-
-    creditResponse = deductAndTrackCredit(storename=necklace_try_on_id.storename, endpoint="/necklaceTryOnID")
-    if creditResponse == "No Credits Available":
-        response = {
-            "error": "No Credits Remaining"
-        }
-
-        return JSONResponse(content=response)
-
-    else:
-        return JSONResponse(content=response)
-
-
-@app.post("/necklaceTryOnWithPoints")
-async def necklace_try_on_with_points(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
-                                      image: UploadFile = File(...),
-                                      left_x: int = Form(...),
-                                      left_y: int = Form(...),
-                                      right_x: int = Form(...),
-                                      right_y: int = Form(...)):
-    imageBytes = await image.read()
-
-    jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
-
-    try:
-        image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
-
-    except:
-        error_message = {
-            "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
-        }
-
-        return JSONResponse(content=error_message, status_code=404)
-
-    result, headerText, mask = await pipeline.necklaceTryOnWithPoints_(
-        image=image, jewellery=jewellery, left_shoulder=(left_x, left_y), right_shoulder=(right_x, right_y),
-        storename=necklace_try_on_id.storename
-    )
-
-    inMemFile = BytesIO()
-    inMemFileMask = BytesIO()
-    result.save(inMemFile, format="WEBP", quality=85)
-    mask.save(inMemFileMask, format="WEBP", quality=85)
-    outputBytes = inMemFile.getvalue()
-    maskBytes = inMemFileMask.getvalue()
-    response = {
-        "output": f"data:image/WEBP;base64,{base64.b64encode(outputBytes).decode('utf-8')}",
-        "mask": f"data:image/WEBP;base64,{base64.b64encode(maskBytes).decode('utf-8')}"
-    }
-
-    creditResponse = deductAndTrackCredit(storename=necklace_try_on_id.storename, endpoint="/necklaceTryOnID")
-    if creditResponse == "No Credits Available":
-        response = {
-            "error": "No Credits Remaining"
-        }
-
-        return JSONResponse(content=response)
-
-    else:
-        return JSONResponse(content=response)

src/api/__init__.py

@@ -0,0 +1,5 @@
+"""
+project @ NTO-TCP-HF
+created @ 2024-10-28
+author  @ github/ishworrsubedii
+"""

src/api/image_prep_api.py

@@ -0,0 +1,158 @@
+"""
+project @ NTO-TCP-HF
+created @ 2024-10-28
+author  @ github.com/ishworrsubedii
+"""
+import base64
+import os
+from io import BytesIO
+import numpy as np
+import replicate
+import requests
+from PIL import Image
+from fastapi import APIRouter, UploadFile, File, HTTPException
+from fastapi.responses import JSONResponse
+
+from src.components.auto_crop import crop_transparent_image
+
+preprocessing_router = APIRouter()
+
+rmbg: str = os.getenv("RMBG")
+
+enhancer: str = os.getenv("ENHANCER")
+
+
+def replicate_bg(input):
+    output = replicate.run(
+        rmbg,
+        input=input
+    )
+    return output
+
+
+def replicate_enhancer(input):
+    output = replicate.run(
+        enhancer,
+        input=input
+    )
+    return output
+
+
+@preprocessing_router.post("/rem_bg")
+async def remove_background(image: UploadFile = File(...)):
+    image_bytes = await image.read()
+    image = Image.open(BytesIO(image_bytes)).convert("RGB")
+    act_img_base_64 = BytesIO()
+    image.save(act_img_base_64, format="WEBP")
+    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
+
+    image_data_uri = f"data:image/WEBP;base64,{image_bytes_}"
+
+    input = {
+        "image": image_data_uri,
+    }
+    output = replicate_bg(input)
+
+    response = requests.get(output)
+    base_64 = base64.b64encode(response.content).decode('utf-8')
+    base64_prefix = "data:image/WEBP;base64,"
+
+    try:
+        response = {
+            "output": f"{base64_prefix}{base_64}",
+            'code': 200
+        }
+
+        return JSONResponse(content=response, status_code=200)
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to process image: {e}")
+
+
+@preprocessing_router.post("/upscale_image")
+async def upscale_image(image: UploadFile = File(...), scale: int = 1):
+    image_bytes = await image.read()
+    image = Image.open(BytesIO(image_bytes)).convert("RGB")
+    act_img_base_64 = BytesIO()
+    image.save(act_img_base_64, format="WEBP")
+    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
+
+    image_data_uri = f"data:image/WEBP;base64,{image_bytes_}"
+
+    input = {
+        "image": image_data_uri,
+        "scale": scale,
+        "face_enhance": False
+
+    }
+    output = replicate_enhancer(input)
+
+    response = requests.get(output)
+    base_64 = base64.b64encode(response.content).decode('utf-8')
+    base64_prefix = image_data_uri.split(",")[0] + ","
+
+    try:
+        response = {
+            "output": f"{base64_prefix}{base_64}",
+            'code': 200
+        }
+
+        return JSONResponse(content=response, status_code=200)
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to process image: {e}")
+
+
+@preprocessing_router.post("/crop_transparent")
+async def crop_transparent(image: UploadFile):
+    if not image.content_type == "image/png":
+        raise HTTPException(status_code=400, detail="Only PNG files are supported")
+
+    try:
+        contents = await image.read()
+
+        cropped_image_bytes, metadata = crop_transparent_image(contents)
+
+        base64_image = base64.b64encode(cropped_image_bytes).decode('utf-8')
+        base64_prefix = "data:image/png;base64,"
+
+        return {
+            "status": "success",
+            "data": {
+                "image": f"{base64_prefix}{base64_image}",
+                "metadata": metadata
+            }
+        }
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
+
+
+@preprocessing_router.post("/background_replace")
+async def nto_id(image: UploadFile = File(...), bg_image: UploadFile = File(...)):
+    image_bytes = await image.read()
+    bg_image = await bg_image.read()
+
+    image, bg_image = Image.open(BytesIO(image_bytes)).convert("RGBA"), Image.open(BytesIO(bg_image)).convert("RGB")
+    width, height = bg_image.size
+    background = Image.fromarray(np.array(bg_image)).resize((width, height))
+    orig_img = Image.fromarray(np.array(image)).resize((width, height))
+    background.paste(orig_img, (0, 0), mask=orig_img)
+    act_img_base_64 = BytesIO()
+    background.save(act_img_base_64, format="WEBP")
+    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
+
+    image_data_uri = f"data:image/webp;base64,{image_bytes_}"
+
+    try:
+        response = {
+            "output": f"{image_data_uri}",
+            'code': 200
+        }
+
+        return JSONResponse(content=response, status_code=200)
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to process image: {e}")

src/api/nto_api.py

@@ -0,0 +1,426 @@
+"""
+project @ NTO-TCP-HF
+created @ 2024-10-28
+author  @ github/ishworrsubedii
+"""
+import cv2
+import numpy as np
+from PIL.ImageOps import grayscale
+from fastapi.encoders import jsonable_encoder
+from src.utils import supabaseGetPublicURL, deductAndTrackCredit, returnBytesData
+from fastapi import File, UploadFile, Header, HTTPException, Form, Depends, APIRouter
+from src.pipelines.completePipeline import Pipeline
+from fastapi.responses import JSONResponse
+from supabase import create_client, Client
+from typing import Dict, Union, List
+from io import BytesIO
+from PIL import Image
+import pandas as pd
+import base64
+import os
+from pydantic import BaseModel
+import replicate
+
+pipeline = Pipeline()
+
+nto_cto_router = APIRouter()
+
+url: str = os.getenv("SUPABASE_URL")
+key: str = os.getenv("SUPABASE_KEY")
+
+supabase_storage: str = os.getenv("SUPABASE_STORAGE")
+
+cto_replicate: str = os.getenv(
+    "CTO")
+
+supabase: Client = create_client(supabase_key=key, supabase_url=url)
+bucket = supabase.storage.from_(supabase_storage)
+
+
+def replicate_run_cto(input):
+    output = replicate.run(
+        cto_replicate,
+        input=input)
+    return output
+
+
+class NecklaceTryOnIDEntity(BaseModel):
+    necklaceImageId: str
+    necklaceCategory: str
+    storename: str
+    api_token: str
+
+
+@nto_cto_router.post("/clothingTryOnV2")
+async def clothing_try_on_v2(image: UploadFile = File(...), clothing_type: str = Form(...)):
+    image_bytes = await image.read()
+    image = Image.open(BytesIO(image_bytes)).convert("RGB")
+
+    mask = await pipeline.shoulderPointMaskGeneration_(image=image)
+
+    mask_img_base_64, act_img_base_64 = BytesIO(), BytesIO()
+    mask.save(mask_img_base_64, format="WEBP")
+    image.save(act_img_base_64, format="WEBP")
+    mask_bytes_ = base64.b64encode(mask_img_base_64.getvalue()).decode("utf-8")
+    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
+
+    mask_data_uri = f"data:image/webp;base64,{mask_bytes_}"
+    image_data_uri = f"data:image/webp;base64,{image_bytes_}"
+
+    input = {
+        "mask": mask_data_uri,
+        "image": image_data_uri,
+        "prompt": f"Dull {clothing_type}, non-reflective clothing, properly worn, natural setting, elegant, natural look, neckline without jewellery, simple, perfect eyes, perfect face, perfect body, high quality, realistic, photorealistic, high resolution,traditional full sleeve blouse",
+        "negative_prompt": "necklaces, jewellery, jewelry, necklace, neckpiece, garland, chain, neck wear, jewelled neck, jeweled neck, necklace on neck, jewellery on neck, accessories, watermark, text, changed background, wider body, narrower body, bad proportions, extra limbs, mutated hands, changed sizes, altered proportions, unnatural body proportions, blury, ugly",
+        "num_inference_steps": 25
+    }
+
+    output = replicate_run_cto(input)
+
+    response = {
+        "output": f"{output[0]}",
+        'code': 200
+    }
+
+    return JSONResponse(content=response, status_code=200)
+
+
+@nto_cto_router.post("/clothingTryOn")
+async def clothing_try_on(image: UploadFile = File(...),
+                          mask: UploadFile = File(...), clothing_type: str = Form(...)):
+    image_bytes = await image.read()
+    mask_bytes = await mask.read()
+    image, mask = Image.open(BytesIO(image_bytes)).convert("RGB"), Image.open(
+        BytesIO(mask_bytes)).convert("RGB")
+
+    actual_image = image.copy()
+
+    jewellery_mask = Image.fromarray(
+        np.bitwise_and(np.array(mask), np.array(image))
+    )
+    arr_orig = np.array(grayscale(mask))
+
+    image = cv2.inpaint(np.array(image), arr_orig, 15, cv2.INPAINT_TELEA)
+    image = Image.fromarray(image).resize((512, 512))
+
+    arr = arr_orig.copy()
+    mask_y = np.where(arr == arr[arr != 0][0])[0][0]
+    arr[mask_y:, :] = 255
+
+    mask = Image.fromarray(arr).resize((512, 512))
+
+    mask_img_base_64, act_img_base_64 = BytesIO(), BytesIO()
+    mask.save(mask_img_base_64, format="WEBP")
+    image.save(act_img_base_64, format="WEBP")
+    mask_bytes_ = base64.b64encode(mask_img_base_64.getvalue()).decode("utf-8")
+    image_bytes_ = base64.b64encode(act_img_base_64.getvalue()).decode("utf-8")
+
+    mask_data_uri = f"data:image/webp;base64,{mask_bytes_}"
+    image_data_uri = f"data:image/webp;base64,{image_bytes_}"
+
+    input = {
+        "mask": mask_data_uri,
+        "image": image_data_uri,
+        "prompt": f"Dull {clothing_type}, non-reflective clothing, properly worn, natural setting, elegant, natural look, neckline without jewellery, simple, perfect eyes, perfect face, perfect body, high quality, realistic, photorealistic, high resolution,traditional full sleeve blouse",
+        "negative_prompt": "necklaces, jewellery, jewelry, necklace, neckpiece, garland, chain, neck wear, jewelled neck, jeweled neck, necklace on neck, jewellery on neck, accessories, watermark, text, changed background, wider body, narrower body, bad proportions, extra limbs, mutated hands, changed sizes, altered proportions, unnatural body proportions, blury, ugly",
+        "num_inference_steps": 25
+    }
+
+    output = replicate_run_cto(input)
+    image_base = str(output[0])
+    image_base_64 = image_base.split(",")[1]
+    image_data = base64.b64decode(image_base_64)
+
+    output_image = Image.open(BytesIO(image_data)).resize(actual_image.size)
+    output_image = np.bitwise_and(
+        np.array(output_image),
+        np.bitwise_not(np.array(Image.fromarray(arr_orig).convert("RGB"))),
+    )
+
+    result = Image.fromarray(np.bitwise_or(np.array(output_image), np.array(jewellery_mask)))
+
+    in_mem_file = BytesIO()
+    result.save(in_mem_file, format="WEBP", quality=85)
+    base_64_output = base64.b64encode(in_mem_file.getvalue()).decode('utf-8')
+
+    response = {
+        "output": f"data:image/WEBP;base64,{base_64_output}",
+        'code': 200
+    }
+
+    return JSONResponse(content=response, status_code=200)
+
+
+@nto_cto_router.post("/productData/{storeId}")
+async def product_data(
+        storeId: str,
+        filterattributes: List[Dict[str, Union[str, int, float]]],
+        storename: str = Header(default="default")
+):
+    """Filters product data based on the provided attributes and store ID."""
+
+    try:
+        response = supabase.table('MagicMirror').select("*").execute()
+        df = pd.DataFrame(response.dict()["data"])
+
+        df = df[df["StoreName"] == storeId]
+
+        # Preprocess filterattributes to handle multiple or duplicated attributes
+        attribute_dict = {}
+        for attr in filterattributes:
+            key, value = list(attr.items())[
+                0]  # This will convert the dictionary into a list and get the key and value.
+            if key in attribute_dict:  # This will check if the key is already present in the dictionary.
+                if isinstance(attribute_dict[key],
+                              list):  # This will create a list if there are multiple values for the same key and we are doing or operation.
+                    attribute_dict[key].append(value)  # This will append the value to the list.
+                else:
+                    attribute_dict[key] = [attribute_dict[key], value]
+            else:
+                attribute_dict[key] = [value]  # This will create a list if there is only one value for the key.
+
+        priceFrom = None
+        priceTo = None
+        weightFrom = None
+        weightTo = None
+        weightAscending = None
+        priceAscending = None
+        idAscending = None
+        dateAscending = None
+
+        for key, value in attribute_dict.items():
+            if key == 'priceFrom':
+                priceFrom = value[0]
+
+            elif key == "priceTo":
+                priceTo = value[0]
+
+            elif key == "priceAscending":
+                priceAscending = value[0]
+
+            elif key == "weightFrom":
+                weightFrom = value[0]
+
+            elif key == "weightTo":
+                weightTo = value[0]
+
+            elif key == "weightAscending":
+                weightAscending = value[0]
+
+            elif key == "idAscending":
+                idAscending = value[0]
+
+            elif key == "dateAscending":
+                dateAscending = value[0]
+
+        df["image_url"] = df.apply(
+            lambda row: supabaseGetPublicURL(f"{row['StoreName']}/{row['Category']}/image/{row['Id']}.png"),
+            axis=1)
+        df["thumbnail_url"] = df.apply(
+            lambda row: supabaseGetPublicURL(f"{row['StoreName']}/{row['Category']}/thumbnail/{row['Id']}.png"),
+            axis=1)
+
+        df.reset_index(drop=True, inplace=True)
+        for key, values in attribute_dict.items():
+            try:
+                df = df[df[key].isin(values)]
+
+            except:
+                pass
+
+        # applying filter for price and weight
+        if priceFrom is not None:
+            df = df[df["Price"] >= priceFrom]
+        if priceTo is not None:
+            df = df[df["Price"] <= priceTo]
+        if weightFrom is not None:
+            df = df[df["Weight"] >= weightFrom]
+        if weightTo is not None:
+            df = df[df["Weight"] <= weightTo]
+
+        if priceAscending is not None:
+            if priceAscending == 1:
+                value = True
+
+            else:
+                value = False
+            df = df.sort_values(by="Price", ascending=value)
+        if weightAscending is not None:
+            if weightAscending == 1:
+                value = True
+
+            else:
+                value = False
+            df = df.sort_values(by="Weight", ascending=value)
+
+        if idAscending is not None:
+            if idAscending == 1:
+                value = True
+            else:
+                value = False
+            df = df.sort_values(by="Id", ascending=value)
+
+        if dateAscending is not None:
+            if dateAscending == 1:
+                value = True
+            else:
+                value = False
+            df = df.sort_values(by="UpdatedAt", ascending=value)
+
+        df = df.drop(["CreatedAt", "EstimatedPrice"], axis=1)
+
+        result = {}
+        for _, row in df.iterrows():
+            category = row["Category"]
+            if category not in result:  # this is for checking duplicate category
+                result[category] = []
+            result[category].append(row.to_dict())
+
+        return JSONResponse(content=jsonable_encoder(result))  # this will convert the result into json format.
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to fetch or process data: {e}")
+
+
+async def parse_necklace_try_on_id(necklaceImageId: str = Form(...),
+                                   necklaceCategory: str = Form(...),
+                                   storename: str = Form(...),
+                                   api_token: str = Form(...)) -> NecklaceTryOnIDEntity:
+    return NecklaceTryOnIDEntity(
+        necklaceImageId=necklaceImageId,
+        necklaceCategory=necklaceCategory,
+        storename=storename,
+        api_token=api_token
+    )
+
+
+@nto_cto_router.post("/necklaceTryOnID")
+async def necklace_try_on_id(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
+                             image: UploadFile = File(...)):
+    data, _ = supabase.table("APIKeyList").select("*").filter("API_KEY", "eq",
+                                                              necklace_try_on_id.api_token).execute()
+
+    api_key_actual = data[1][0]['API_KEY']
+    if api_key_actual != necklace_try_on_id.api_token:
+        return JSONResponse(content={"error": "Invalid API Key"}, status_code=401)
+
+    else:
+        imageBytes = await image.read()
+
+        jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
+
+        try:
+            image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
+
+        except:
+            error_message = {
+                "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
+            }
+
+            return JSONResponse(content=error_message, status_code=404)
+
+        result, headetText, mask = await pipeline.necklaceTryOn_(image=image, jewellery=jewellery,
+                                                                 storename=necklace_try_on_id.storename)
+
+        inMemFile = BytesIO()
+        inMemFileMask = BytesIO()
+        result.save(inMemFile, format="WEBP", quality=85)
+        mask.save(inMemFileMask, format="WEBP", quality=85)
+        outputBytes = inMemFile.getvalue()
+        maskBytes = inMemFileMask.getvalue()
+        response = {
+            "output": f"data:image/WEBP;base64,{base64.b64encode(outputBytes).decode('utf-8')}",
+            "mask": f"data:image/WEBP;base64,{base64.b64encode(maskBytes).decode('utf-8')}"
+        }
+        creditResponse = deductAndTrackCredit(storename=necklace_try_on_id.storename, endpoint="/necklaceTryOnID")
+        if creditResponse == "No Credits Available":
+            response = {
+                "error": "No Credits Remaining"
+            }
+
+            return JSONResponse(content=response)
+
+        else:
+            return JSONResponse(content=response)
+
+
+@nto_cto_router.post("/canvasPoints")
+async def canvas_points(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
+                        image: UploadFile = File(...)):
+    imageBytes = await image.read()
+
+    jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
+
+    try:
+        image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
+
+    except:
+        error_message = {
+            "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
+        }
+
+        return JSONResponse(content=error_message, status_code=404)
+
+    response = await pipeline.canvasPoint(
+        image=image, jewellery=jewellery, storename=necklace_try_on_id.storename
+    )
+
+    creditResponse = deductAndTrackCredit(storename=necklace_try_on_id.storename, endpoint="/necklaceTryOnID")
+    if creditResponse == "No Credits Available":
+        response = {
+            "error": "No Credits Remaining"
+        }
+
+        return JSONResponse(content=response)
+
+    else:
+        return JSONResponse(content=response)
+
+
+@nto_cto_router.post("/necklaceTryOnWithPoints")
+async def necklace_try_on_with_points(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(parse_necklace_try_on_id),
+                                      image: UploadFile = File(...),
+                                      left_x: int = Form(...),
+                                      left_y: int = Form(...),
+                                      right_x: int = Form(...),
+                                      right_y: int = Form(...)):
+    imageBytes = await image.read()
+
+    jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{necklace_try_on_id.storename}/{necklace_try_on_id.necklaceCategory}/image/{necklace_try_on_id.necklaceImageId}.png"
+
+    try:
+        image, jewellery = Image.open(BytesIO(imageBytes)), Image.open(returnBytesData(url=jewellery_url))
+
+    except:
+        error_message = {
+            "error": "The requested resource (Image, necklace category, or store) is not available. Please verify the availability and try again."
+        }
+
+        return JSONResponse(content=error_message, status_code=404)
+
+    result, headerText, mask = await pipeline.necklaceTryOnWithPoints_(
+        image=image, jewellery=jewellery, left_shoulder=(left_x, left_y), right_shoulder=(right_x, right_y),
+        storename=necklace_try_on_id.storename
+    )
+
+    inMemFile = BytesIO()
+    inMemFileMask = BytesIO()
+    result.save(inMemFile, format="WEBP", quality=85)
+    mask.save(inMemFileMask, format="WEBP", quality=85)
+    outputBytes = inMemFile.getvalue()
+    maskBytes = inMemFileMask.getvalue()
+    response = {
+        "output": f"data:image/WEBP;base64,{base64.b64encode(outputBytes).decode('utf-8')}",
+        "mask": f"data:image/WEBP;base64,{base64.b64encode(maskBytes).decode('utf-8')}"
+    }
+
+    creditResponse = deductAndTrackCredit(storename=necklace_try_on_id.storename, endpoint="/necklaceTryOnID")
+    if creditResponse == "No Credits Available":
+        response = {
+            "error": "No Credits Remaining"
+        }
+
+        return JSONResponse(content=response)
+
+    else:
+        return JSONResponse(content=response)

src/components/auto_crop.py

@@ -0,0 +1,61 @@
+"""
+project @ NTO-TCP-HF
+created @ 2024-10-28
+author  @ github.com/ishworrsubedii
+"""
+from io import BytesIO
+
+from PIL import Image
+
+
+def crop_transparent_image(image_data: bytes) -> tuple[bytes, dict]:
+    try:
+        image = Image.open(BytesIO(image_data))
+
+        if image.format != 'PNG':
+            raise ValueError("Only PNG images are supported")
+
+        width = image.size[0]
+        height = image.size[1]
+        pixels = image.load()
+
+        top = height
+        bottom = 0
+        left = width
+        right = 0
+
+        # Find boundaries of non-transparent pixels
+        for y in range(height):
+            for x in range(width):
+                pixel = pixels[x, y]
+                if isinstance(pixel, tuple) and len(pixel) == 4:
+                    if pixel[3] != 0:
+                        left = min(left, x)
+                        top = min(top, y)
+                        right = max(right, x)
+                        bottom = max(bottom, y)
+
+        left = max(0, left)
+        top = max(0, top)
+        right = min(width, right + 1)
+        bottom = min(height, bottom + 1)
+
+        if left >= right or top >= bottom:
+            left, top, right, bottom = 0, 0, width, height
+
+        # Crop image
+        cropped_image = image.crop((left, top, right, bottom))
+
+        output_buffer = BytesIO()
+        cropped_image.save(output_buffer, format='PNG')
+        output_buffer.seek(0)
+
+        metadata = {
+            "original_size": f"{width}x{height}",
+            "cropped_size": f"{cropped_image.width}x{cropped_image.height}"
+        }
+
+        return output_buffer.getvalue(), metadata
+
+    except Exception as e:
+        raise ValueError(f"Error processing image: {str(e)}")

src/pipelines/completePipeline.py

@@ -28,3 +28,4 @@ class Pipeline:
                                                                                  right_point=right_shoulder,
                                                                                  storename=storename)
         return [result, headerText, mask]
+