add: necklace try on points

app.py

@@ -31,7 +31,7 @@ app.add_middleware(
 )
 url: str = os.getenv("SUPABASE_URL")
 key: str = os.getenv("SUPABASE_KEY")
-supabase: Client = create_client(url, key)
+supabase: Client = create_client(supabase_key=key, supabase_url=url)
 bucket = supabase.storage.from_("JewelMirrorOutputs")
 
 
@@ -411,3 +411,52 @@ async def canvas_points(necklace_try_on_id: NecklaceTryOnIDEntity = Depends(pars
 
     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/components/necklaceTryOn.py

@@ -343,3 +343,64 @@ class NecklaceTryOn:
         except Exception as e:
             logger.error(f"{CustomException(e)}:: {storename}")
             raise CustomException(e)
+
+    def necklaceTryOnWithPoints(self, image: Image.Image, jewellery: Image.Image, storename: str,
+                                left_point: tuple[int, int], right_point: tuple[int, int]) -> list[
+        Union[Image.Image, str]]:
+        try:
+            logger.info(f">>> NECKLACE TRY ON WITH POINTS STARTED :: {storename} <<<")
+
+            image = np.array(image.convert("RGB"))
+            # .resize((3000, 3000)))
+            copy_image = image.copy()
+            jewellery = np.array(jewellery.convert("RGBA"))
+
+            logger.info(f"NECKLACE TRY ON :: scaling the necklace image based on given points :: {storename}")
+
+            avg_x1 = left_point[0]
+            avg_x2 = right_point[0]
+            avg_y1 = left_point[1]
+            avg_y2 = right_point[1]
+
+            angle = math.ceil(self.detector.findAngle((avg_x2, avg_y2), (avg_x1, avg_y1), (avg_x2, avg_y1))[0])
+            if avg_y2 >= avg_y1:
+                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)
+            offset = int(0.8 * xdist * (np.argmax(image_gray[0, :] != 255) / 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
+
+            headerText = "To see more of the necklace, please step back slightly." if extra > 0 else "success"
+
+            logger.info(f"NECKLACE TRY ON :: generating output with given points :: {storename}")
+            result = cvzone.overlayPNG(copy_image, jewellery, (avg_x1, y_coordinate))
+            image = Image.fromarray(result.astype(np.uint8))
+
+            if storename not in self.logo_cache:
+                self.logo_cache[storename] = Image.open(
+                    returnBytesData(url=self.necklaceTryOnConfig.logoURL.format(storename)))
+            result = addWatermark(background=image, logo=self.logo_cache[storename])
+
+            # Create binary mask
+            blackedNecklace = np.zeros_like(copy_image)
+            cvzone.overlayPNG(blackedNecklace, jewellery, (avg_x1, y_coordinate))
+            binaryMask = cv2.cvtColor(blackedNecklace.astype(np.uint8), cv2.COLOR_BGR2GRAY)
+            binaryMask = (binaryMask > 5).astype(np.uint8) * 255
+            mask = Image.fromarray(binaryMask).convert("RGB")
+
+            gc.collect()
+            return [result, headerText, mask]
+
+        except Exception as e:
+            logger.error(f"{CustomException(e)}:: {storename}")
+            raise CustomException(e)

src/pipelines/completePipeline.py

@@ -20,3 +20,11 @@ class Pipeline:
     async def canvasPoint(self, image: Image.Image, jewellery: Image.Image, storename: str) -> dict:
         points = self.necklaceTryOnObj.canvasPoints(image=image, jewellery=jewellery, storename=storename)
         return points
+
+    async def necklaceTryOnWithPoints_(self, image: Image.Image, jewellery: Image.Image, left_shoulder: tuple[int, int],
+                                       right_shoulder: tuple[int, int], storename: str) -> list[Union[Image.Image, str]]:
+        result, headerText, mask = self.necklaceTryOnObj.necklaceTryOnWithPoints(image=image, jewellery=jewellery,
+                                                                                 left_point=left_shoulder,
+                                                                                 right_point=right_shoulder,
+                                                                                 storename=storename)
+        return [result, headerText, mask]