add: rt_ntocto combined

app.py

@@ -45,7 +45,7 @@ async def verify_login_token(credentials: HTTPAuthorizationCredentials = Depends
         raise HTTPException(status_code=401, detail="Invalid token")
 
 
-app = FastAPI(dependencies=[Depends(verify_login_token)])
+app = FastAPI()
 app.include_router(nto_cto_router, tags=["NTO-CTO"])
 app.include_router(preprocessing_router, tags=["Image-Preprocessing"])
 app.include_router(image_regeneration_router, tags=["Image-Regeneration"])
@@ -60,4 +60,4 @@ app.add_middleware(
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
-)
+)

src/api/batch_api.py

@@ -9,6 +9,7 @@ import time
 from io import BytesIO
 import json
 import asyncio
+import aiohttp
 
 from PIL import Image
 from fastapi import File, UploadFile, Form
@@ -241,3 +242,134 @@ async def rt_nto(
             "Transfer-Encoding": "chunked"
         }
     )
+
+
+@batch_router.post("/rt_cto_nto")
+async def rt_cto_nto(
+        image: UploadFile = File(...),
+        c_list: str = Form(...),
+        necklace_id: str = Form(...),
+        necklace_category: str = Form(...),
+        storename: str = Form(...)
+):
+    logger.info("-" * 50)
+    logger.info(">>> REAL-TIME CTO-NTO STARTED <<<")
+    logger.info(f"Parameters: storename={storename}, necklace_category={necklace_category}, "
+                f"necklace_id={necklace_id}, clothing_list={c_list}")
+    
+    try:
+        clothing_list = [item.strip() for item in c_list.split(",")]
+        image_bytes = await image.read()
+        source_image = Image.open(BytesIO(image_bytes)).convert("RGB")
+        
+        jewellery_url = f"https://lvuhhlrkcuexzqtsbqyu.supabase.co/storage/v1/object/public/Stores/{storename}/{necklace_category}/image/{necklace_id}.png"
+        jewellery = Image.open(returnBytesData(url=jewellery_url)).convert("RGBA")
+        logger.info(">>> IMAGES LOADED SUCCESSFULLY <<<")
+    except Exception as e:
+        logger.error(f">>> INITIAL SETUP ERROR: {str(e)} <<<")
+        return JSONResponse(
+            content={"error": "Error in initial setup", "details": str(e), "code": 500},
+            status_code=500
+        )
+
+    async def generate():
+        start_time = time.time()
+        
+        try:
+            mask, _, _ = await pipeline.shoulderPointMaskGeneration_(image=source_image)
+            logger.info(">>> MASK GENERATION COMPLETED <<<")
+        except Exception as e:
+            logger.error(f">>> MASK GENERATION ERROR: {str(e)} <<<")
+            yield json.dumps({"error": "Error generating mask", "code": 500}) + "\n"
+            return
+
+        try:
+            mask_img_base_64, act_img_base_64 = BytesIO(), BytesIO()
+            mask.save(mask_img_base_64, format="WEBP")
+            source_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_}"
+            logger.info(">>> IMAGE ENCODING COMPLETED <<<")
+        except Exception as e:
+            logger.error(f">>> IMAGE ENCODING ERROR: {str(e)} <<<")
+            yield json.dumps({"error": "Error converting images to base64", "code": 500}) + "\n"
+            return
+
+        for idx, clothing_type in enumerate(clothing_list):
+            try:
+                # Perform CTO
+                cto_output = replicate_run_cto({
+                    "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
+                })
+                
+                if not cto_output or not isinstance(cto_output, (list, tuple)) or not cto_output[0]:
+                    raise ValueError("Invalid output from clothing try-on")
+
+                # Get CTO result image
+                async with aiohttp.ClientSession() as session:
+                    async with session.get(str(cto_output[0])) as response:
+                        if response.status != 200:
+                            raise ValueError("Failed to fetch CTO output")
+                        cto_result_bytes = await response.read()
+
+                # Process NTO
+                with BytesIO(cto_result_bytes) as buf:
+                    cto_result_image = Image.open(buf).convert("RGB")
+                    result, headerText, mask = await pipeline.necklaceTryOn_(
+                        image=cto_result_image,
+                        jewellery=jewellery,
+                        storename=storename
+                    )
+
+                if result is None:
+                    raise ValueError("Failed to process necklace try-on")
+
+                # Upload result
+                result_url = await supabase_upload_and_return_url(prefix="clothing_necklace_try_on", image=result)
+                if not result_url:
+                    raise ValueError("Failed to upload result image")
+
+                # Stream result
+                output_result = {
+                    "code": 200,
+                    "output": result_url,
+                    "inference_time": round((time.time() - start_time), 2),
+                    "clothing_type": clothing_type,
+                    "progress": f"{idx + 1}/{len(clothing_list)}"
+                }
+                yield json.dumps(output_result) + "\n"
+                await asyncio.sleep(0.1)
+
+                # Clean up
+                del result
+                gc.collect()
+
+            except Exception as e:
+                logger.error(f">>> PROCESSING ERROR FOR {clothing_type}: {str(e)} <<<")
+                error_result = {
+                    "error": f"Error processing clothing {clothing_type}",
+                    "details": str(e),
+                    "code": 500,
+                    "clothing_type": clothing_type,
+                    "progress": f"{idx + 1}/{len(clothing_list)}"
+                }
+                yield json.dumps(error_result) + "\n"
+                await asyncio.sleep(0.1)
+
+    return StreamingResponse(
+        generate(),
+        media_type="application/x-ndjson",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+            "Transfer-Encoding": "chunked"
+        }
+    )