import hashlib import json import os import uuid from enum import Enum import folder_paths import node_helpers import numpy as np import torch from PIL import Image, ImageOps, ImageSequence from bizyair.common.env_var import BIZYAIR_SERVER_ADDRESS from bizyair.image_utils import ( decode_base64_to_np, decode_data, encode_data, encode_image_to_base64, ) from nodes import LoadImage from .utils import ( decode_and_deserialize, get_api_key, send_post_request, serialize_and_encode, ) class RemoveBackground: API_URL = f"{BIZYAIR_SERVER_ADDRESS}/supernode/removebg" @classmethod def INPUT_TYPES(s): return { "required": { "image": ("IMAGE",), } } RETURN_TYPES = ("IMAGE", "MASK") FUNCTION = "remove_background" CATEGORY = "☁️BizyAir" def remove_background(self, image): API_KEY = get_api_key() device = image.device _, h, w, _ = image.shape assert ( w <= 1536 and h <= 1536 ), f"width and height must be less than 1536, but got {w} and {h}" payload = { "is_compress": True, "image": None, } auth = f"Bearer {API_KEY}" headers = { "accept": "application/json", "content-type": "application/json", "authorization": auth, } input_image, compress = serialize_and_encode(image, compress=True) payload["image"] = input_image payload["is_compress"] = compress response: str = send_post_request( self.API_URL, payload=payload, headers=headers ) tensors = decode_and_deserialize(response) t_images = tensors["images"].to(device) t_mask = tensors["mask"].to(device) return (t_images, t_mask) class GenerateLightningImage: API_URL = f"{BIZYAIR_SERVER_ADDRESS}/supernode/realvis4lightning" @classmethod def INPUT_TYPES(s): return { "required": { "prompt": ( "STRING", {"multiline": True, "dynamicPrompts": True, "default": "a dog"}, ), "seed": ("INT", {"default": 1, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}), "width": ("INT", {"default": 1024, "min": 16, "max": 1024, "step": 8}), "height": ("INT", {"default": 1024, "min": 16, "max": 1024, "step": 8}), "cfg": ( "FLOAT", { "default": 1.5, "min": 0.0, "max": 10.0, "step": 0.1, "round": 0.01, }, ), "batch_size": ("INT", {"default": 1, "min": 1, "max": 4}), } } RETURN_TYPES = ("IMAGE",) FUNCTION = "generate_image" CATEGORY = "☁️BizyAir" def generate_image(self, prompt, seed, width, height, cfg, batch_size): API_KEY = get_api_key() assert ( width <= 1024 and height <= 1024 ), f"width and height must be less than 1024, but got {width} and {height}" payload = { "batch_size": batch_size, "width": width, "height": height, "prompt": prompt, "cfg": cfg, "seed": seed, } auth = f"Bearer {API_KEY}" headers = { "accept": "application/json", "content-type": "application/json", "authorization": auth, } response: str = send_post_request( self.API_URL, payload=payload, headers=headers ) tensors_np = decode_and_deserialize(response) tensors = torch.from_numpy(tensors_np) return (tensors,) class ToggleServerEndpoint: BIZYAIR_SERVER_ENDPOINTS = [ "https://bizyair-api-st.siliconflow.cn/x/v1", "https://bizyair-api.siliconflow.cn/x/v1", ] def __init__(self): self.current_index = 0 @classmethod def INPUT_TYPES(s): return { "required": { "endpoint": (s.BIZYAIR_SERVER_ENDPOINTS,), } } RETURN_TYPES = () FUNCTION = "toggle_endpoint" OUTPUT_NODE = True CATEGORY = "☁️BizyAir" def toggle_endpoint(self, endpoint): BIZYAIR_SERVER_ADDRESS.address = endpoint from server import PromptServer PromptServer.instance.send_sync( "bizyair.server.endpoint.switch", {"message": f"Switch server endpoint to {endpoint}"}, ) return () NODE_CLASS_MAPPINGS = { "BizyAirRemoveBackground": RemoveBackground, "BizyAirGenerateLightningImage": GenerateLightningImage, "BizyAirToggleServerEndpoint": ToggleServerEndpoint, } NODE_DISPLAY_NAME_MAPPINGS = { "BizyAirRemoveBackground": "☁️BizyAir Remove Image Background", "BizyAirGenerateLightningImage": "☁️BizyAir Generate Photorealistic Images", "BizyAirToggleServerEndpoint": "☁️BizyAir Switch Server Endpoint", }