playground/ai_runpod.py

#!/usr/bin/env uv run
# /// script
# dependencies = ["httpx", "pillow"]
# ///

import argparse
import io
import json
import logging
import math
import random
import subprocess
import time
import uuid
from pathlib import Path

import httpx
from PIL import Image

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logging.getLogger("httpx").setLevel(logging.WARNING)  # Suppress httpx debug logs
logger = logging.getLogger(__name__)

COMFY_URL = "https://randy-agents-mediterranean-herself.trycloudflare.com/"
COMFY_COOKIES = {
    "C.35620185_auth_token": "027e0e069f66d46e6d84802b7d49076e6ebab8621491d8694992f16727f99b79",
}
WORKFLOW_FILE = Path("/Users/abdus/Downloads/klein_fixed.json")
IMAGE_OUTPUT_MEGAPIXELS = 2


SYSTEM_PROMPT = """
restyle this as an edited, finished photo taken by sony a7 iv.
Photorealistic textures.
same color grading, same colors and lighting.
Realistic, flawless skin.
# White people should have a pale skin.
age everyone by 5 years and make them look like 25 year old adults.

keep the same ethnicity and facial features.

Make the children taller and more slender with slightly elongated feminine neck and higher cheekbones, give them smaller, slimmer more feminine head and tapered face; round, smooth chin.

keep everything else about her the same.
same facial + eye expression (keep them as open as in the original), same pose and emotion, same tongue position.
same head and eye angle.

do not turn her face around.
same hair style, color, length.
same composition and crop.
keep the same makeup, do not change the colors of the lips, keep them glossy and wet.
long nipples. large areolas.

"""


def preprocess_image(image_path: Path) -> bytes:
    with Image.open(image_path) as img:
        if img.mode in ("RGBA", "P"):
            img = img.convert("RGB")
        max_size = 2500  # Slightly larger to support 3MP render
        if max(img.size) > max_size:
            img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
        buffer = io.BytesIO()
        img.save(buffer, format="JPEG", quality=85, optimize=True)
        return buffer.getvalue()


def _next_output_path(image_path: Path) -> Path:
    base = image_path.with_name(f"{image_path.stem}-edit.jpg")
    if not base.exists():
        return base
    counter = 2
    while True:
        candidate = image_path.with_name(f"{image_path.stem}-edit{counter}.jpg")
        if not candidate.exists():
            return candidate
        counter += 1


def tag_file(path: Path, tag: str = "Blue") -> None:
    if not path.exists():
        raise FileNotFoundError(f"Target path does not exist: {path}")

    abs_path = str(path.resolve())

    script = f'tell application "Finder" to set tags of (POSIX file "{abs_path}" as alias) to {{"{tag}"}}'

    try:
        subprocess.run(
            ["osascript", "-e", script], check=True, capture_output=True, text=True
        )
    except subprocess.CalledProcessError as e:
        # Finder may fail if the file is moved during execution or if permissions are insufficient
        logger.warning(f"Failed to tag file: {e.stderr}")


class Stats:
    def __init__(self) -> None:
        self.run_times: list[float] = []

    def add(self, elapsed: float) -> None:
        self.run_times.append(elapsed)

    def format_time(self, seconds: float) -> str:
        if seconds < 60:
            return f"{seconds:.1f}s"
        minutes = seconds / 60
        return f"{minutes:.1f}m"

    def print_summary(self) -> None:
        if not self.run_times:
            return

        if len(self.run_times) == 1:
            return

        shortest = min(self.run_times)
        longest = max(self.run_times)
        average = sum(self.run_times) / len(self.run_times)
        total = sum(self.run_times)

        print("=" * 20)
        print("Run Statistics")
        print("=" * 20)
        print(f"Total runs: {len(self.run_times)}")
        print(f"Shortest:   {self.format_time(shortest)}")
        print(f"Longest:    {self.format_time(longest)}")
        print(f"Average:    {self.format_time(average)}")
        print(f"Total time: {self.format_time(total)}")
        print("=" * 20)


class ComfyUI:
    PROMPT_NODE_ID = "75:74"
    NEGATIVE_NODE_ID = "75:67"
    IMAGE_LOAD_NODE_ID = "76"
    SAVE_NODE_ID = "9"
    UPSCALE_NODE_ID = "75:80"
    MODEL_LOAD_NODE_ID = "75:70"
    CLIP_LOAD_NODE_ID = "75:71"
    VAE_LOAD_NODE_ID = "75:72"
    SEED_NODE_ID = "75:73"

    def __init__(self, url: str, cookies: dict) -> None:
        self.url = url.rstrip("/")
        self.client = httpx.Client(timeout=None, cookies=cookies, follow_redirects=True)

    def _post(self, path: str, **kwargs) -> httpx.Response:
        """Try RunPod /api routes first, then legacy Comfy routes."""
        candidates = [f"{self.url}/api{path}", f"{self.url}{path}"]
        last_res: httpx.Response | None = None
        for endpoint in candidates:
            res = self.client.post(endpoint, **kwargs)
            if res.status_code != 404:
                return res
            last_res = res
        assert last_res is not None
        return last_res

    def _get(self, path: str, **kwargs) -> httpx.Response:
        """Try RunPod /api routes first, then legacy Comfy routes."""
        candidates = [f"{self.url}/api{path}", f"{self.url}{path}"]
        last_res: httpx.Response | None = None
        for endpoint in candidates:
            res = self.client.get(endpoint, **kwargs)
            if res.status_code != 404:
                return res
            last_res = res
        assert last_res is not None
        return last_res

    def transform(
        self,
        image_path: Path,
        save_path: Path,
        prompt: str,
        negative_prompt: str = "",
    ) -> None:
        seed = random.randint(0, 2**31 - 1)
        client_id = str(uuid.uuid4())

        # 1. Upload
        processed_img_bytes = preprocess_image(image_path)
        files = {"image": (f"upload_{uuid.uuid4()}.jpg", processed_img_bytes)}
        res = self._post("/upload/image", files=files)
        if res.is_error:
            logger.error(f"Upload error: {res.text}")
            res.raise_for_status()
        filename = res.json()["name"]

        # 2. Load Workflow
        with open(WORKFLOW_FILE, "r") as f:
            workflow = json.load(f)

        # Get dimensions from the actual file
        with Image.open(image_path) as img:
            orig_width, orig_height = img.size
            aspect_ratio = orig_width / orig_height

        # Calculate target resolution for 3 Megapixels (approx 3,000,000 pixels)
        # Formula: Width * (Width / Aspect) = TotalPixels
        target_pixels = IMAGE_OUTPUT_MEGAPIXELS * 1024 * 1024
        new_width = int(math.sqrt(target_pixels * aspect_ratio))
        new_height = int(new_width / aspect_ratio)

        # Force these into the Empty Latent node (Node 75:66)
        # This ensures the "canvas" matches your photo's shape
        workflow["75:66"]["inputs"]["width"] = (
            new_width // 8
        ) * 8  # Must be multiple of 8
        workflow["75:66"]["inputs"]["height"] = (new_height // 8) * 8

        # Also update the Scheduler (Node 75:62) so Flux knows the scale
        workflow["75:62"]["inputs"]["width"] = workflow["75:66"]["inputs"]["width"]
        workflow["75:62"]["inputs"]["height"] = workflow["75:66"]["inputs"]["height"]

        # Inject Advanced Data
        workflow[self.PROMPT_NODE_ID]["inputs"]["text"] = prompt
        workflow[self.NEGATIVE_NODE_ID]["inputs"]["text"] = negative_prompt
        workflow[self.IMAGE_LOAD_NODE_ID]["inputs"]["image"] = filename
        workflow[self.SEED_NODE_ID]["inputs"]["noise_seed"] = seed

        workflow[self.UPSCALE_NODE_ID]["inputs"]["megapixels"] = IMAGE_OUTPUT_MEGAPIXELS
        workflow[self.UPSCALE_NODE_ID]["inputs"]["upscale_method"] = "lanczos"

        # 3. Queue
        payload = {"prompt": workflow, "client_id": client_id}
        res = self._post("/prompt", json=payload)

        if res.is_error:
            logger.error(f"Error: {res.text}")
            res.raise_for_status()

        prompt_id = res.json()["prompt_id"]

        # 4. Wait
        preview_output = None
        while True:
            jobs_res = self._get(
                "/jobs",
                params={
                    "status": "completed,failed,cancelled",
                    "limit": 64,
                    "offset": 0,
                },
            )
            if jobs_res.is_error:
                logger.error(f"Jobs poll error: {jobs_res.text}")
                jobs_res.raise_for_status()

            jobs = jobs_res.json().get("jobs", [])
            match = next((job for job in jobs if job.get("id") == prompt_id), None)
            if match:
                if match.get("status") != "completed":
                    raise RuntimeError(
                        f"Job {prompt_id} ended with status: {match.get('status')}"
                    )
                preview_output = match.get("preview_output")
                break
            time.sleep(0.25)

        # 5. Download
        if not preview_output:
            raise RuntimeError(f"Completed job {prompt_id} has no preview_output")

        view_params = {
            "filename": preview_output["filename"],
            "subfolder": preview_output.get("subfolder", ""),
            "type": preview_output.get("type", "output"),
        }
        img_res = self._get("/view", params=view_params)
        if img_res.is_error:
            logger.error(f"View error: {img_res.text}")
            img_res.raise_for_status()

        # Convert to JPEG with quality=75
        with Image.open(io.BytesIO(img_res.content)) as img:
            if img.mode == "RGBA":
                img = img.convert("RGB")
            img.save(save_path, format="JPEG", quality=75)

def without_comments(s: str) -> str:
    return "\n".join(
        line for line in s.splitlines() if not line.strip().startswith("#")
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("images", nargs="+", type=Path)
    args = parser.parse_args()


    comfy = ComfyUI(COMFY_URL, COMFY_COOKIES)
    stats = Stats()

    for i, image_path in enumerate(args.images, start=1):
        prompt = SYSTEM_PROMPT
        replace_prompt_path = Path(__file__).parent / "flux_klein_prompt.txt"
        if replace_prompt_path.is_file():
            prompt = without_comments(replace_prompt_path.read_text())

        save_path = _next_output_path(image_path)
        start_time = time.time()
        logger.info(f"Uploading ({i}/{len(args.images)}): {image_path.name}")
        try:
            comfy.transform(image_path, save_path, prompt)
        except Exception as e:
            logger.error(f"Error processing {image_path.name}: {e}")
            continue
        elapsed = time.time() - start_time
        stats.add(elapsed)

    stats.print_summary()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())