Add files using upload-large-folder tool
c225a63 verified 6 months ago
32.9 kB
# ---
# ComfyUI INSTARAW - Parallel Batch Generator Node
# Runs N API requests in parallel, accepts prompt lists from RPG
# Copyright © 2025 Instara. All rights reserved.
# PROPRIETARY SOFTWARE - ALL RIGHTS RESERVED
# ---

import requests
import base64
import io
import time
import os
import hashlib
import numpy as np
import torch
from PIL import Image
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

from .generative_api_nodes import MODEL_CONFIG, INSTARAW_GenerativeAPIBase


class INSTARAW_ParallelBatchGenerator(INSTARAW_GenerativeAPIBase):
    """
    Parallel batch image generation node.
    Runs multiple API requests concurrently using ThreadPoolExecutor.
    Accepts prompt lists from RPG and returns all images as batch tensor.
    """

    INPUT_IS_LIST = True  # Accept lists from RPG

    # Multi endpoint aspect ratios (limited)
    MULTI_ASPECT_RATIOS = ["3:2", "2:3", "3:4", "4:3"]

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "api_key": (
                    "STRING",
                    {
                        "forceInput": True,
                        "tooltip": "API key from provider",
                    },
                ),
                "provider": (
                    "STRING",
                    {
                        "forceInput": True,
                        "tooltip": "API provider (fal.ai or wavespeed.ai)",
                    },
                ),
                "model": (
                    "STRING",
                    {
                        "forceInput": True,
                        "tooltip": "Model name from Model Selector",
                    },
                ),
                "prompts": (
                    "STRING",
                    {
                        "forceInput": True,
                        "tooltip": "Prompt list from RPG (prompt_list_positive)",
                    },
                ),
                "max_parallel": (
                    "INT",
                    {
                        "default": 10,
                        "min": 1,
                        "max": 50,
                        "tooltip": "Maximum concurrent API requests",
                    },
                ),
                "aspect_ratio": (
                    "STRING",
                    {
                        "forceInput": True,
                        "tooltip": "Aspect ratio from Nano Banana Aspect Ratio node",
                    },
                ),
            },
            "optional": {
                "seeds": (
                    "INT",
                    {
                        "forceInput": True,
                        "tooltip": "Seed list from RPG (seed_list)",
                    },
                ),
                "width": (
                    "INT",
                    {
                        "forceInput": True,
                        "tooltip": "Image width from Aspect Ratio node (for Seedream)",
                    },
                ),
                "height": (
                    "INT",
                    {
                        "forceInput": True,
                        "tooltip": "Image height from Aspect Ratio node (for Seedream)",
                    },
                ),
                "resolution": (
                    "STRING",
                    {
                        "forceInput": True,
                        "tooltip": "Resolution tier (1K, 2K, 4K) for Nano Banana Pro",
                    },
                ),
                "images": (
                    "IMAGE",
                    {
                        "forceInput": True,
                        "tooltip": "Reference images from AIL for I2I/edit mode (batch tensor)",
                    },
                ),
                "use_multi_endpoint": (
                    "BOOLEAN",
                    {
                        "default": False,
                        "tooltip": "Use Multi endpoint (Nano Banana Pro on WaveSpeed only). Returns 2 images per request at half cost.",
                    },
                ),
                "enable_safety_checker": (
                    "BOOLEAN",
                    {
                        "default": True,
                        "tooltip": "Enable safety checker (for Seedream)",
                    },
                ),
            },
        }

    RETURN_TYPES = ("IMAGE", "INT")
    RETURN_NAMES = ("images", "count")
    OUTPUT_IS_LIST = (False, False)  # Returns single batch tensor and count
    FUNCTION = "generate_batch"
    CATEGORY = "INSTARAW/API"
    DESCRIPTION = (
        "Parallel batch image generator. Accepts prompt lists from RPG and runs "
        "multiple API requests concurrently. Supports all models and Multi endpoint."
    )

    def __init__(self):
        super().__init__()
        self._progress_lock = threading.Lock()
        self._completed_count = 0
        self._total_count = 0

    def _get_cache_dir(self):
        """Get or create cache directory."""
        cache_dir = os.path.join(os.path.dirname(__file__), "..", "..", "cache")
        os.makedirs(cache_dir, exist_ok=True)
        return cache_dir

    def _compute_cache_key(self, prompt, seed, model, provider, aspect_ratio, resolution, is_multi=False):
        """Compute cache key for a single generation."""
        hasher = hashlib.sha256()
        hasher.update(prompt.encode("utf-8"))
        hasher.update(str(seed).encode("utf-8"))
        hasher.update(model.encode("utf-8"))
        hasher.update(provider.encode("utf-8"))
        hasher.update(aspect_ratio.encode("utf-8"))
        if resolution:
            hasher.update(resolution.encode("utf-8"))
        if is_multi:
            hasher.update(b"_multi")
        return hasher.hexdigest()

    def _check_cache(self, cache_key, is_multi=False):
        """Check if image(s) exist in cache."""
        cache_dir = self._get_cache_dir()
        if is_multi:
            cache_files = [
                os.path.join(cache_dir, f"{cache_key}_batch_multi_{i}.png")
                for i in range(2)
            ]
            if all(os.path.exists(f) for f in cache_files):
                return cache_files
            return None
        else:
            cache_file = os.path.join(cache_dir, f"{cache_key}_batch.png")
            if os.path.exists(cache_file):
                return [cache_file]
            return None

    def _load_from_cache(self, cache_files):
        """Load image tensor(s) from cache files."""
        tensors = []
        for cache_file in cache_files:
            img = Image.open(cache_file).convert("RGB")
            img_np = np.array(img).astype(np.float32) / 255.0
            tensors.append(torch.from_numpy(img_np))
        return tensors

    def _save_to_cache(self, cache_key, images, is_multi=False):
        """Save image tensor(s) to cache."""
        cache_dir = self._get_cache_dir()
        cache_files = []
        for i, img_tensor in enumerate(images):
            if is_multi:
                cache_file = os.path.join(cache_dir, f"{cache_key}_batch_multi_{i}.png")
            else:
                cache_file = os.path.join(cache_dir, f"{cache_key}_batch.png")

            # Convert tensor to PIL and save
            img_np = img_tensor.cpu().numpy()
            if img_np.max() <= 1.0:
                img_np = (img_np * 255).astype(np.uint8)
            else:
                img_np = img_np.astype(np.uint8)
            img_pil = Image.fromarray(img_np)
            img_pil.save(cache_file, "PNG")
            cache_files.append(cache_file)
        return cache_files

    def _submit_wavespeed_multi(self, api_key, payload):
        """Submit request to WaveSpeed Multi endpoint (returns 2 images)."""
        base_url = "https://api.wavespeed.ai/api/v3"
        endpoint = "google/nano-banana-pro/text-to-image-multi"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

        # Submit task
        response = requests.post(
            f"{base_url}/{endpoint}",
            json=payload,
            headers=headers,
            timeout=30,
        )

        if not response.ok:
            raise Exception(
                f"WaveSpeed Multi API submission failed: {response.status_code} - {response.text}"
            )

        result = response.json()
        request_id = result["data"]["id"]

        # Poll for result
        poll_url = f"{base_url}/predictions/{request_id}/result"
        start_time = time.time()
        timeout = 300  # 5 minutes

        while time.time() - start_time < timeout:
            poll_response = requests.get(poll_url, headers=headers, timeout=30)

            if not poll_response.ok:
                raise Exception(
                    f"WaveSpeed Multi polling failed: {poll_response.status_code} - {poll_response.text}"
                )

            data = poll_response.json()["data"]
            status = data.get("status")

            if status == "completed":
                return data["outputs"]  # Returns list of image URLs

            if status == "failed":
                error_msg = data.get("error", "Unknown error")
                raise Exception(f"WaveSpeed Multi task failed: {error_msg}")

            time.sleep(2)

        raise Exception("WaveSpeed Multi task timed out after 5 minutes.")

    def _submit_fal_with_retry(self, endpoint, payload, api_key, max_retries=2, timeout=300):
        """Submit to fal.ai with retry logic and longer timeout."""
        url = f"https://fal.run/{endpoint}"
        headers = {
            "Authorization": f"Key {api_key}",
            "Content-Type": "application/json",
        }

        last_error = None
        for attempt in range(max_retries + 1):
            try:
                if attempt > 0:
                    print(f"   🔄 Retry {attempt}/{max_retries}...", flush=True)

                print(f"   📡 Submitting to fal.ai: {endpoint} (timeout={timeout}s)", flush=True)
                response = requests.post(url, json=payload, headers=headers, timeout=timeout)
                print(f"   📦 Response received: {response.status_code}", flush=True)

                if not response.ok:
                    if response.status_code == 422:
                        # Don't retry validation errors
                        raise Exception(f"API Error (422): Request rejected - {response.text[:200]}")
                    raise Exception(f"API request failed: {response.status_code} - {response.text[:200]}")

                result = response.json()
                if "images" in result and len(result["images"]) > 0:
                    return result["images"][0]["url"]
                raise Exception(f"API response did not contain an image URL")

            except requests.exceptions.Timeout as e:
                last_error = f"Timeout after {timeout}s"
                print(f"   ⏱️ Request timed out after {timeout}s", flush=True)
                if attempt < max_retries:
                    continue
            except requests.exceptions.RequestException as e:
                last_error = str(e)
                print(f"   ❌ Request error: {last_error[:100]}", flush=True)
                if attempt < max_retries:
                    time.sleep(2)  # Brief pause before retry
                    continue
            except Exception as e:
                print(f"   ❌ Non-retryable error: {str(e)[:100]}", flush=True)
                # Non-retryable errors
                raise

        raise Exception(last_error or "Request failed after retries")

    def _generate_single(self, idx, prompt, seed, api_key, provider, model, aspect_ratio,
                         resolution, width, height, enable_safety_checker, input_image=None):
        """Generate a single image. Returns (idx, [tensor], error)."""
        try:
            print(f"   🎯 Starting request #{idx + 1}: {prompt[:50]}...", flush=True)

            # Build kwargs for payload
            kwargs = {
                "api_key": api_key,
                "provider": provider,
                "model": model,
                "prompt": prompt,
                "seed": seed,
                "aspect_ratio": aspect_ratio,
                "width": width,
                "height": height,
                "enable_safety_checker": enable_safety_checker,
            }
            if resolution:
                kwargs["resolution"] = resolution
            if input_image is not None:
                kwargs["image_1"] = input_image
                print(f"   🖼️ Request #{idx + 1} includes reference image, shape: {input_image.shape}", flush=True)

            # Check cache first
            cache_key = self._compute_cache_key(prompt, seed, model, provider, aspect_ratio, resolution)
            cached = self._check_cache(cache_key)
            if cached:
                tensors = self._load_from_cache(cached)
                return (idx, tensors, None)

            # Get model config
            model_conf = MODEL_CONFIG.get(model)
            if not model_conf:
                raise ValueError(f"Invalid model: {model}")
            provider_conf = model_conf["providers"].get(provider)
            if not provider_conf:
                raise ValueError(f"Provider '{provider}' not supported for model '{model}'")

            # Determine endpoint
            is_i2i = input_image is not None
            endpoint = provider_conf["i2i_endpoint"] if is_i2i else provider_conf["t2i_endpoint"]
            build_payload_func = provider_conf["build_payload"]

            # Build payload
            self.set_api_key(api_key)
            payload = build_payload_func(self, **kwargs)

            # Submit request with provider-specific handling
            if provider == "fal.ai":
                # Use our own method with longer timeout and retries
                image_url = self._submit_fal_with_retry(endpoint, payload, api_key)
            else:
                # Use base class method for wavespeed (already has polling)
                image_url = self.submit_request(provider, endpoint, payload)

            # Download image with longer timeout for large images
            image_response = requests.get(image_url, timeout=120)
            image_response.raise_for_status()
            image_pil = Image.open(io.BytesIO(image_response.content)).convert("RGB")

            # Convert to tensor
            image_np = np.array(image_pil).astype(np.float32) / 255.0
            tensor = torch.from_numpy(image_np)

            # Save to cache
            self._save_to_cache(cache_key, [tensor])

            return (idx, [tensor], None)

        except Exception as e:
            return (idx, None, str(e))

    def _generate_single_multi(self, idx, prompt, api_key, aspect_ratio):
        """Generate using Multi endpoint (2 images per request). Returns (idx, [tensor1, tensor2], error)."""
        try:
            # Check cache first
            cache_key = self._compute_cache_key(
                prompt, -1, "Nano Banana Pro", "wavespeed.ai", aspect_ratio, None, is_multi=True
            )
            cached = self._check_cache(cache_key, is_multi=True)
            if cached:
                tensors = self._load_from_cache(cached)
                return (idx, tensors, None)

            # Build payload for multi endpoint
            payload = {
                "prompt": prompt,
                "aspect_ratio": aspect_ratio,
                "num_images": 2,
                "output_format": "jpeg",
            }

            # Submit request
            image_urls = self._submit_wavespeed_multi(api_key, payload)

            if not image_urls or len(image_urls) < 2:
                raise Exception(f"Multi endpoint returned {len(image_urls) if image_urls else 0} images, expected 2")

            # Download and convert images
            tensors = []
            for image_url in image_urls[:2]:
                image_response = requests.get(image_url, timeout=60)
                image_response.raise_for_status()
                image_pil = Image.open(io.BytesIO(image_response.content)).convert("RGB")
                image_np = np.array(image_pil).astype(np.float32) / 255.0
                tensors.append(torch.from_numpy(image_np))

            # Save to cache
            self._save_to_cache(cache_key, tensors, is_multi=True)

            return (idx, tensors, None)

        except Exception as e:
            return (idx, None, str(e))

    def _report_progress(self, completed, total, is_multi=False):
        """Report progress to console."""
        mode = "Multi (2x)" if is_multi else "Standard"
        pct = 100 * completed // total if total > 0 else 0
        print(f"🚀 [{mode}] Progress: {completed}/{total} requests ({pct}%)", flush=True)

    def generate_batch(
        self,
        api_key,
        provider,
        model,
        prompts,
        max_parallel,
        aspect_ratio,
        seeds=None,
        width=None,
        height=None,
        resolution=None,
        images=None,
        use_multi_endpoint=None,
        enable_safety_checker=None,
    ):
        """
        Generate images in parallel batches.

        Args:
            api_key: API key (list from INPUT_IS_LIST)
            provider: Provider name (list)
            model: Model name (list)
            prompts: List of prompts from RPG
            max_parallel: Max concurrent requests (list)
            aspect_ratio: Aspect ratio (list)
            seeds: List of seeds from RPG (optional)
            width: Image width (list, optional)
            height: Image height (list, optional)
            resolution: Resolution tier (list, optional)
            images: Input images for I2I mode (batch tensor, optional)
            use_multi_endpoint: Use Multi endpoint (list, optional)
            enable_safety_checker: Enable safety checker (list, optional)

        Returns:
            Tuple of (batch_tensor, count)
        """
        # Extract single values from lists (INPUT_IS_LIST wraps everything)
        api_key_val = api_key[0] if isinstance(api_key, list) else api_key
        provider_val = provider[0] if isinstance(provider, list) else provider
        model_val = model[0] if isinstance(model, list) else model
        max_parallel_val = max_parallel[0] if isinstance(max_parallel, list) else max_parallel
        aspect_ratio_val = aspect_ratio[0] if isinstance(aspect_ratio, list) else aspect_ratio

        # Optional values
        width_val = width[0] if isinstance(width, list) and width else 1024
        height_val = height[0] if isinstance(height, list) and height else 1024
        resolution_val = resolution[0] if isinstance(resolution, list) and resolution else None
        use_multi_val = use_multi_endpoint[0] if isinstance(use_multi_endpoint, list) and use_multi_endpoint else False
        enable_safety_val = enable_safety_checker[0] if isinstance(enable_safety_checker, list) and enable_safety_checker else True

        # prompts should already be a list from RPG
        prompt_list = prompts if isinstance(prompts, list) else [prompts]

        # Handle seeds list
        if seeds is None or (isinstance(seeds, list) and len(seeds) == 0):
            seed_list = [-1] * len(prompt_list)
        elif isinstance(seeds, list):
            seed_list = seeds
            # Pad with -1 if seeds list is shorter
            while len(seed_list) < len(prompt_list):
                seed_list.append(-1)
        else:
            seed_list = [seeds] * len(prompt_list)

        # Handle input images for I2I mode
        # With INPUT_IS_LIST=True, images can come as:
        # 1. A list of individual tensors (from PromptFilter with OUTPUT_IS_LIST=True)
        # 2. A list containing a single batch tensor (from other nodes)
        # 3. A single batch tensor
        input_images = None
        has_reference_images = False

        if images is not None:
            print(f"   🖼️ Received images input: type={type(images)}, len={len(images) if isinstance(images, list) else 'N/A'}", flush=True)

            if isinstance(images, list) and len(images) > 0:
                # Check if it's a list of individual tensors or a list with one batch tensor
                first_item = images[0]
                if first_item is not None:
                    if isinstance(first_item, torch.Tensor):
                        # Check dimensions to determine format
                        if first_item.dim() == 3:
                            # List of individual [H, W, C] tensors - stack them
                            print(f"   🖼️ Detected list of {len(images)} individual image tensors", flush=True)
                            valid_tensors = [img for img in images if img is not None and isinstance(img, torch.Tensor)]
                            if valid_tensors:
                                # Add batch dimension to each and stack
                                stacked = []
                                for img in valid_tensors:
                                    if img.dim() == 3:
                                        stacked.append(img.unsqueeze(0))
                                    else:
                                        stacked.append(img)
                                input_images = torch.cat(stacked, dim=0)
                                print(f"   🖼️ Stacked into batch: shape={input_images.shape}", flush=True)
                        elif first_item.dim() == 4:
                            # First item is already a batch tensor [B, H, W, C]
                            if len(images) == 1:
                                input_images = first_item
                                print(f"   🖼️ Single batch tensor: shape={input_images.shape}", flush=True)
                            else:
                                # Multiple batch tensors - concatenate them
                                valid_tensors = [img for img in images if img is not None and isinstance(img, torch.Tensor)]
                                input_images = torch.cat(valid_tensors, dim=0)
                                print(f"   🖼️ Concatenated batches: shape={input_images.shape}", flush=True)
            elif isinstance(images, torch.Tensor):
                input_images = images
                print(f"   🖼️ Direct tensor: shape={input_images.shape}", flush=True)

            # Validate we have usable images
            if input_images is not None and isinstance(input_images, torch.Tensor) and len(input_images) > 0:
                has_reference_images = True
                num_images = len(input_images)
                print(f"   🖼️ I2I MODE: Using {num_images} reference images (edit endpoint)", flush=True)

                if num_images < len(prompt_list):
                    print(f"   ⚠️ Only {num_images} images for {len(prompt_list)} prompts - will cycle through available images", flush=True)
                elif num_images > len(prompt_list):
                    print(f"   ⚠️ {num_images} images but only {len(prompt_list)} prompts - extra images will be ignored", flush=True)

        if not has_reference_images:
            input_images = None
            print(f"   📝 T2I MODE: No reference images provided", flush=True)

        total_prompts = len(prompt_list)
        print(f"🎨 Parallel Batch Generator: {total_prompts} prompts, max {max_parallel_val} parallel", flush=True)
        print(f"   Model: {model_val} | Provider: {provider_val} | Aspect: {aspect_ratio_val}", flush=True)

        # Check if we should use Multi endpoint
        # Multi endpoint is T2I ONLY - cannot use with reference images!
        is_multi_mode = False
        if use_multi_val:
            if has_reference_images:
                print(f"   ⚠️ Multi endpoint requested but reference images provided - using standard I2I endpoint instead", flush=True)
            elif model_val != "Nano Banana Pro":
                print(f"   ⚠️ Multi endpoint only available for Nano Banana Pro", flush=True)
            elif provider_val != "wavespeed.ai":
                print(f"   ⚠️ Multi endpoint only available on wavespeed.ai", flush=True)
            elif aspect_ratio_val not in self.MULTI_ASPECT_RATIOS:
                print(f"   ⚠️ Multi endpoint doesn't support aspect ratio {aspect_ratio_val} (only: {self.MULTI_ASPECT_RATIOS})", flush=True)
            else:
                is_multi_mode = True

        if is_multi_mode:
            print(f"   🍌 Using Multi endpoint (2 images per request)", flush=True)
            return self._generate_batch_multi(
                api_key_val, prompt_list, max_parallel_val, aspect_ratio_val
            )
        else:
            return self._generate_batch_standard(
                api_key_val, provider_val, model_val, prompt_list, seed_list,
                max_parallel_val, aspect_ratio_val, resolution_val, width_val,
                height_val, enable_safety_val, input_images
            )

    def _generate_batch_standard(
        self, api_key, provider, model, prompts, seeds, max_parallel,
        aspect_ratio, resolution, width, height, enable_safety, input_images
    ):
        """Generate images using standard endpoints (1 image per request)."""
        total = len(prompts)
        results = [None] * total
        errors = []
        completed = 0

        with ThreadPoolExecutor(max_workers=max_parallel) as executor:
            futures = {}
            num_ref_images = len(input_images) if input_images is not None else 0

            for i in range(total):
                # Get input image for this index - cycle through if fewer images than prompts
                input_img = None
                if input_images is not None and num_ref_images > 0:
                    img_idx = i % num_ref_images  # Cycle through available images
                    input_img = input_images[img_idx:img_idx+1]  # Keep batch dimension [1, H, W, C]

                future = executor.submit(
                    self._generate_single,
                    i, prompts[i], seeds[i], api_key, provider, model,
                    aspect_ratio, resolution, width, height, enable_safety, input_img
                )
                futures[future] = i

            mode_str = f"I2I with {num_ref_images} ref images" if num_ref_images > 0 else "T2I"
            print(f"   📋 Submitted {total} requests ({mode_str}), waiting for completion...", flush=True)

            for future in as_completed(futures):
                idx, tensors, error = future.result()
                completed += 1

                if error:
                    errors.append((idx, prompts[idx], error))
                    print(f"❌ Request {idx + 1}/{total} failed: {error}", flush=True)
                else:
                    results[idx] = tensors[0]  # Single image
                    self._report_progress(completed, total, is_multi=False)

        # Report failures
        if errors:
            print(f"⚠️ {len(errors)} of {total} generations failed:", flush=True)
            for idx, prompt, err in errors[:5]:  # Show first 5
                print(f"   - #{idx + 1}: {err[:100]}...", flush=True)

        # Filter out None results
        successful = [r for r in results if r is not None]
        if not successful:
            raise Exception("All generations failed. Check API key and parameters.")

        # Ensure all images have the same size (use first image's size as reference)
        # This handles cases where API returns slightly different sizes
        if len(successful) > 1:
            ref_shape = successful[0].shape  # (H, W, C)
            resized = []
            for i, tensor in enumerate(successful):
                if tensor.shape != ref_shape:
                    print(f"   ⚠️ Image {i + 1} has different size {tensor.shape}, resizing to {ref_shape}", flush=True)
                    # Resize using PIL for quality
                    img_np = tensor.cpu().numpy()
                    if img_np.max() <= 1.0:
                        img_np = (img_np * 255).astype(np.uint8)
                    img_pil = Image.fromarray(img_np)
                    img_pil = img_pil.resize((ref_shape[1], ref_shape[0]), Image.LANCZOS)
                    img_np = np.array(img_pil).astype(np.float32) / 255.0
                    resized.append(torch.from_numpy(img_np))
                else:
                    resized.append(tensor)
            successful = resized

        batch_tensor = torch.stack(successful)
        print(f"✅ Parallel Batch Generator complete! Generated {len(successful)} images, shape: {batch_tensor.shape}", flush=True)

        return (batch_tensor, len(successful))

    def _generate_batch_multi(self, api_key, prompts, max_parallel, aspect_ratio):
        """Generate images using Multi endpoint (2 images per request)."""
        total = len(prompts)
        results = []
        errors = []
        completed = 0

        with ThreadPoolExecutor(max_workers=max_parallel) as executor:
            futures = {}
            for i in range(total):
                future = executor.submit(
                    self._generate_single_multi,
                    i, prompts[i], api_key, aspect_ratio
                )
                futures[future] = i

            for future in as_completed(futures):
                idx, tensors, error = future.result()
                completed += 1

                if error:
                    errors.append((idx, prompts[idx], error))
                    print(f"❌ Multi request {idx + 1}/{total} failed: {error}")
                else:
                    # Multi returns 2 images per request
                    results.append((idx, tensors))
                    self._report_progress(completed, total, is_multi=True)

        # Report failures
        if errors:
            print(f"⚠️ {len(errors)} of {total} multi-generations failed:")
            for idx, prompt, err in errors[:5]:
                print(f"   - #{idx + 1}: {err[:100]}...")

        # Sort by original index and flatten
        results.sort(key=lambda x: x[0])
        all_tensors = []
        for idx, tensors in results:
            all_tensors.extend(tensors)

        if not all_tensors:
            raise Exception("All multi-generations failed. Check API key and parameters.")

        # Ensure all images have the same size (use first image's size as reference)
        if len(all_tensors) > 1:
            ref_shape = all_tensors[0].shape  # (H, W, C)
            resized = []
            for i, tensor in enumerate(all_tensors):
                if tensor.shape != ref_shape:
                    print(f"   ⚠️ Image {i + 1} has different size {tensor.shape}, resizing to {ref_shape}", flush=True)
                    img_np = tensor.cpu().numpy()
                    if img_np.max() <= 1.0:
                        img_np = (img_np * 255).astype(np.uint8)
                    img_pil = Image.fromarray(img_np)
                    img_pil = img_pil.resize((ref_shape[1], ref_shape[0]), Image.LANCZOS)
                    img_np = np.array(img_pil).astype(np.float32) / 255.0
                    resized.append(torch.from_numpy(img_np))
                else:
                    resized.append(tensor)
            all_tensors = resized

        batch_tensor = torch.stack(all_tensors)
        expected_count = total * 2
        actual_count = len(all_tensors)
        print(f"✅ Multi Batch Generator complete! Generated {actual_count} images (expected {expected_count}), shape: {batch_tensor.shape}", flush=True)

        return (batch_tensor, actual_count)


# =================================================================================
# NODE REGISTRATION
# =================================================================================

NODE_CLASS_MAPPINGS = {
    "INSTARAW_ParallelBatchGenerator": INSTARAW_ParallelBatchGenerator,
}

NODE_DISPLAY_NAME_MAPPINGS = {
    "INSTARAW_ParallelBatchGenerator": "🚀 INSTARAW Parallel Batch Generator",
}