Initial commit

.github/workflows/dockerhub.yaml

@@ -0,0 +1,39 @@
+name: Publish Docker image
+
+on:
+  push:
+    branches: [main]
+
+jobs:
+  push_to_registry:
+    name: Push Docker image to Docker Hub
+    runs-on: ubuntu-latest
+    permissions:
+      packages: write
+      contents: read
+      attestations: write
+    steps:
+      - name: Check out the repo
+        uses: actions/checkout@v4
+
+      - name: Log in to Docker Hub
+        uses: docker/login-action@f4ef78c080cd8ba55a85445d5b36e214a81df20a
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@9ec57ed1fcdbf14dcef7dfbe97b2010124a938b7
+        with:
+          images: ishworrsubedii/web_plugin_api
+
+      - name: Build and push Docker image
+        id: push
+        uses: docker/build-push-action@3b5e8027fcad23fda98b2e3ac259d8d67585f671
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

.gitignore

@@ -0,0 +1,7 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+.idea/
+
+*.log

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.10-slim
+
+WORKDIR /api
+
+COPY . /api
+
+RUN apt-get update && apt-get install -y
+
+RUN apt install libgl1-mesa-glx -y
+
+RUN apt-get install 'ffmpeg'\
+    'libsm6'\
+    'libxext6'  -y
+
+RUN pip install -r requirements.txt
+
+RUN ulimit -s 2000
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app:app", "--host","0.0.0.0","--port","8000"]

app.py

@@ -0,0 +1,224 @@
+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
+
+pipeline = Pipeline()
+app = FastAPI(title="Magical Mirror Web Plugin")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+url: str = os.environ["SUPABASE_URL"]
+key: str = os.environ["SUPABASE_KEY"]
+supabase: Client = create_client(url, key)
+
+
+@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}")
+
+
+class NecklaceTryOnIDEntity(BaseModel):
+    necklaceImageId: str
+    necklaceCategory: str
+    storename: str
+    api_token: str
+
+
+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 = await pipeline.necklaceTryOn(image=image, jewellery=jewellery,
+                                                          storename=necklace_try_on_id.storename)
+
+        inMemFile = BytesIO()
+        result.save(inMemFile, format="WEBP", quality=85)
+        outputBytes = inMemFile.getvalue()
+        response = {
+            "output": f"data:image/WEBP;base64,{base64.b64encode(outputBytes).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)

requirements.txt

@@ -0,0 +1,80 @@
+absl-py==2.1.0
+annotated-types==0.7.0
+anyio==4.4.0
+attrs==23.2.0
+certifi==2024.7.4
+cffi==1.16.0
+charset-normalizer==3.3.2
+click==8.1.7
+contourpy==1.2.1
+cvzone==1.6.1
+cycler==0.12.1
+deprecation==2.1.0
+dnspython==2.6.1
+email_validator==2.2.0
+exceptiongroup==1.2.1
+fastapi==0.111.0
+fastapi-cli==0.0.4
+flatbuffers==24.3.25
+fonttools==4.53.1
+gotrue==2.5.4
+h11==0.14.0
+httpcore==1.0.5
+httptools==0.6.1
+httpx==0.27.0
+idna==3.7
+jax==0.4.30
+jaxlib==0.4.30
+Jinja2==3.1.4
+kiwisolver==1.4.5
+markdown-it-py==3.0.0
+MarkupSafe==2.1.5
+matplotlib==3.9.1
+mdurl==0.1.2
+mediapipe==0.10.14
+ml-dtypes==0.4.0
+numpy==2.0.0
+opencv-contrib-python==4.10.0.84
+opencv-python==4.10.0.84
+opt-einsum==3.3.0
+orjson==3.10.6
+packaging==24.1
+pandas==2.2.2
+pillow==10.4.0
+postgrest==0.16.8
+protobuf==4.25.3
+pycparser==2.22
+pydantic==2.8.2
+pydantic_core==2.20.1
+Pygments==2.18.0
+pyparsing==3.1.2
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+python-multipart==0.0.9
+pytz==2024.1
+PyYAML==6.0.1
+realtime==1.0.6
+requests==2.32.3
+rich==13.7.1
+scikit-build==0.18.0
+scipy==1.14.0
+shellingham==1.5.4
+six==1.16.0
+sniffio==1.3.1
+sounddevice==0.4.7
+starlette==0.37.2
+storage3==0.7.6
+StrEnum==0.4.15
+supabase==2.5.1
+supafunc==0.4.6
+tomli==2.0.1
+typer==0.12.3
+typing_extensions==4.12.2
+tzdata==2024.1
+ujson==5.10.0
+urllib3==2.2.2
+uvicorn==0.30.1
+uvloop==0.19.0
+watchfiles==0.22.0
+websockets==12.0
+-e .

setup.py

@@ -0,0 +1,17 @@
+from setuptools import setup, find_packages
+
+HYPER_E_DOT = "-e ."
+def getRequirements(requirementsPath: str) -> list[str]:
+    with open(requirementsPath) as file:
+        requirements = file.read().split("\n")
+    requirements.remove(HYPER_E_DOT)
+    return requirements
+
+setup(
+    name = "Magical-Mirror",
+    author = "Subramani Sivakumar",
+    author_email = "bwsubbu@gmail.com",
+    version = "0.1",
+    packages = find_packages(),
+    install_requires = getRequirements(requirementsPath = "./requirements.txt")
+)

src/components/necklaceTryOn.py

@@ -0,0 +1,235 @@
+from cvzone.FaceMeshModule import FaceMeshDetector
+from src.utils import addWatermark, returnBytesData
+from src.utils.exceptions import CustomException
+from cvzone.PoseModule import PoseDetector
+from src.utils.logger import logger
+from dataclasses import dataclass
+from typing import Union
+from PIL import Image
+import numpy as np
+import cvzone
+import math
+import cv2
+import gc
+
+
+@dataclass
+class NecklaceTryOnConfig:
+    logoURL: str = "https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/MagicMirror/FullImages/{}.png"
+
+
+class NecklaceTryOn:
+    def __init__(self) -> None:
+        self.detector = PoseDetector()
+        self.necklaceTryOnConfig = NecklaceTryOnConfig()
+        self.meshDetector = FaceMeshDetector(staticMode=True, maxFaces=1)
+
+    def necklaceTryOn(self, image: Image.Image, jewellery: Image.Image, storename: str) -> list[
+        Union[Image.Image, str]]:
+        try:
+            logger.info(f">>> NECKLACE TRY ON STARTED :: {storename} <<<")
+
+            # reading the images
+            image, jewellery = image.convert("RGB").resize((3000, 3000)), jewellery.convert("RGBA")
+            image = np.array(image)
+            copy_image = image.copy()
+            jewellery = np.array(jewellery)
+
+            logger.info(f"NECKLACE TRY ON :: detecting pose and landmarks :: {storename}")
+
+            image = self.detector.findPose(image)
+            lmList, _ = self.detector.findPosition(image, bboxWithHands=False, draw=False)
+
+            img, faces = self.meshDetector.findFaceMesh(image, draw=False)
+            leftLandmarkIndex = 172
+            rightLandmarkIndex = 397
+
+            leftLandmark, rightLandmark = faces[0][leftLandmarkIndex], faces[0][rightLandmarkIndex]
+            landmarksDistance = int(
+                ((leftLandmark[0] - rightLandmark[0]) ** 2 + (leftLandmark[1] - rightLandmark[1]) ** 2) ** 0.5)
+
+            logger.info(f"NECKLACE TRY ON :: estimating neck points :: {storename}")
+
+            # avg_x1 = int(leftLandmark[0] - landmarksDistance * 0.15) -> V2.1
+            avg_x1 = int(leftLandmark[0] - landmarksDistance * 0.12)
+            avg_x2 = int(rightLandmark[0] + landmarksDistance * 0.12)
+
+            avg_y1 = int(leftLandmark[1] + landmarksDistance * 0.5)
+            avg_y2 = int(rightLandmark[1] + landmarksDistance * 0.5)
+
+            logger.info(f"NECKLACE TRY ON :: scaling the necklace image :: {storename}")
+
+            if avg_y2 < avg_y1:
+                angle = math.ceil(
+                    self.detector.findAngle(
+                        p1=(avg_x2, avg_y2), p2=(avg_x1, avg_y1), p3=(avg_x2, avg_y1)
+                    )[0]
+                )
+            else:
+                angle = math.ceil(
+                    self.detector.findAngle(
+                        p1=(avg_x2, avg_y2), p2=(avg_x1, avg_y1), p3=(avg_x2, avg_y1)
+                    )[0]
+                )
+                angle = angle * -1
+
+            xdist = avg_x2 - avg_x1
+            origImgRatio = xdist / jewellery.shape[1]
+            ydist = jewellery.shape[0] * origImgRatio
+
+            logger.info(f"NECKLACE TRY ON :: adding offset based on the necklace shape :: {storename}")
+
+            image_gray = cv2.cvtColor(jewellery, cv2.COLOR_BGRA2GRAY)
+            for offset_orig in range(image_gray.shape[1]):
+                pixel_value = image_gray[0, :][offset_orig]
+                if (pixel_value != 255) & (pixel_value != 0):
+                    break
+                else:
+                    continue
+            offset = int(0.8 * xdist * (offset_orig / jewellery.shape[1]))
+            jewellery = cv2.resize(
+                jewellery, (int(xdist), int(ydist)), interpolation=cv2.INTER_AREA
+            )
+            jewellery = cvzone.rotateImage(jewellery, angle)
+            y_coordinate = avg_y1 - offset
+            available_space = copy_image.shape[0] - y_coordinate
+            extra = jewellery.shape[0] - available_space
+
+            logger.info(f"NECKLACE TRY ON :: generating necklace placement status :: {storename}")
+
+            if extra > 0:
+                headerText = "To see more of the necklace, please step back slightly."
+            else:
+                headerText = "success"
+
+            logger.info(f"NECKLACE TRY ON :: generating output :: {storename}")
+
+            result = cvzone.overlayPNG(copy_image, jewellery, (avg_x1, y_coordinate))
+            image = Image.fromarray(result.astype(np.uint8))
+            logo = Image.open(returnBytesData(url=self.necklaceTryOnConfig.logoURL.format(storename)))
+            result = addWatermark(background=image, logo=logo)
+
+            gc.collect()
+
+            return [result, headerText]
+
+        except Exception as e:
+            logger.error(f"{CustomException(e)}:: {storename}")
+            raise CustomException(e)
+
+    def necklaceTryOnV3(self, image: Image.Image, jewellery: Image.Image, storename: str) -> list[
+        Union[Image.Image, str]]:
+        try:
+            logger.info(f">>> NECKLACE TRY ON STARTED :: {storename} <<<")
+
+            # reading the images
+            image, jewellery = image.convert("RGB"), jewellery.convert("RGBA")
+            image = np.array(image.resize((4000, 4000)))
+            copy_image = image.copy()
+            jewellery = np.array(jewellery)
+
+            logger.info(f"NECKLACE TRY ON :: detecting pose and landmarks :: {storename}")
+
+            image = self.detector.findPose(image)
+            lmList, _ = self.detector.findPosition(image, bboxWithHands=False, draw=False)
+            meshDetector = FaceMeshDetector(staticMode=True, maxFaces=1)
+            img, faces = meshDetector.findFaceMesh(image, draw=False)
+            left_lip_point = faces[0][61]
+            right_lip_point = faces[0][291]
+            pt12, pt11, pt10, pt9 = (
+                lmList[12][:2],
+                lmList[11][:2],
+                lmList[10][:2],
+                lmList[9][:2],
+            )
+
+            mid_lips = (
+                int((left_lip_point[0] + right_lip_point[0]) / 2), int((left_lip_point[1] + right_lip_point[1]) / 2))
+
+            mid_lips_x1 = int(pt12[0] + (mid_lips[0] - pt12[0]) / 2)
+            mid_lips_y1 = int(pt12[1] + (mid_lips[1] - pt12[1]) / 2)
+
+            mid_lips_x2 = int(pt11[0] + (mid_lips[0] - pt11[0]) / 2)
+            mid_lips_y2 = int(pt11[1] + (mid_lips[1] - pt11[1]) / 2)
+
+            # left right lip
+            left_right_lip_org_x11 = int(pt12[0] + (right_lip_point[0] - pt12[0]) / 2)
+            left_right_lip_org_y11 = int(pt12[1] + (right_lip_point[1] - pt12[1]) / 2)
+
+            left_right_lip_org_x12 = int(pt11[0] + (left_lip_point[0] - pt11[0]) / 2)
+            left_right_lip_org_y12 = int(pt11[1] + (left_lip_point[1] - pt11[1]) / 2)
+
+            # left right lip 2
+            left_right_lip_org_x21 = int(pt12[0] + (left_lip_point[0] - pt12[0]) / 2)
+            left_right_lip_org_y21 = int(pt12[1] + (left_lip_point[1] - pt12[1]) / 2)
+
+            left_right_lip_org_x22 = int(pt11[0] + (right_lip_point[0] - pt11[0]) / 2)
+            left_right_lip_org_y22 = int(pt11[1] + (right_lip_point[1] - pt11[1]) / 2)
+
+            logger.info(f"NECKLACE TRY ON :: estimating neck points :: {storename}")
+
+            avg_x1 = int((mid_lips_x1 + left_right_lip_org_x11 + left_right_lip_org_x21) / 3)
+            avg_y1 = int((mid_lips_y1 + left_right_lip_org_y11 + left_right_lip_org_y21) / 3)
+
+            avg_x2 = int((mid_lips_x2 + left_right_lip_org_x12 + left_right_lip_org_x22) / 3)
+            avg_y2 = int((mid_lips_y2 + left_right_lip_org_y12 + left_right_lip_org_y22) / 3)
+
+            logger.info(f"NECKLACE TRY ON :: scaling the necklace image :: {storename}")
+
+            if avg_y2 < avg_y1:
+                angle = math.ceil(
+                    self.detector.findAngle(
+                        p1=(avg_x2, avg_y2), p2=(avg_x1, avg_y1), p3=(avg_x2, avg_y1)
+                    )[0]
+                )
+            else:
+                angle = math.ceil(
+                    self.detector.findAngle(
+                        p1=(avg_x2, avg_y2), p2=(avg_x1, avg_y1), p3=(avg_x2, avg_y1)
+                    )[0]
+                )
+                angle = angle * -1
+
+            xdist = avg_x2 - avg_x1
+            origImgRatio = xdist / jewellery.shape[1]
+            ydist = jewellery.shape[0] * origImgRatio
+
+            logger.info(f"NECKLACE TRY ON :: adding offset based on the necklace shape :: {storename}")
+
+            image_gray = cv2.cvtColor(jewellery, cv2.COLOR_BGRA2GRAY)
+            for offset_orig in range(image_gray.shape[1]):
+                pixel_value = image_gray[0, :][offset_orig]
+                if (pixel_value != 255) & (pixel_value != 0):
+                    break
+                else:
+                    continue
+            offset = int(0.8 * xdist * (offset_orig / jewellery.shape[1]))
+            jewellery = cv2.resize(
+                jewellery, (int(xdist), int(ydist)), interpolation=cv2.INTER_AREA
+            )
+            jewellery = cvzone.rotateImage(jewellery, angle)
+            y_coordinate = avg_y1 - offset
+            available_space = copy_image.shape[0] - y_coordinate
+            extra = jewellery.shape[0] - available_space
+
+            logger.info(f"NECKLACE TRY ON :: generating necklace placement status :: {storename}")
+
+            if extra > 0:
+                headerText = "To see more of the necklace, please step back slightly."
+            else:
+                headerText = "success"
+
+            logger.info(f"NECKLACE TRY ON :: generating output :: {storename}")
+
+            result = cvzone.overlayPNG(copy_image, jewellery, (avg_x1, y_coordinate))
+            image = Image.fromarray(result.astype(np.uint8))
+            logo = Image.open(returnBytesData(url=self.necklaceTryOnConfig.logoURL.format({storename})))
+            result = addWatermark(background=image, logo=logo)
+
+            gc.collect()
+
+            return [result, headerText]
+
+        except Exception as e:
+            logger.error(f"{CustomException(e)}:: {storename}")
+            raise CustomException(e)

src/pipelines/completePipeline.py

@@ -0,0 +1,13 @@
+from src.components.necklaceTryOn import NecklaceTryOn
+from typing import Union
+from PIL import Image
+
+
+class Pipeline:
+    def __init__(self) -> None:
+        self.necklaceTryOnObj = NecklaceTryOn()
+
+    async def necklaceTryOn(self, image: Image.Image, jewellery: Image.Image, storename: str) -> list[
+        Union[Image.Image, str]]:
+        result, headerText = self.necklaceTryOnObj.necklaceTryOn(image=image, jewellery=jewellery, storename=storename)
+        return [result, headerText]

src/utils/__init__.py

@@ -0,0 +1,45 @@
+from supabase import create_client, Client
+from PIL import Image
+from io import BytesIO
+import requests
+import os
+
+
+# function to add watermark to images
+def addWatermark(background: Image.Image, logo: Image.Image) -> Image.Image:
+    background = background.convert("RGBA")
+    logo = logo.convert("RGBA").resize((int(0.08 * background.size[0]), int(0.08 * background.size[0])))
+    background.paste(logo, (10, background.size[1] - logo.size[1] - 10), logo)
+    return background
+
+
+# function to download an image from url and return as bytes objects
+def returnBytesData(url: str) -> BytesIO:
+    response = requests.get(url)
+    return BytesIO(response.content)
+
+
+# function to get public URLs of paths
+def supabaseGetPublicURL(path: str) -> str:
+    url_string = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{path}"
+    return url_string.replace(" ", "%20")
+
+
+# function to deduct credit
+def deductAndTrackCredit(storename: str, endpoint: str) -> str:
+    url: str = os.environ["SUPABASE_URL"]
+    key: str = os.environ["SUPABASE_KEY"]
+    supabase: Client = create_client(url, key)
+    current, _ = supabase.table('ClientConfig').select('CreditBalance').eq("StoreName", f"{storename}").execute()
+    if current[1] == []:
+        return "Not Found"
+    else:
+        current = current[1][0]["CreditBalance"]
+        if current > 0:
+            data, _ = supabase.table('ClientConfig').update({'CreditBalance': current - 1}).eq("StoreName",
+                                                                                               f"{storename}").execute()
+            data, _ = supabase.table('UsageHistory').insert(
+                {'StoreName': f"{storename}", 'APIEndpoint': f"{endpoint}"}).execute()
+            return "Success"
+        else:
+            return "No Credits Available"

src/utils/backgroundEnhancerArchitecture.py

@@ -0,0 +1,454 @@
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from huggingface_hub import PyTorchModelHubMixin
+
+
+class REBNCONV(nn.Module):
+    def __init__(self, in_ch=3, out_ch=3, dirate=1, stride=1):
+        super(REBNCONV, self).__init__()
+
+        self.conv_s1 = nn.Conv2d(in_ch, out_ch, 3, padding=1 * dirate, dilation=1 * dirate, stride=stride)
+        self.bn_s1 = nn.BatchNorm2d(out_ch)
+        self.relu_s1 = nn.ReLU(inplace=True)
+
+    def forward(self, x):
+        hx = x
+        xout = self.relu_s1(self.bn_s1(self.conv_s1(hx)))
+
+        return xout
+
+
+## upsample tensor 'src' to have the same spatial size with tensor 'tar'
+def _upsample_like(src, tar):
+    src = F.interpolate(src, size=tar.shape[2:], mode='bilinear')
+
+    return src
+
+
+### RSU-7 ###
+class RSU7(nn.Module):
+
+    def __init__(self, in_ch=3, mid_ch=12, out_ch=3, img_size=512):
+        super(RSU7, self).__init__()
+
+        self.in_ch = in_ch
+        self.mid_ch = mid_ch
+        self.out_ch = out_ch
+
+        self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1)  ## 1 -> 1/2
+
+        self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1)
+        self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool3 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool4 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv5 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool5 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv6 = REBNCONV(mid_ch, mid_ch, dirate=1)
+
+        self.rebnconv7 = REBNCONV(mid_ch, mid_ch, dirate=2)
+
+        self.rebnconv6d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv5d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv4d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1)
+
+    def forward(self, x):
+        b, c, h, w = x.shape
+
+        hx = x
+        hxin = self.rebnconvin(hx)
+
+        hx1 = self.rebnconv1(hxin)
+        hx = self.pool1(hx1)
+
+        hx2 = self.rebnconv2(hx)
+        hx = self.pool2(hx2)
+
+        hx3 = self.rebnconv3(hx)
+        hx = self.pool3(hx3)
+
+        hx4 = self.rebnconv4(hx)
+        hx = self.pool4(hx4)
+
+        hx5 = self.rebnconv5(hx)
+        hx = self.pool5(hx5)
+
+        hx6 = self.rebnconv6(hx)
+
+        hx7 = self.rebnconv7(hx6)
+
+        hx6d = self.rebnconv6d(torch.cat((hx7, hx6), 1))
+        hx6dup = _upsample_like(hx6d, hx5)
+
+        hx5d = self.rebnconv5d(torch.cat((hx6dup, hx5), 1))
+        hx5dup = _upsample_like(hx5d, hx4)
+
+        hx4d = self.rebnconv4d(torch.cat((hx5dup, hx4), 1))
+        hx4dup = _upsample_like(hx4d, hx3)
+
+        hx3d = self.rebnconv3d(torch.cat((hx4dup, hx3), 1))
+        hx3dup = _upsample_like(hx3d, hx2)
+
+        hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1))
+        hx2dup = _upsample_like(hx2d, hx1)
+
+        hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1))
+
+        return hx1d + hxin
+
+
+### RSU-6 ###
+class RSU6(nn.Module):
+
+    def __init__(self, in_ch=3, mid_ch=12, out_ch=3):
+        super(RSU6, self).__init__()
+
+        self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1)
+
+        self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1)
+        self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool3 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool4 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv5 = REBNCONV(mid_ch, mid_ch, dirate=1)
+
+        self.rebnconv6 = REBNCONV(mid_ch, mid_ch, dirate=2)
+
+        self.rebnconv5d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv4d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1)
+
+    def forward(self, x):
+        hx = x
+
+        hxin = self.rebnconvin(hx)
+
+        hx1 = self.rebnconv1(hxin)
+        hx = self.pool1(hx1)
+
+        hx2 = self.rebnconv2(hx)
+        hx = self.pool2(hx2)
+
+        hx3 = self.rebnconv3(hx)
+        hx = self.pool3(hx3)
+
+        hx4 = self.rebnconv4(hx)
+        hx = self.pool4(hx4)
+
+        hx5 = self.rebnconv5(hx)
+
+        hx6 = self.rebnconv6(hx5)
+
+        hx5d = self.rebnconv5d(torch.cat((hx6, hx5), 1))
+        hx5dup = _upsample_like(hx5d, hx4)
+
+        hx4d = self.rebnconv4d(torch.cat((hx5dup, hx4), 1))
+        hx4dup = _upsample_like(hx4d, hx3)
+
+        hx3d = self.rebnconv3d(torch.cat((hx4dup, hx3), 1))
+        hx3dup = _upsample_like(hx3d, hx2)
+
+        hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1))
+        hx2dup = _upsample_like(hx2d, hx1)
+
+        hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1))
+
+        return hx1d + hxin
+
+
+### RSU-5 ###
+class RSU5(nn.Module):
+
+    def __init__(self, in_ch=3, mid_ch=12, out_ch=3):
+        super(RSU5, self).__init__()
+
+        self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1)
+
+        self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1)
+        self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool3 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=1)
+
+        self.rebnconv5 = REBNCONV(mid_ch, mid_ch, dirate=2)
+
+        self.rebnconv4d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1)
+
+    def forward(self, x):
+        hx = x
+
+        hxin = self.rebnconvin(hx)
+
+        hx1 = self.rebnconv1(hxin)
+        hx = self.pool1(hx1)
+
+        hx2 = self.rebnconv2(hx)
+        hx = self.pool2(hx2)
+
+        hx3 = self.rebnconv3(hx)
+        hx = self.pool3(hx3)
+
+        hx4 = self.rebnconv4(hx)
+
+        hx5 = self.rebnconv5(hx4)
+
+        hx4d = self.rebnconv4d(torch.cat((hx5, hx4), 1))
+        hx4dup = _upsample_like(hx4d, hx3)
+
+        hx3d = self.rebnconv3d(torch.cat((hx4dup, hx3), 1))
+        hx3dup = _upsample_like(hx3d, hx2)
+
+        hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1))
+        hx2dup = _upsample_like(hx2d, hx1)
+
+        hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1))
+
+        return hx1d + hxin
+
+
+### RSU-4 ###
+class RSU4(nn.Module):
+
+    def __init__(self, in_ch=3, mid_ch=12, out_ch=3):
+        super(RSU4, self).__init__()
+
+        self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1)
+
+        self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1)
+        self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1)
+        self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1)
+
+        self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=2)
+
+        self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1)
+        self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1)
+
+    def forward(self, x):
+        hx = x
+
+        hxin = self.rebnconvin(hx)
+
+        hx1 = self.rebnconv1(hxin)
+        hx = self.pool1(hx1)
+
+        hx2 = self.rebnconv2(hx)
+        hx = self.pool2(hx2)
+
+        hx3 = self.rebnconv3(hx)
+
+        hx4 = self.rebnconv4(hx3)
+
+        hx3d = self.rebnconv3d(torch.cat((hx4, hx3), 1))
+        hx3dup = _upsample_like(hx3d, hx2)
+
+        hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1))
+        hx2dup = _upsample_like(hx2d, hx1)
+
+        hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1))
+
+        return hx1d + hxin
+
+
+### RSU-4F ###
+class RSU4F(nn.Module):
+
+    def __init__(self, in_ch=3, mid_ch=12, out_ch=3):
+        super(RSU4F, self).__init__()
+
+        self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1)
+
+        self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1)
+        self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=2)
+        self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=4)
+
+        self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=8)
+
+        self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=4)
+        self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=2)
+        self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1)
+
+    def forward(self, x):
+        hx = x
+
+        hxin = self.rebnconvin(hx)
+
+        hx1 = self.rebnconv1(hxin)
+        hx2 = self.rebnconv2(hx1)
+        hx3 = self.rebnconv3(hx2)
+
+        hx4 = self.rebnconv4(hx3)
+
+        hx3d = self.rebnconv3d(torch.cat((hx4, hx3), 1))
+        hx2d = self.rebnconv2d(torch.cat((hx3d, hx2), 1))
+        hx1d = self.rebnconv1d(torch.cat((hx2d, hx1), 1))
+
+        return hx1d + hxin
+
+
+class myrebnconv(nn.Module):
+    def __init__(self, in_ch=3,
+                 out_ch=1,
+                 kernel_size=3,
+                 stride=1,
+                 padding=1,
+                 dilation=1,
+                 groups=1):
+        super(myrebnconv, self).__init__()
+
+        self.conv = nn.Conv2d(in_ch,
+                              out_ch,
+                              kernel_size=kernel_size,
+                              stride=stride,
+                              padding=padding,
+                              dilation=dilation,
+                              groups=groups)
+        self.bn = nn.BatchNorm2d(out_ch)
+        self.rl = nn.ReLU(inplace=True)
+
+    def forward(self, x):
+        return self.rl(self.bn(self.conv(x)))
+
+
+class BackgroundEnhancerArchitecture(nn.Module, PyTorchModelHubMixin):
+
+    def __init__(self, config: dict = {"in_ch": 3, "out_ch": 1}):
+        super(BackgroundEnhancerArchitecture, self).__init__()
+        in_ch = config["in_ch"]
+        out_ch = config["out_ch"]
+        self.conv_in = nn.Conv2d(in_ch, 64, 3, stride=2, padding=1)
+        self.pool_in = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.stage1 = RSU7(64, 32, 64)
+        self.pool12 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.stage2 = RSU6(64, 32, 128)
+        self.pool23 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.stage3 = RSU5(128, 64, 256)
+        self.pool34 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.stage4 = RSU4(256, 128, 512)
+        self.pool45 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.stage5 = RSU4F(512, 256, 512)
+        self.pool56 = nn.MaxPool2d(2, stride=2, ceil_mode=True)
+
+        self.stage6 = RSU4F(512, 256, 512)
+
+        # decoder
+        self.stage5d = RSU4F(1024, 256, 512)
+        self.stage4d = RSU4(1024, 128, 256)
+        self.stage3d = RSU5(512, 64, 128)
+        self.stage2d = RSU6(256, 32, 64)
+        self.stage1d = RSU7(128, 16, 64)
+
+        self.side1 = nn.Conv2d(64, out_ch, 3, padding=1)
+        self.side2 = nn.Conv2d(64, out_ch, 3, padding=1)
+        self.side3 = nn.Conv2d(128, out_ch, 3, padding=1)
+        self.side4 = nn.Conv2d(256, out_ch, 3, padding=1)
+        self.side5 = nn.Conv2d(512, out_ch, 3, padding=1)
+        self.side6 = nn.Conv2d(512, out_ch, 3, padding=1)
+
+        # self.outconv = nn.Conv2d(6*out_ch,out_ch,1)
+
+    def forward(self, x):
+        hx = x
+
+        hxin = self.conv_in(hx)
+        # hx = self.pool_in(hxin)
+
+        # stage 1
+        hx1 = self.stage1(hxin)
+        hx = self.pool12(hx1)
+
+        # stage 2
+        hx2 = self.stage2(hx)
+        hx = self.pool23(hx2)
+
+        # stage 3
+        hx3 = self.stage3(hx)
+        hx = self.pool34(hx3)
+
+        # stage 4
+        hx4 = self.stage4(hx)
+        hx = self.pool45(hx4)
+
+        # stage 5
+        hx5 = self.stage5(hx)
+        hx = self.pool56(hx5)
+
+        # stage 6
+        hx6 = self.stage6(hx)
+        hx6up = _upsample_like(hx6, hx5)
+
+        # -------------------- decoder --------------------
+        hx5d = self.stage5d(torch.cat((hx6up, hx5), 1))
+        hx5dup = _upsample_like(hx5d, hx4)
+
+        hx4d = self.stage4d(torch.cat((hx5dup, hx4), 1))
+        hx4dup = _upsample_like(hx4d, hx3)
+
+        hx3d = self.stage3d(torch.cat((hx4dup, hx3), 1))
+        hx3dup = _upsample_like(hx3d, hx2)
+
+        hx2d = self.stage2d(torch.cat((hx3dup, hx2), 1))
+        hx2dup = _upsample_like(hx2d, hx1)
+
+        hx1d = self.stage1d(torch.cat((hx2dup, hx1), 1))
+
+        # side output
+        d1 = self.side1(hx1d)
+        d1 = _upsample_like(d1, x)
+
+        d2 = self.side2(hx2d)
+        d2 = _upsample_like(d2, x)
+
+        d3 = self.side3(hx3d)
+        d3 = _upsample_like(d3, x)
+
+        d4 = self.side4(hx4d)
+        d4 = _upsample_like(d4, x)
+
+        d5 = self.side5(hx5d)
+        d5 = _upsample_like(d5, x)
+
+        d6 = self.side6(hx6)
+        d6 = _upsample_like(d6, x)
+
+        return [F.sigmoid(d1), F.sigmoid(d2), F.sigmoid(d3), F.sigmoid(d4), F.sigmoid(d5), F.sigmoid(d6)], [hx1d, hx2d,
+                                                                                                            hx3d, hx4d,
+                                                                                                            hx5d, hx6]

src/utils/exceptions.py

@@ -0,0 +1,16 @@
+import sys
+
+def error_message_detail(error):
+    _, _, exc_info = sys.exc_info()
+    filename = exc_info.tb_frame.f_code.co_filename
+    lineno = exc_info.tb_lineno
+    error_message = "Error encountered in line no [{}], filename : [{}], saying [{}]".format(lineno, filename, error)
+    return error_message
+
+class CustomException(Exception):
+    def __init__(self, error_message):
+        super().__init__(error_message)
+        self.error_message = error_message_detail(error_message)
+
+    def __str__(self) -> str:
+        return self.error_message

src/utils/logger.py

@@ -0,0 +1,22 @@
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+log_dir = os.path.join(os.getcwd(), "logs")
+os.makedirs(log_dir, exist_ok = True)
+
+LOG_FILE = os.path.join(log_dir, "running_logs.log")
+
+logFormat = "[%(asctime)s: %(levelname)s: %(module)s: %(message)s]"
+logFormatter = logging.Formatter(fmt = logFormat, style = "%")
+
+streamHandler = logging.StreamHandler()
+streamHandler.setFormatter(logFormatter)
+
+fileHandler = logging.FileHandler(filename = LOG_FILE)
+fileHandler.setFormatter(logFormatter)
+
+logger.addHandler(streamHandler)
+logger.addHandler(fileHandler)