xinyun/ComfyUI
1
0
Fork 0
mirror of https://git.datalinker.icu/comfyanonymous/ComfyUI synced 2026-09-06 13:57:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add a bunch of nodes, 3 ready to use, the rest waiting for endpoint support (#108)

Browse Source
This commit is contained in:
Jedrzej Kosinski 2025-05-02 05:49:05 -05:00 committed by GitHub
parent 3957839ff1
commit 3850c47fe1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 1557 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
comfy_api_nodes/apis/bfl_api.py
View File

@ -11,7 +11,104 @@ class BFLOutputFormat(str, Enum):
jpeg = 'jpeg'
class BFLFluxExpandImageRequest(BaseModel):
prompt: str = Field(..., description='The description of the changes you want to make. This text guides the expansion process, allowing you to specify features, styles, or modifications for the expanded areas.')
prompt_upsampling: Optional[bool] = Field(
None, description='Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation.'
)
seed: Optional[int] = Field(None, description='The seed value for reproducibility.')
top: conint(ge=0, le=2048) = Field(..., description='Number of pixels to expand at the top of the image')
bottom: conint(ge=0, le=2048) = Field(..., description='Number of pixels to expand at the bottom of the image')
left: conint(ge=0, le=2048) = Field(..., description='Number of pixels to expand at the left side of the image')
right: conint(ge=0, le=2048) = Field(..., description='Number of pixels to expand at the right side of the image')
steps: conint(ge=15, le=50) = Field(..., description='Number of steps for the image generation process')
guidance: confloat(ge=1.5, le=100) = Field(..., description='Guidance strength for the image generation process')
safety_tolerance: Optional[conint(ge=0, le=6)] = Field(
6, description='Tolerance level for input and output moderation. Between 0 and 6, 0 being most strict, 6 being least strict. Defaults to 2.'
)
output_format: Optional[BFLOutputFormat] = Field(
BFLOutputFormat.png, description="Output format for the generated image. Can be 'jpeg' or 'png'.", examples=['png']
)
image: str = Field(None, description='A Base64-encoded string representing the image you wish to expand')
class BFLFluxFillImageRequest(BaseModel):
prompt: str = Field(..., description='The description of the changes you want to make. This text guides the expansion process, allowing you to specify features, styles, or modifications for the expanded areas.')
prompt_upsampling: Optional[bool] = Field(
None, description='Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation.'
)
seed: Optional[int] = Field(None, description='The seed value for reproducibility.')
steps: conint(ge=15, le=50) = Field(..., description='Number of steps for the image generation process')
guidance: confloat(ge=1.5, le=100) = Field(..., description='Guidance strength for the image generation process')
safety_tolerance: Optional[conint(ge=0, le=6)] = Field(
6, description='Tolerance level for input and output moderation. Between 0 and 6, 0 being most strict, 6 being least strict. Defaults to 2.'
)
output_format: Optional[BFLOutputFormat] = Field(
BFLOutputFormat.png, description="Output format for the generated image. Can be 'jpeg' or 'png'.", examples=['png']
)
image: str = Field(None, description='A Base64-encoded string representing the image you wish to modify. Can contain alpha mask if desired.')
mask: str = Field(None, description='A Base64-encoded string representing the mask of the areas you with to modify.')
class BFLFluxCannyImageRequest(BaseModel):
prompt: str = Field(..., description='Text prompt for image generation')
prompt_upsampling: Optional[bool] = Field(
None, description='Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation.'
)
canny_low_threshold: Optional[int] = Field(None, description='Low threshold for Canny edge detection')
canny_high_threshold: Optional[int] = Field(None, description='High threshold for Canny edge detection')
seed: Optional[int] = Field(None, description='The seed value for reproducibility.')
steps: conint(ge=15, le=50) = Field(..., description='Number of steps for the image generation process')
guidance: confloat(ge=1, le=100) = Field(..., description='Guidance strength for the image generation process')
safety_tolerance: Optional[conint(ge=0, le=6)] = Field(
6, description='Tolerance level for input and output moderation. Between 0 and 6, 0 being most strict, 6 being least strict. Defaults to 2.'
)
output_format: Optional[BFLOutputFormat] = Field(
BFLOutputFormat.png, description="Output format for the generated image. Can be 'jpeg' or 'png'.", examples=['png']
)
control_image: Optional[str] = Field(None, description='Base64 encoded image to use as control input if no preprocessed image is provided')
preprocessed_image: Optional[str] = Field(None, description='Optional pre-processed image that will bypass the control preprocessing step')
class BFLFluxDepthImageRequest(BaseModel):
prompt: str = Field(..., description='Text prompt for image generation')
prompt_upsampling: Optional[bool] = Field(
None, description='Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation.'
)
seed: Optional[int] = Field(None, description='The seed value for reproducibility.')
steps: conint(ge=15, le=50) = Field(..., description='Number of steps for the image generation process')
guidance: confloat(ge=1, le=100) = Field(..., description='Guidance strength for the image generation process')
safety_tolerance: Optional[conint(ge=0, le=6)] = Field(
6, description='Tolerance level for input and output moderation. Between 0 and 6, 0 being most strict, 6 being least strict. Defaults to 2.'
)
output_format: Optional[BFLOutputFormat] = Field(
BFLOutputFormat.png, description="Output format for the generated image. Can be 'jpeg' or 'png'.", examples=['png']
)
control_image: Optional[str] = Field(None, description='Base64 encoded image to use as control input if no preprocessed image is provided')
preprocessed_image: Optional[str] = Field(None, description='Optional pre-processed image that will bypass the control preprocessing step')
class BFLFluxProGenerateRequest(BaseModel):
prompt: str = Field(..., description='The text prompt for image generation.')
prompt_upsampling: Optional[bool] = Field(
None, description='Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation.'
)
seed: Optional[int] = Field(None, description='The seed value for reproducibility.')
width: conint(ge=256, le=1440) = Field(1024, description='Width of the generated image in pixels. Must be a multiple of 32.')
height: conint(ge=256, le=1440) = Field(768, description='Height of the generated image in pixels. Must be a multiple of 32.')
safety_tolerance: Optional[conint(ge=0, le=6)] = Field(
6, description='Tolerance level for input and output moderation. Between 0 and 6, 0 being most strict, 6 being least strict. Defaults to 2.'
)
output_format: Optional[BFLOutputFormat] = Field(
BFLOutputFormat.png, description="Output format for the generated image. Can be 'jpeg' or 'png'.", examples=['png']
)
image_prompt: Optional[str] = Field(None, description='Optional image to remix in base64 format')
# image_prompt_strength: Optional[confloat(ge=0.0, le=1.0)] = Field(
# None, description='Blend between the prompt and the image prompt.'
# )
class BFLFluxProUltraGenerateRequest(BaseModel):
prompt: str = Field(..., description='The text prompt for image generation.')
prompt_upsampling: Optional[bool] = Field(
None, description='Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation.'

10
comfy_api_nodes/apis/recraft_api.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, conint
from pydantic import BaseModel, Field, conint, confloat
class RecraftColor:
@ -238,14 +238,15 @@ class RecraftControlsObject(BaseModel):
class RecraftImageGenerationRequest(BaseModel):
prompt: str = Field(..., description='The text prompt describing the image to generate')
size: RecraftImageSize = Field(..., description='The size of the generated image (e.g., "1024x1024")')
size: Optional[RecraftImageSize] = Field(None, description='The size of the generated image (e.g., "1024x1024")')
n: conint(ge=1, le=6) = Field(..., description='The number of images to generate')
negative_prompts: Optional[str] = Field(None, description='A text description of undesired elements on an image')
negative_prompt: Optional[str] = Field(None, description='A text description of undesired elements on an image')
model: Optional[RecraftModel] = Field(RecraftModel.recraftv3, description='The model to use for generation (e.g., "recraftv3")')
style: Optional[str] = Field(None, description='The style to apply to the generated image (e.g., "digital_illustration")')
substyle: Optional[str] = Field(None, description='The substyle to apply to the generated image, depending on the style input')
controls: Optional[RecraftControlsObject] = Field(None, description='A set of custom parameters to tweak generation process')
style_id: Optional[str] = Field(None, description='Use a previously uploaded style as a reference; UUID')
strength: Optional[confloat(ge=0.0, le=1.0)] = Field(None, description='Defines the difference with the original image, should lie in [0, 1], where 0 means almost identical, and 1 means miserable similarity')
# text_layout
@ -257,4 +258,5 @@ class RecraftReturnedObject(BaseModel):
class RecraftImageGenerationResponse(BaseModel):
created: int = Field(..., description='Unix timestamp when the generation was created')
credits: int = Field(..., description='Number of credits used for the generation')
data: list[RecraftReturnedObject] = Field(..., description=' Array of generated image information')
data: Optional[list[RecraftReturnedObject]] = Field(None, description='Array of generated image information')
image: Optional[RecraftReturnedObject] = Field(None, description='Single generated image')

25
comfy_api_nodes/apis/stability_api.py
View File

@ -51,6 +51,31 @@ class StabilityStylePreset(str, Enum):
tile_texture = "tile-texture"
class Stability_SD3_5_Model(str, Enum):
sd3_5_large = "sd3.5-large"
sd3_5_large_turbo = "sd3.5-large-turbo"
#sd3_5_medium = "sd3.5-medium"
class Stability_SD3_5_GenerationMode(str, Enum):
text_to_image = "text-to-image"
image_to_image = "image-to-image"
class StabilityStable3_5Request(BaseModel):
model: str = Field(...)
mode: str = Field(...)
prompt: str = Field(...)
negative_prompt: Optional[str] = Field(None)
aspect_ratio: Optional[str] = Field(None)
seed: Optional[int] = Field(None)
output_format: Optional[str] = Field(StabilityFormat.png.value)
image: Optional[str] = Field(None)
style_preset: Optional[str] = Field(None)
cfg_scale: float = Field(...)
strength: Optional[confloat(ge=0.0, le=1.0)] = Field(None)
class StabilityStableUltraRequest(BaseModel):
prompt: str = Field(...)
negative_prompt: Optional[str] = Field(None)

782
comfy_api_nodes/nodes_bfl.py
View File

@ -3,7 +3,12 @@ from inspect import cleandoc
from comfy.comfy_types.node_typing import IO, ComfyNodeABC
from comfy_api_nodes.apis.bfl_api import (
BFLStatus,
BFLFluxExpandImageRequest,
BFLFluxFillImageRequest,
BFLFluxCannyImageRequest,
BFLFluxDepthImageRequest,
BFLFluxProGenerateRequest,
BFLFluxProUltraGenerateRequest,
BFLFluxProGenerateResponse,
)
from comfy_api_nodes.apis.client import (
@ -25,6 +30,84 @@ import base64
import time
def convert_mask_to_image(mask: torch.Tensor):
"""
Make mask have the expected amount of dims (4) and channels (3) to be recognized as an image.
"""
mask = mask.unsqueeze(-1)
mask = torch.cat([mask]*3, dim=-1)
return mask
def handle_bfl_synchronous_operation(
operation: SynchronousOperation, timeout_bfl_calls=360
):
response_api: BFLFluxProGenerateResponse = operation.execute()
return _poll_until_generated(
response_api.polling_url, timeout=timeout_bfl_calls
)
def _poll_until_generated(polling_url: str, timeout=360):
# used bfl-comfy-nodes to verify code implementation:
# https://github.com/black-forest-labs/bfl-comfy-nodes/tree/main
start_time = time.time()
retries_404 = 0
max_retries_404 = 5
retry_404_seconds = 2
retry_202_seconds = 2
retry_pending_seconds = 1
request = requests.Request(method=HttpMethod.GET, url=polling_url)
# NOTE: should True loop be replaced with checking if workflow has been interrupted?
while True:
response = requests.Session().send(request.prepare())
if response.status_code == 200:
result = response.json()
if result["status"] == BFLStatus.ready:
img_url = result["result"]["sample"]
img_response = requests.get(img_url)
return process_image_response(img_response)
elif result["status"] in [
BFLStatus.request_moderated,
BFLStatus.content_moderated,
]:
status = result["status"]
raise Exception(
f"BFL API did not return an image due to: {status}."
)
elif result["status"] == BFLStatus.error:
raise Exception(f"BFL API encountered an error: {result}.")
elif result["status"] == BFLStatus.pending:
time.sleep(retry_pending_seconds)
continue
elif response.status_code == 404:
if retries_404 < max_retries_404:
retries_404 += 1
time.sleep(retry_404_seconds)
continue
raise Exception(
f"BFL API could not find task after {max_retries_404} tries."
)
elif response.status_code == 202:
time.sleep(retry_202_seconds)
elif time.time() - start_time > timeout:
raise Exception(
f"BFL API experienced a timeout; could not return request under {timeout} seconds."
)
else:
raise Exception(f"BFL API encountered an error: {response.json()}")
def convert_image_to_base64(image: torch.Tensor):
scaled_image = downscale_image_tensor(image, total_pixels=2048 * 2048)
# remove batch dimension if present
if len(scaled_image.shape) > 3:
scaled_image = scaled_image[0]
image_np = (scaled_image.numpy() * 255).astype(np.uint8)
img = Image.fromarray(image_np)
img_byte_arr = io.BytesIO()
img.save(img_byte_arr, format="PNG")
return base64.b64encode(img_byte_arr.getvalue()).decode()
class FluxProUltraImageNode(ComfyNodeABC):
"""
Generates images synchronously based on prompt and resolution.
@ -133,10 +216,10 @@ class FluxProUltraImageNode(ComfyNodeABC):
endpoint=ApiEndpoint(
path="/proxy/bfl/flux-pro-1.1-ultra/generate",
method=HttpMethod.POST,
request_model=BFLFluxProGenerateRequest,
request_model=BFLFluxProUltraGenerateRequest,
response_model=BFLFluxProGenerateResponse,
),
request=BFLFluxProGenerateRequest(
request=BFLFluxProUltraGenerateRequest(
prompt=prompt,
prompt_upsampling=prompt_upsampling,
seed=seed,
@ -151,7 +234,7 @@ class FluxProUltraImageNode(ComfyNodeABC):
image_prompt=(
image_prompt
if image_prompt is None
else self._convert_image_to_base64(image_prompt)
else convert_image_to_base64(image_prompt)
),
image_prompt_strength=(
None if image_prompt is None else round(image_prompt_strength, 2)
@ -159,85 +242,650 @@ class FluxProUltraImageNode(ComfyNodeABC):
),
auth_token=auth_token,
)
output_image = self._handle_bfl_synchronous_operation(operation)
output_image = handle_bfl_synchronous_operation(operation)
return (output_image,)
def _handle_bfl_synchronous_operation(
self, operation: SynchronousOperation, timeout_bfl_calls=360
class FluxProImageNode(ComfyNodeABC):
"""
Generates images synchronously based on prompt and resolution.
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation",
},
),
"prompt_upsampling": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation, but results are nondeterministic (same seed will not produce exactly the same result).",
},
),
"width": (
IO.INT,
{
"default": 1024,
"min": 256,
"max": 1440,
"step": 32,
},
),
"height": (
IO.INT,
{
"default": 768,
"min": 256,
"max": 1440,
"step": 32,
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "The random seed used for creating the noise.",
},
),
},
"optional": {
"image_prompt": (IO.IMAGE,),
# "image_prompt_strength": (
# IO.FLOAT,
# {
# "default": 0.1,
# "min": 0.0,
# "max": 1.0,
# "step": 0.01,
# "tooltip": "Blend between the prompt and the image prompt.",
# },
# ),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/bfl"
def api_call(
self,
prompt: str,
prompt_upsampling,
width: int,
height: int,
seed=0,
image_prompt=None,
# image_prompt_strength=0.1,
auth_token=None,
**kwargs,
):
response_api: BFLFluxProGenerateResponse = operation.execute()
return self._poll_until_generated(
response_api.polling_url, timeout=timeout_bfl_calls
image_prompt = (
image_prompt
if image_prompt is None
else convert_image_to_base64(image_prompt)
)
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/bfl/flux-pro-1.1/generate",
method=HttpMethod.POST,
request_model=BFLFluxProGenerateRequest,
response_model=BFLFluxProGenerateResponse,
),
request=BFLFluxProGenerateRequest(
prompt=prompt,
prompt_upsampling=prompt_upsampling,
width=width,
height=height,
seed=seed,
image_prompt=image_prompt,
),
auth_token=auth_token,
)
output_image = handle_bfl_synchronous_operation(operation)
return (output_image,)
def _poll_until_generated(self, polling_url: str, timeout=360):
# used bfl-comfy-nodes to verify code implementation:
# https://github.com/black-forest-labs/bfl-comfy-nodes/tree/main
start_time = time.time()
retries_404 = 0
max_retries_404 = 5
retry_404_seconds = 2
retry_202_seconds = 2
retry_pending_seconds = 1
request = requests.Request(method=HttpMethod.GET, url=polling_url)
# NOTE: should True loop be replaced with checking if workflow has been interrupted?
while True:
response = requests.Session().send(request.prepare())
if response.status_code == 200:
result = response.json()
if result["status"] == BFLStatus.ready:
img_url = result["result"]["sample"]
img_response = requests.get(img_url)
return process_image_response(img_response)
elif result["status"] in [
BFLStatus.request_moderated,
BFLStatus.content_moderated,
]:
status = result["status"]
raise Exception(
f"BFL API did not return an image due to: {status}."
)
elif result["status"] == BFLStatus.error:
raise Exception(f"BFL API encountered an error: {result}.")
elif result["status"] == BFLStatus.pending:
time.sleep(retry_pending_seconds)
continue
elif response.status_code == 404:
if retries_404 < max_retries_404:
retries_404 += 1
time.sleep(retry_404_seconds)
continue
raise Exception(
f"BFL API could not find task after {max_retries_404} tries."
)
elif response.status_code == 202:
time.sleep(retry_202_seconds)
elif time.time() - start_time > timeout:
raise Exception(
f"BFL API experienced a timeout; could not return request under {timeout} seconds."
)
else:
raise Exception(f"BFL API encountered an error: {response.json()}")
def _convert_image_to_base64(self, image: torch.Tensor):
scaled_image = downscale_image_tensor(image, total_pixels=2048 * 2048)
# remove batch dimension if present
if len(scaled_image.shape) > 3:
scaled_image = scaled_image[0]
image_np = (scaled_image.numpy() * 255).astype(np.uint8)
img = Image.fromarray(image_np)
img_byte_arr = io.BytesIO()
img.save(img_byte_arr, format="PNG")
return base64.b64encode(img_byte_arr.getvalue()).decode()
class FluxProExpandNode(ComfyNodeABC):
"""
Outpaints image based on prompt.
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE,),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation",
},
),
"prompt_upsampling": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation, but results are nondeterministic (same seed will not produce exactly the same result).",
},
),
"top": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 2048,
"tooltip": "Number of pixels to expand at the top of the image"
},
),
"bottom": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 2048,
"tooltip": "Number of pixels to expand at the bottom of the image"
},
),
"left": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 2048,
"tooltip": "Number of pixels to expand at the left side of the image"
},
),
"right": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 2048,
"tooltip": "Number of pixels to expand at the right side of the image"
},
),
"guidance": (
IO.FLOAT,
{
"default": 60,
"min": 1.5,
"max": 100,
"tooltip": "Guidance strength for the image generation process"
},
),
"steps": (
IO.INT,
{
"default": 50,
"min": 15,
"max": 50,
"tooltip": "Number of steps for the image generation process"
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "The random seed used for creating the noise.",
},
),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/bfl"
def api_call(
self,
image: torch.Tensor,
prompt: str,
prompt_upsampling: bool,
top: int,
bottom: int,
left: int,
right: int,
steps: int,
guidance: float,
seed=0,
auth_token=None,
**kwargs,
):
image = convert_image_to_base64(image)
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/bfl/flux-pro-1.0-expand/generate",
method=HttpMethod.POST,
request_model=BFLFluxExpandImageRequest,
response_model=BFLFluxProGenerateResponse,
),
request=BFLFluxExpandImageRequest(
prompt=prompt,
prompt_upsampling=prompt_upsampling,
top=top,
bottom=bottom,
left=left,
right=right,
steps=steps,
guidance=guidance,
seed=seed,
image=image,
),
auth_token=auth_token,
)
output_image = handle_bfl_synchronous_operation(operation)
return (output_image,)
class FluxProFillNode(ComfyNodeABC):
"""
Inpaints image based on mask and prompt.
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE,),
"mask": (IO.MASK,),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation",
},
),
"prompt_upsampling": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation, but results are nondeterministic (same seed will not produce exactly the same result).",
},
),
"guidance": (
IO.FLOAT,
{
"default": 60,
"min": 1.5,
"max": 100,
"tooltip": "Guidance strength for the image generation process"
},
),
"steps": (
IO.INT,
{
"default": 50,
"min": 15,
"max": 50,
"tooltip": "Number of steps for the image generation process"
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "The random seed used for creating the noise.",
},
),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/bfl"
def api_call(
self,
image: torch.Tensor,
mask: torch.Tensor,
prompt: str,
prompt_upsampling: bool,
steps: int,
guidance: float,
seed=0,
auth_token=None,
**kwargs,
):
# make sure image will have alpha channel removed
image = convert_image_to_base64(image[:,:,:,:3])
mask = convert_image_to_base64(convert_mask_to_image(mask))
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/bfl/flux-pro-1.0-fill/generate",
method=HttpMethod.POST,
request_model=BFLFluxFillImageRequest,
response_model=BFLFluxProGenerateResponse,
),
request=BFLFluxFillImageRequest(
prompt=prompt,
prompt_upsampling=prompt_upsampling,
steps=steps,
guidance=guidance,
seed=seed,
image=image,
mask=mask,
),
auth_token=auth_token,
)
output_image = handle_bfl_synchronous_operation(operation)
return (output_image,)
class FluxProCannyNode(ComfyNodeABC):
"""
Generate image using a control image (canny).
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"control_image": (IO.IMAGE,),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation",
},
),
"prompt_upsampling": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation, but results are nondeterministic (same seed will not produce exactly the same result).",
},
),
"canny_low_threshold": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 500,
"tooltip": "Low threshold for Canny edge detection; ignored if skip_processing is True"
},
),
"canny_high_threshold": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 500,
"tooltip": "High threshold for Canny edge detection; ignored if skip_processing is True"
},
),
"skip_preprocessing": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to skip preprocessing; set to True if control_image already is canny-fied, False if it is a raw image.",
},
),
"guidance": (
IO.FLOAT,
{
"default": 30,
"min": 1,
"max": 100,
"tooltip": "Guidance strength for the image generation process"
},
),
"steps": (
IO.INT,
{
"default": 50,
"min": 15,
"max": 50,
"tooltip": "Number of steps for the image generation process"
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "The random seed used for creating the noise.",
},
),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/bfl"
def api_call(
self,
control_image: torch.Tensor,
prompt: str,
prompt_upsampling: bool,
canny_low_threshold: int,
canny_high_threshold: int,
skip_preprocessing: bool,
steps: int,
guidance: float,
seed=0,
auth_token=None,
**kwargs,
):
control_image = convert_image_to_base64(control_image[:,:,:,:3])
preprocessed_image = None
if skip_preprocessing:
preprocessed_image = control_image
control_image = None
canny_low_threshold = None
canny_high_threshold = None
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/bfl/flux-pro-1.0-canny/generate",
method=HttpMethod.POST,
request_model=BFLFluxCannyImageRequest,
response_model=BFLFluxProGenerateResponse,
),
request=BFLFluxCannyImageRequest(
prompt=prompt,
prompt_upsampling=prompt_upsampling,
steps=steps,
guidance=guidance,
seed=seed,
control_image=control_image,
canny_low_threshold=canny_low_threshold,
canny_high_threshold=canny_high_threshold,
preprocessed_image=preprocessed_image,
),
auth_token=auth_token,
)
output_image = handle_bfl_synchronous_operation(operation)
return (output_image,)
class FluxProDepthNode(ComfyNodeABC):
"""
Generate image using a control image (depth).
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"control_image": (IO.IMAGE,),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation",
},
),
"prompt_upsampling": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to perform upsampling on the prompt. If active, automatically modifies the prompt for more creative generation, but results are nondeterministic (same seed will not produce exactly the same result).",
},
),
"skip_preprocessing": (
IO.BOOLEAN,
{
"default": False,
"tooltip": "Whether to skip preprocessing; set to True if control_image already is depth-ified, False if it is a raw image.",
},
),
"guidance": (
IO.FLOAT,
{
"default": 15,
"min": 1,
"max": 100,
"tooltip": "Guidance strength for the image generation process"
},
),
"steps": (
IO.INT,
{
"default": 50,
"min": 15,
"max": 50,
"tooltip": "Number of steps for the image generation process"
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "The random seed used for creating the noise.",
},
),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/bfl"
def api_call(
self,
control_image: torch.Tensor,
prompt: str,
prompt_upsampling: bool,
skip_preprocessing: bool,
steps: int,
guidance: float,
seed=0,
auth_token=None,
**kwargs,
):
control_image = convert_image_to_base64(control_image[:,:,:,:3])
preprocessed_image = None
if skip_preprocessing:
preprocessed_image = control_image
control_image = None
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/bfl/flux-pro-1.0-canny/generate",
method=HttpMethod.POST,
request_model=BFLFluxDepthImageRequest,
response_model=BFLFluxProGenerateResponse,
),
request=BFLFluxDepthImageRequest(
prompt=prompt,
prompt_upsampling=prompt_upsampling,
steps=steps,
guidance=guidance,
seed=seed,
control_image=control_image,
preprocessed_image=preprocessed_image,
),
auth_token=auth_token,
)
output_image = handle_bfl_synchronous_operation(operation)
return (output_image,)
# A dictionary that contains all nodes you want to export with their names
# NOTE: names should be globally unique
NODE_CLASS_MAPPINGS = {
"FluxProUltraImageNode": FluxProUltraImageNode,
# "FluxProImageNode": FluxProImageNode,
# "FluxProExpandNode": FluxProExpandNode,
# "FluxProFillNode": FluxProFillNode,
# "FluxProCannyNode": FluxProCannyNode,
# "FluxProDepthNode": FluxProDepthNode,
}
# A dictionary that contains the friendly/humanly readable titles for the nodes
NODE_DISPLAY_NAME_MAPPINGS = {
"FluxProUltraImageNode": "Flux 1.1 [pro] Ultra Image",
# "FluxProImageNode": "Flux 1.1 [pro] Image",
# "FluxProExpandNode": "Flux.1 Expand Image",
# "FluxProFillNode": "Flux.1 Fill Image",
# "FluxProCannyNode": "Flux.1 Canny Control Image",
# "FluxProDepthNode": "Flux.1 Depth Control Image",
}

575
comfy_api_nodes/nodes_recraft.py
View File

@ -1,4 +1,6 @@
from __future__ import annotations
from inspect import cleandoc
from comfy.utils import ProgressBar
from comfy.comfy_types.node_typing import IO
from comfy_api_nodes.apis.recraft_api import (
RecraftImageGenerationRequest,
@ -17,10 +19,12 @@ from comfy_api_nodes.apis.client import (
ApiEndpoint,
HttpMethod,
SynchronousOperation,
EmptyRequest,
)
from comfy_api_nodes.apinode_utils import (
bytesio_to_image_tensor,
download_url_to_bytesio,
tensor_to_bytesio,
)
import folder_paths
import json
@ -29,6 +33,50 @@ import torch
from io import BytesIO
def handle_recraft_file_request(
image: torch.Tensor,
path: str,
mask: torch.Tensor=None,
total_pixels=4096*4096,
timeout=1024,
request=None,
auth_token=None
) -> list[BytesIO]:
"""
Handle sending common Recraft file-only request to get back file bytes.
"""
if request is None:
request = EmptyRequest()
files = {
'image': tensor_to_bytesio(image, total_pixels=total_pixels).read()
}
if mask is not None:
files['mask'] = tensor_to_bytesio(mask, total_pixels=total_pixels).read()
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path=path,
method=HttpMethod.POST,
request_model=type(request),
response_model=RecraftImageGenerationResponse,
),
request=request,
files=files,
content_type="multipart/form-data",
auth_token=auth_token,
)
response: RecraftImageGenerationResponse = operation.execute()
all_bytesio = []
if response.image is not None:
all_bytesio.append(download_url_to_bytesio(response.image.url, timeout=timeout))
else:
for data in response.data:
all_bytesio.append(download_url_to_bytesio(data.url, timeout=timeout))
return all_bytesio
class SVG:
"""
Stores SVG representations via a list of BytesIO objects.
@ -36,6 +84,16 @@ class SVG:
def __init__(self, data: list[BytesIO]):
self.data = data
def combine(self, other: SVG):
return SVG(self.data + other.data)
@staticmethod
def combine_all(svgs: list[SVG]):
all_svgs = []
for svg in svgs:
all_svgs.extend(svg.data)
return SVG(all_svgs)
class SaveSVGNode:
"""
@ -349,7 +407,7 @@ class RecraftTextToImageNode:
),
request=RecraftImageGenerationRequest(
prompt=prompt,
negative_prompts=negative_prompt,
negative_prompt=negative_prompt,
model=RecraftModel.recraftv3,
size=size,
n=n,
@ -374,6 +432,243 @@ class RecraftTextToImageNode:
return (output_image,)
class RecraftImageToImageNode:
"""
Modify image based on prompt and strength.
"""
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE, ),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation.",
},
),
"n": (
IO.INT,
{
"default": 1,
"min": 1,
"max": 6,
"tooltip": "The number of images to generate.",
},
),
"strength": (
IO.FLOAT,
{
"default": 0.5,
"min": 0.0,
"max": 1.0,
"step": 0.01,
"tooltip": "Defines the difference with the original image, should lie in [0, 1], where 0 means almost identical, and 1 means miserable similarity."
}
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "Seed to determine if node should re-run; actual results are nondeterministic regardless of seed.",
},
),
},
"optional": {
"recraft_style": (RecraftIO.STYLEV3,),
"negative_prompt": (
IO.STRING,
{
"default": "",
"forceInput": True,
"tooltip": "An optional text description of undesired elements on an image.",
},
),
"recraft_controls": (
RecraftIO.CONTROLS,
{
"tooltip": "Optional additional controls over the generation via the Recraft Controls node."
},
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(
self,
image: torch.Tensor,
prompt: str,
n: int,
strength: float,
seed,
auth_token=None,
recraft_style: RecraftStyle = None,
negative_prompt: str = None,
recraft_controls: RecraftControls = None,
**kwargs,
):
default_style = RecraftStyle(RecraftStyleV3.realistic_image)
if recraft_style is None:
recraft_style = default_style
controls_api = None
if recraft_controls:
controls_api = recraft_controls.create_api_model()
if not negative_prompt:
negative_prompt = None
request = RecraftImageGenerationRequest(
prompt=prompt,
negative_prompt=negative_prompt,
model=RecraftModel.recraftv3,
n=n,
strength=round(strength, 2),
style=recraft_style.style,
substyle=recraft_style.substyle,
style_id=recraft_style.style_id,
controls=controls_api,
)
images = []
total = image.shape[0]
pbar = ProgressBar(total)
for i in range(total):
sub_bytes = handle_recraft_file_request(
image=image[i],
path="/proxy/recraft/images/imageToImage",
request=request,
auth_token=auth_token,
)
images.append(torch.cat([bytesio_to_image_tensor(x) for x in sub_bytes], dim=0))
pbar.update(1)
images_tensor = torch.cat(images, dim=0)
return (images_tensor, )
class RecraftImageInpaintingNode:
"""
Modify image based on prompt and mask.
"""
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE, ),
"mask": (IO.MASK, ),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation.",
},
),
"n": (
IO.INT,
{
"default": 1,
"min": 1,
"max": 6,
"tooltip": "The number of images to generate.",
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "Seed to determine if node should re-run; actual results are nondeterministic regardless of seed.",
},
),
},
"optional": {
"recraft_style": (RecraftIO.STYLEV3,),
"negative_prompt": (
IO.STRING,
{
"default": "",
"forceInput": True,
"tooltip": "An optional text description of undesired elements on an image.",
},
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(
self,
image: torch.Tensor,
mask: torch.Tensor,
prompt: str,
n: int,
seed,
auth_token=None,
recraft_style: RecraftStyle = None,
negative_prompt: str = None,
**kwargs,
):
default_style = RecraftStyle(RecraftStyleV3.realistic_image)
if recraft_style is None:
recraft_style = default_style
if not negative_prompt:
negative_prompt = None
request = RecraftImageGenerationRequest(
prompt=prompt,
negative_prompt=negative_prompt,
model=RecraftModel.recraftv3,
n=n,
style=recraft_style.style,
substyle=recraft_style.substyle,
style_id=recraft_style.style_id,
)
images = []
total = image.shape[0]
pbar = ProgressBar(total)
for i in range(total):
sub_bytes = handle_recraft_file_request(
image=image[i],
mask=mask[i:i+1],
path="/proxy/recraft/images/imageInpainting",
request=request,
auth_token=auth_token,
)
images.append(torch.cat([bytesio_to_image_tensor(x) for x in sub_bytes], dim=0))
pbar.update(1)
images_tensor = torch.cat(images, dim=0)
return (images_tensor, )
class RecraftTextToVectorNode:
"""
Generates SVG synchronously based on prompt and resolution.
@ -477,7 +772,7 @@ class RecraftTextToVectorNode:
),
request=RecraftImageGenerationRequest(
prompt=prompt,
negative_prompts=negative_prompt,
negative_prompt=negative_prompt,
model=RecraftModel.recraftv3,
size=size,
n=n,
@ -495,11 +790,280 @@ class RecraftTextToVectorNode:
return (SVG(svg_data),)
class RecraftVectorizeImageNode:
"""
Generates SVG synchronously from an input image.
"""
RETURN_TYPES = (RecraftIO.SVG,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE, ),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(
self,
image: torch.Tensor,
auth_token=None,
**kwargs,
):
svgs = []
total = image.shape[0]
pbar = ProgressBar(total)
for i in range(total):
sub_bytes = handle_recraft_file_request(
image=image[i],
path="/proxy/recraft/images/vectorize",
auth_token=auth_token,
)
svgs.append(SVG(sub_bytes))
pbar.update(1)
return (SVG.combine_all(svgs), )
class RecraftReplaceBackgroundNode:
"""
Replace background on image, based on provided prompt.
"""
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE, ),
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Prompt for the image generation.",
},
),
"n": (
IO.INT,
{
"default": 1,
"min": 1,
"max": 6,
"tooltip": "The number of images to generate.",
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF,
"control_after_generate": True,
"tooltip": "Seed to determine if node should re-run; actual results are nondeterministic regardless of seed.",
},
),
},
"optional": {
"recraft_style": (RecraftIO.STYLEV3,),
"negative_prompt": (
IO.STRING,
{
"default": "",
"forceInput": True,
"tooltip": "An optional text description of undesired elements on an image.",
},
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(
self,
image: torch.Tensor,
prompt: str,
n: int,
seed,
auth_token=None,
recraft_style: RecraftStyle = None,
negative_prompt: str = None,
**kwargs,
):
default_style = RecraftStyle(RecraftStyleV3.realistic_image)
if recraft_style is None:
recraft_style = default_style
if not negative_prompt:
negative_prompt = None
request = RecraftImageGenerationRequest(
prompt=prompt,
negative_prompt=negative_prompt,
model=RecraftModel.recraftv3,
n=n,
style=recraft_style.style,
substyle=recraft_style.substyle,
style_id=recraft_style.style_id,
)
images = []
total = image.shape[0]
pbar = ProgressBar(total)
for i in range(total):
sub_bytes = handle_recraft_file_request(
image=image[i],
path="/proxy/recraft/images/replaceBackground",
request=request,
auth_token=auth_token,
)
images.append(torch.cat([bytesio_to_image_tensor(x) for x in sub_bytes], dim=0))
pbar.update(1)
images_tensor = torch.cat(images, dim=0)
return (images_tensor, )
class RecraftRemoveBackgroundNode:
"""
Remove background from image, and return processed image and mask.
"""
RETURN_TYPES = (IO.IMAGE, IO.MASK)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE, ),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(
self,
image: torch.Tensor,
auth_token=None,
**kwargs,
):
images = []
total = image.shape[0]
pbar = ProgressBar(total)
for i in range(total):
sub_bytes = handle_recraft_file_request(
image=image[i],
path="/proxy/recraft/images/removeBackground",
auth_token=auth_token,
)
images.append(torch.cat([bytesio_to_image_tensor(x) for x in sub_bytes], dim=0))
pbar.update(1)
images_tensor = torch.cat(images, dim=0)
# use alpha channel as masks, in B,H,W format
masks_tensor = images_tensor[:,:,:,-1:].squeeze(-1)
return (images_tensor, masks_tensor)
class RecraftCrispUpscaleNode:
"""
Upscale image synchronously.
Enhances a given raster image using ‘crisp upscale’ tool, increasing image resolution, making the image sharper and cleaner.
"""
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
RECRAFT_PATH = "/proxy/recraft/images/crispUpscale"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": (IO.IMAGE, ),
},
"optional": {
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(
self,
image: torch.Tensor,
auth_token=None,
**kwargs,
):
images = []
total = image.shape[0]
pbar = ProgressBar(total)
for i in range(total):
sub_bytes = handle_recraft_file_request(
image=image[i],
path=self.RECRAFT_PATH,
auth_token=auth_token,
)
images.append(torch.cat([bytesio_to_image_tensor(x) for x in sub_bytes], dim=0))
pbar.update(1)
images_tensor = torch.cat(images, dim=0)
return (images_tensor,)
class RecraftCreativeUpscaleNode(RecraftCrispUpscaleNode):
"""
Upscale image synchronously.
Enhances a given raster image using ‘creative upscale’ tool, boosting resolution with a focus on refining small details and faces.
"""
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/Recraft"
RECRAFT_PATH = "/proxy/recraft/images/creativeUpscale"
# A dictionary that contains all nodes you want to export with their names
# NOTE: names should be globally unique
NODE_CLASS_MAPPINGS = {
"RecraftTextToImageNode": RecraftTextToImageNode,
# "RecraftImageToImageNode": RecraftImageToImageNode,
# "RecraftImageInpaintingNode": RecraftImageInpaintingNode,
"RecraftTextToVectorNode": RecraftTextToVectorNode,
"RecraftVectorizeImageNode": RecraftVectorizeImageNode,
"RecraftRemoveBackgroundNode": RecraftRemoveBackgroundNode,
# "RecraftReplaceBackgroundNode": RecraftReplaceBackgroundNode,
"RecraftCrispUpscaleNode": RecraftCrispUpscaleNode,
# "RecraftCreativeUpscaleNode": RecraftCreativeUpscaleNode,
"RecraftStyleV3RealisticImage": RecraftStyleV3RealisticImageNode,
"RecraftStyleV3DigitalIllustration": RecraftStyleV3DigitalIllustrationNode,
"RecraftStyleV3LogoRaster": RecraftStyleV3LogoRasterNode,
@ -511,7 +1075,14 @@ NODE_CLASS_MAPPINGS = {
# A dictionary that contains the friendly/humanly readable titles for the nodes
NODE_DISPLAY_NAME_MAPPINGS = {
"RecraftTextToImageNode": "Recraft Text to Image",
# "RecraftImageToImageNode": "Recraft Image to Image",
# "RecraftImageInpaintingNode": "Recraft Image Inpainting",
"RecraftTextToVectorNode": "Recraft Text to Vector",
"RecraftVectorizeImageNode": "Recraft Vectorize Image",
"RecraftRemoveBackgroundNode": "Recraft Remove Background",
# "RecraftReplaceBackgroundNode": "Recraft Replace Background",
"RecraftCrispUpscaleNode": "Recraft Crisp Upscale Image",
# "RecraftCreativeUpscaleNode": "Recraft Creative Upscale Image",
"RecraftStyleV3RealisticImage": "Recraft Style - Realistic Image",
"RecraftStyleV3DigitalIllustration": "Recraft Style - Digital Illustration",
"RecraftStyleV3LogoRaster": "Recraft Style - Logo Raster",

141
comfy_api_nodes/nodes_stability.py
View File

@ -1,9 +1,12 @@
from inspect import cleandoc
from comfy.comfy_types.node_typing import IO
from comfy_api_nodes.apis.stability_api import (
StabilityStable3_5Request,
StabilityStableUltraRequest,
StabilityStableUltraResponse,
StabilityAspectRatio,
Stability_SD3_5_Model,
Stability_SD3_5_GenerationMode,
get_stability_style_presets,
)
from comfy_api_nodes.apis.client import (
@ -148,13 +151,151 @@ class StabilityStableImageUltraNode:
return (returned_image,)
class StabilityStableImageSD_3_5Node:
"""
Generates images synchronously based on prompt and resolution.
"""
RETURN_TYPES = (IO.IMAGE,)
DESCRIPTION = cleandoc(__doc__ or "") # Handle potential None value
FUNCTION = "api_call"
API_NODE = True
CATEGORY = "api node/image/stability"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "What you wish to see in the output image. A strong, descriptive prompt that clearly defines elements, colors, and subjects will lead to better results."
},
),
"aspect_ratio": ([x.value for x in StabilityAspectRatio],
{
"default": StabilityAspectRatio.ratio_1_1,
"tooltip": "Aspect ratio of generated image.",
},
),
"style_preset": (get_stability_style_presets(),
{
"tooltip": "Optional desired style of generated image.",
},
),
"cfg_scale": (
IO.FLOAT,
{
"default": 4.0,
"min": 1.0,
"max": 10.0,
"step": 0.1,
"tooltip": "How strictly the diffusion process adheres to the prompt text (higher values keep your image closer to your prompt)",
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 4294967294,
"control_after_generate": True,
"tooltip": "The random seed used for creating the noise.",
},
),
},
"optional": {
"image": (IO.IMAGE,),
"negative_prompt": (
IO.STRING,
{
"default": "",
"forceInput": True,
"tooltip": "Keywords of what you do not wish to see in the output image. This is an advanced feature."
},
),
"image_denoise": (
IO.FLOAT,
{
"default": 0.5,
"min": 0.0,
"max": 1.0,
"step": 0.01,
"tooltip": "Denoise of input image; 0.0 yields image identical to input, 1.0 is as if no image was provided at all.",
},
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
def api_call(self, prompt: str, aspect_ratio: str, style_preset: str, seed: int, cfg_scale: float,
negative_prompt: str=None, image: torch.Tensor = None, image_denoise: float=None,
auth_token=None):
model = Stability_SD3_5_Model.sd3_5_large.value
# prepare image binary if image present
image_binary = None
mode = Stability_SD3_5_GenerationMode.text_to_image.value
if image is not None:
image_binary = tensor_to_bytesio(image, 1504 * 1504).read()
mode = Stability_SD3_5_GenerationMode.image_to_image.value
else:
image_denoise = None
if not negative_prompt:
negative_prompt = None
if style_preset == "None":
style_preset = None
files = {
"image": image_binary
}
operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/stability/v2beta/stable-image/generate/sd3",
method=HttpMethod.POST,
request_model=StabilityStable3_5Request,
response_model=StabilityStableUltraResponse,
),
request=StabilityStable3_5Request(
prompt=prompt,
negative_prompt=negative_prompt,
aspect_ratio=aspect_ratio,
seed=seed,
strength=image_denoise,
style_preset=style_preset,
cfg_scale=cfg_scale,
model=model,
mode=mode,
),
files=files,
content_type="multipart/form-data",
auth_token=auth_token,
)
response_api = operation.execute()
if response_api.finish_reason != "SUCCESS":
raise Exception(f"Stable Diffusion 3.5 Image generation failed: {response_api.finish_reason}.")
image_data = base64.b64decode(response_api.image)
returned_image = bytesio_to_image_tensor(BytesIO(image_data))
return (returned_image,)
# A dictionary that contains all nodes you want to export with their names
# NOTE: names should be globally unique
NODE_CLASS_MAPPINGS = {
"StabilityStableImageUltraNode": StabilityStableImageUltraNode,
# "StabilityStableImageSD_3_5Node": StabilityStableImageSD_3_5Node,
}
# A dictionary that contains the friendly/humanly readable titles for the nodes
NODE_DISPLAY_NAME_MAPPINGS = {
"StabilityStableImageUltraNode": "Stability Stable Image Ultra",
# "StabilityStableImageSD_3_5Node": "Stability Stable Diffusion 3.5 Image",
}