1094 lines
41 KiB
Python
1094 lines
41 KiB
Python
import re
|
|
|
|
import torch
|
|
from typing_extensions import override
|
|
|
|
from comfy_api.latest import IO, ComfyExtension, Input
|
|
from comfy_api_nodes.apis.grok import (
|
|
ImageEditRequest,
|
|
ImageGenerationRequest,
|
|
ImageGenerationResponse,
|
|
InputUrlObject,
|
|
VideoEditRequest,
|
|
VideoExtensionRequest,
|
|
VideoGenerationRequest,
|
|
VideoGenerationResponse,
|
|
VideoStatusResponse,
|
|
VoiceReferenceObject,
|
|
)
|
|
from comfy_api_nodes.util import (
|
|
ApiEndpoint,
|
|
download_url_to_image_tensor,
|
|
download_url_to_video_output,
|
|
get_fs_object_size,
|
|
get_number_of_images,
|
|
poll_op,
|
|
sync_op,
|
|
tensor_to_base64_string,
|
|
upload_images_to_comfyapi,
|
|
upload_video_to_comfyapi,
|
|
validate_string,
|
|
validate_video_duration,
|
|
)
|
|
|
|
|
|
_GROK_VIDEO_MODEL_API_IDS = {
|
|
"grok-imagine-video-1.5": "grok-imagine-video-1.5",
|
|
}
|
|
|
|
_GROK_VOICE_OPTIONS = [
|
|
"none",
|
|
"ara",
|
|
"eve",
|
|
"leo",
|
|
"rex",
|
|
"sal",
|
|
"carina",
|
|
"zagan",
|
|
"helix",
|
|
"orion",
|
|
"luna",
|
|
"iris",
|
|
"altair",
|
|
"zenith",
|
|
"perseus",
|
|
"helios",
|
|
"lux",
|
|
"kepler",
|
|
"rigel",
|
|
"cosmo",
|
|
"celeste",
|
|
"ursa",
|
|
"sirius",
|
|
"lumen",
|
|
"castor",
|
|
"naksh",
|
|
"atlas",
|
|
]
|
|
|
|
|
|
_GROK_REF_TAG_RE = re.compile(r"(?<!\w)@(image|audio)(?P<idx>\d*)(?!\w)", re.IGNORECASE | re.ASCII)
|
|
|
|
|
|
def _normalize_grok_reference_prompt(prompt: str, total_images: int, voices: list[str]) -> str:
|
|
"""Rewrite @Image1/@Audio1 style references (1-based, shared partner-node syntax)
|
|
into Grok's native <IMAGE_0>/<AUDIO_0> tags; an unnumbered @image/@audio means the first one.
|
|
Native tags pass through untouched. @ImageN refers to the Nth reference image overall, in
|
|
input order — a batched input contributes one number per image. @AudioN refers to the
|
|
'voice_N' widget; the API only accepts compact arrays, so voices are remapped to array
|
|
positions and 'none' slots between selected voices are harmless. Substitution repeats until
|
|
stable so adjacent tags like '@Image1@Image2' all resolve."""
|
|
audio_indices: dict[int, int] = {}
|
|
for slot, voice in enumerate(voices, start=1):
|
|
if voice != "none":
|
|
audio_indices[slot] = len(audio_indices)
|
|
|
|
def repl(match: re.Match) -> str:
|
|
kind = match.group(1).lower()
|
|
idx = int(match.group("idx") or 1)
|
|
if kind == "image":
|
|
if not 1 <= idx <= total_images:
|
|
raise ValueError(
|
|
f"The prompt references @Image{idx}, but only {total_images} "
|
|
f"reference images are connected (a batched input counts once per image)."
|
|
)
|
|
return f"<IMAGE_{idx - 1}>"
|
|
if idx not in audio_indices:
|
|
if 1 <= idx <= len(voices):
|
|
raise ValueError(f"The prompt references @Audio{idx}, but 'voice_{idx}' is set to 'none'.")
|
|
raise ValueError(f"The prompt references @Audio{idx}, but only voices 1..{len(voices)} exist.")
|
|
return f"<AUDIO_{audio_indices[idx]}>"
|
|
|
|
prev = None
|
|
while prev != prompt:
|
|
prev = prompt
|
|
prompt = _GROK_REF_TAG_RE.sub(repl, prompt)
|
|
return prompt
|
|
|
|
|
|
def _extract_grok_price(response) -> float | None:
|
|
if response.usage and response.usage.cost_in_usd_ticks is not None:
|
|
return response.usage.cost_in_usd_ticks / 10_000_000_000
|
|
return None
|
|
|
|
|
|
def _extract_grok_video_price(response) -> float | None:
|
|
price = _extract_grok_price(response)
|
|
if price is not None:
|
|
return price * 1.43
|
|
return None
|
|
|
|
|
|
class GrokImageNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokImageNode",
|
|
display_name="Grok Image",
|
|
category="partner/image/Grok",
|
|
description="Generate images using Grok based on a text prompt",
|
|
inputs=[
|
|
IO.Combo.Input(
|
|
"model",
|
|
options=[
|
|
"grok-imagine-image-quality",
|
|
"grok-imagine-image-pro",
|
|
"grok-imagine-image",
|
|
],
|
|
),
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
tooltip="The text prompt used to generate the image",
|
|
),
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=[
|
|
"1:1",
|
|
"2:3",
|
|
"3:2",
|
|
"3:4",
|
|
"4:3",
|
|
"9:16",
|
|
"16:9",
|
|
"9:19.5",
|
|
"19.5:9",
|
|
"9:20",
|
|
"20:9",
|
|
"1:2",
|
|
"2:1",
|
|
],
|
|
),
|
|
IO.Int.Input(
|
|
"number_of_images",
|
|
default=1,
|
|
min=1,
|
|
max=10,
|
|
step=1,
|
|
tooltip="Number of images to generate",
|
|
display_mode=IO.NumberDisplay.number,
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
IO.Combo.Input("resolution", options=["1K", "2K"], optional=True),
|
|
],
|
|
outputs=[
|
|
IO.Image.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
depends_on=IO.PriceBadgeDepends(widgets=["model", "number_of_images", "resolution"]),
|
|
expr="""
|
|
(
|
|
$rate := widgets.model = "grok-imagine-image-quality"
|
|
? (widgets.resolution = "1k" ? 0.05 : 0.07)
|
|
: ($contains(widgets.model, "pro") ? 0.07 : 0.02);
|
|
{"type":"usd","usd": $rate * widgets.number_of_images}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
model: str,
|
|
prompt: str,
|
|
aspect_ratio: str,
|
|
number_of_images: int,
|
|
seed: int,
|
|
resolution: str = "1K",
|
|
) -> IO.NodeOutput:
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/images/generations", method="POST"),
|
|
data=ImageGenerationRequest(
|
|
model=model,
|
|
prompt=prompt,
|
|
aspect_ratio=aspect_ratio,
|
|
n=number_of_images,
|
|
seed=seed,
|
|
resolution=resolution.lower(),
|
|
),
|
|
response_model=ImageGenerationResponse,
|
|
)
|
|
if len(response.data) == 1:
|
|
return IO.NodeOutput(await download_url_to_image_tensor(response.data[0].url))
|
|
return IO.NodeOutput(
|
|
torch.cat(
|
|
[await download_url_to_image_tensor(i) for i in [str(d.url) for d in response.data if d.url]],
|
|
)
|
|
)
|
|
|
|
|
|
_GROK_IMAGE_EDIT_ASPECT_RATIO_OPTIONS = [
|
|
"auto",
|
|
"1:1",
|
|
"2:3",
|
|
"3:2",
|
|
"3:4",
|
|
"4:3",
|
|
"9:16",
|
|
"16:9",
|
|
"9:19.5",
|
|
"19.5:9",
|
|
"9:20",
|
|
"20:9",
|
|
"1:2",
|
|
"2:1",
|
|
]
|
|
|
|
|
|
def _grok_image_edit_model_inputs(*, max_ref_images: int, with_aspect_ratio: bool):
|
|
inputs = [
|
|
IO.Autogrow.Input(
|
|
"images",
|
|
template=IO.Autogrow.TemplateNames(
|
|
IO.Image.Input("image"),
|
|
names=[f"image_{i}" for i in range(1, max_ref_images + 1)],
|
|
min=1,
|
|
),
|
|
tooltip=(
|
|
"Reference image to edit."
|
|
if max_ref_images == 1
|
|
else f"Reference image(s) to edit. Up to {max_ref_images} images."
|
|
),
|
|
),
|
|
IO.Combo.Input("resolution", options=["1K", "2K"]),
|
|
IO.Int.Input(
|
|
"number_of_images",
|
|
default=1,
|
|
min=1,
|
|
max=10,
|
|
step=1,
|
|
tooltip="Number of edited images to generate",
|
|
display_mode=IO.NumberDisplay.number,
|
|
),
|
|
]
|
|
if with_aspect_ratio:
|
|
inputs.append(
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=_GROK_IMAGE_EDIT_ASPECT_RATIO_OPTIONS,
|
|
tooltip="Only allowed when multiple images are connected.",
|
|
)
|
|
)
|
|
return inputs
|
|
|
|
|
|
class GrokImageEditNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokImageEditNode",
|
|
display_name="Grok Image Edit",
|
|
category="partner/image/Grok",
|
|
description="Modify an existing image based on a text prompt",
|
|
inputs=[
|
|
IO.Combo.Input(
|
|
"model",
|
|
options=[
|
|
"grok-imagine-image-quality",
|
|
"grok-imagine-image-pro",
|
|
"grok-imagine-image",
|
|
],
|
|
),
|
|
IO.Image.Input("image", display_name="images"),
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
tooltip="The text prompt used to generate the image",
|
|
),
|
|
IO.Combo.Input("resolution", options=["1K", "2K"]),
|
|
IO.Int.Input(
|
|
"number_of_images",
|
|
default=1,
|
|
min=1,
|
|
max=10,
|
|
step=1,
|
|
tooltip="Number of edited images to generate",
|
|
display_mode=IO.NumberDisplay.number,
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=[
|
|
"auto",
|
|
"1:1",
|
|
"2:3",
|
|
"3:2",
|
|
"3:4",
|
|
"4:3",
|
|
"9:16",
|
|
"16:9",
|
|
"9:19.5",
|
|
"19.5:9",
|
|
"9:20",
|
|
"20:9",
|
|
"1:2",
|
|
"2:1",
|
|
],
|
|
optional=True,
|
|
tooltip="Only allowed when multiple images are connected to the image input.",
|
|
),
|
|
],
|
|
outputs=[
|
|
IO.Image.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
depends_on=IO.PriceBadgeDepends(widgets=["model", "number_of_images", "resolution"]),
|
|
expr="""
|
|
(
|
|
$isQualityModel := widgets.model = "grok-imagine-image-quality";
|
|
$isPro := $contains(widgets.model, "pro");
|
|
$rate := $isQualityModel
|
|
? (widgets.resolution = "1k" ? 0.05 : 0.07)
|
|
: ($isPro ? 0.07 : 0.02);
|
|
$base := $isQualityModel ? 0.01 : 0.002;
|
|
$output := $rate * widgets.number_of_images;
|
|
$isPro
|
|
? {"type":"usd","usd": $base + $output}
|
|
: {"type":"range_usd","min_usd": $base + $output, "max_usd": 3 * $base + $output}
|
|
)
|
|
""",
|
|
),
|
|
is_deprecated=True,
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
model: str,
|
|
image: Input.Image,
|
|
prompt: str,
|
|
resolution: str,
|
|
number_of_images: int,
|
|
seed: int,
|
|
aspect_ratio: str = "auto",
|
|
) -> IO.NodeOutput:
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
if model == "grok-imagine-image-pro":
|
|
if get_number_of_images(image) > 1:
|
|
raise ValueError("The pro model supports only 1 input image.")
|
|
elif get_number_of_images(image) > 3:
|
|
raise ValueError("A maximum of 3 input images is supported.")
|
|
if aspect_ratio != "auto" and get_number_of_images(image) == 1:
|
|
raise ValueError(
|
|
"Custom aspect ratio is only allowed when multiple images are connected to the image input."
|
|
)
|
|
response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/images/edits", method="POST"),
|
|
data=ImageEditRequest(
|
|
model=model,
|
|
images=[InputUrlObject(url=f"data:image/png;base64,{tensor_to_base64_string(i)}") for i in image],
|
|
prompt=prompt,
|
|
resolution=resolution.lower(),
|
|
n=number_of_images,
|
|
seed=seed,
|
|
aspect_ratio=None if aspect_ratio == "auto" else aspect_ratio,
|
|
),
|
|
response_model=ImageGenerationResponse,
|
|
)
|
|
if len(response.data) == 1:
|
|
return IO.NodeOutput(await download_url_to_image_tensor(response.data[0].url))
|
|
return IO.NodeOutput(
|
|
torch.cat(
|
|
[await download_url_to_image_tensor(i) for i in [str(d.url) for d in response.data if d.url]],
|
|
)
|
|
)
|
|
|
|
|
|
class GrokImageEditNodeV2(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokImageEditNodeV2",
|
|
display_name="Grok Image Edit",
|
|
category="partner/image/Grok",
|
|
description="Modify an existing image based on a text prompt",
|
|
inputs=[
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
default="",
|
|
tooltip="The text prompt used to generate the image",
|
|
),
|
|
IO.DynamicCombo.Input(
|
|
"model",
|
|
options=[
|
|
IO.DynamicCombo.Option(
|
|
"grok-imagine-image-quality",
|
|
_grok_image_edit_model_inputs(max_ref_images=3, with_aspect_ratio=True),
|
|
),
|
|
IO.DynamicCombo.Option(
|
|
"grok-imagine-image-pro",
|
|
_grok_image_edit_model_inputs(max_ref_images=1, with_aspect_ratio=False),
|
|
),
|
|
IO.DynamicCombo.Option(
|
|
"grok-imagine-image",
|
|
_grok_image_edit_model_inputs(max_ref_images=3, with_aspect_ratio=True),
|
|
),
|
|
],
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
],
|
|
outputs=[
|
|
IO.Image.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
depends_on=IO.PriceBadgeDepends(
|
|
widgets=["model", "model.resolution", "model.number_of_images"],
|
|
),
|
|
expr="""
|
|
(
|
|
$isQualityModel := widgets.model = "grok-imagine-image-quality";
|
|
$isPro := $contains(widgets.model, "pro");
|
|
$res := $lookup(widgets, "model.resolution");
|
|
$n := $lookup(widgets, "model.number_of_images");
|
|
$rate := $isQualityModel
|
|
? ($res = "1k" ? 0.05 : 0.07)
|
|
: ($isPro ? 0.07 : 0.02);
|
|
$base := $isQualityModel ? 0.01 : 0.002;
|
|
$output := $rate * $n;
|
|
$isPro
|
|
? {"type":"usd","usd": $base + $output}
|
|
: {"type":"range_usd","min_usd": $base + $output, "max_usd": 3 * $base + $output}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
prompt: str,
|
|
model: dict,
|
|
seed: int,
|
|
) -> IO.NodeOutput:
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
model_id = model["model"]
|
|
resolution = model["resolution"]
|
|
number_of_images = model["number_of_images"]
|
|
images_dict = model.get("images") or {}
|
|
aspect_ratio = model.get("aspect_ratio", "auto")
|
|
|
|
image_tensors: list[Input.Image] = [t for t in images_dict.values() if t is not None]
|
|
n_images = sum(get_number_of_images(t) for t in image_tensors)
|
|
if n_images < 1:
|
|
raise ValueError("At least one image is required for editing.")
|
|
if model_id == "grok-imagine-image-pro" and n_images > 1:
|
|
raise ValueError("The pro model supports only 1 input image.")
|
|
if model_id != "grok-imagine-image-pro" and n_images > 3:
|
|
raise ValueError("A maximum of 3 input images is supported.")
|
|
if aspect_ratio != "auto" and n_images == 1:
|
|
raise ValueError(
|
|
"Custom aspect ratio is only allowed when multiple images are connected to the image input."
|
|
)
|
|
|
|
flat_tensors: list[torch.Tensor] = []
|
|
for tensor in image_tensors:
|
|
if len(tensor.shape) == 4:
|
|
flat_tensors.extend(tensor[i] for i in range(tensor.shape[0]))
|
|
else:
|
|
flat_tensors.append(tensor)
|
|
|
|
response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/images/edits", method="POST"),
|
|
data=ImageEditRequest(
|
|
model=model_id,
|
|
images=[
|
|
InputUrlObject(url=f"data:image/png;base64,{tensor_to_base64_string(i)}") for i in flat_tensors
|
|
],
|
|
prompt=prompt,
|
|
resolution=resolution.lower(),
|
|
n=number_of_images,
|
|
seed=seed,
|
|
aspect_ratio=None if aspect_ratio == "auto" else aspect_ratio,
|
|
),
|
|
response_model=ImageGenerationResponse,
|
|
)
|
|
if len(response.data) == 1:
|
|
return IO.NodeOutput(await download_url_to_image_tensor(response.data[0].url))
|
|
return IO.NodeOutput(
|
|
torch.cat(
|
|
[await download_url_to_image_tensor(i) for i in [str(d.url) for d in response.data if d.url]],
|
|
)
|
|
)
|
|
|
|
|
|
class GrokVideoNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokVideoNode",
|
|
display_name="Grok Video",
|
|
category="partner/video/Grok",
|
|
description="Generate video from a prompt or an image",
|
|
inputs=[
|
|
IO.Combo.Input(
|
|
"model",
|
|
options=["grok-imagine-video", "grok-imagine-video-1.5"],
|
|
tooltip="The model to use for video generation.",
|
|
),
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
tooltip="Text description of the desired video. "
|
|
"Optional for grok-imagine-video-1.5 when an input image is provided.",
|
|
),
|
|
IO.Combo.Input(
|
|
"resolution",
|
|
options=["480p", "720p", "1080p"],
|
|
tooltip="The resolution of the output video. 1080p is only available for grok-imagine-video-1.5.",
|
|
),
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=["auto", "16:9", "4:3", "3:2", "1:1", "2:3", "3:4", "9:16"],
|
|
tooltip="The aspect ratio of the output video.",
|
|
),
|
|
IO.Int.Input(
|
|
"duration",
|
|
default=6,
|
|
min=1,
|
|
max=15,
|
|
step=1,
|
|
tooltip="The duration of the output video in seconds.",
|
|
display_mode=IO.NumberDisplay.slider,
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
IO.Image.Input(
|
|
"image",
|
|
optional=True,
|
|
tooltip="Optional starting image. If omitted, the video is generated from the text prompt alone.",
|
|
),
|
|
],
|
|
outputs=[
|
|
IO.Video.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
depends_on=IO.PriceBadgeDepends(widgets=["model", "duration", "resolution"], inputs=["image"]),
|
|
expr="""
|
|
(
|
|
$is15 := $contains(widgets.model, "1.5");
|
|
$rate := $is15
|
|
? (widgets.resolution = "1080p" ? 0.25 : (widgets.resolution = "720p" ? 0.14 : 0.08))
|
|
: (widgets.resolution = "720p" ? 0.07 : 0.05);
|
|
$imgCost := $is15 ? 0.01 : 0.002;
|
|
$base := $rate * widgets.duration;
|
|
$total := inputs.image.connected ? $base + $imgCost : $base;
|
|
{"type":"usd","usd": $is15 ? $total * 1.43 : $total}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
model: str,
|
|
prompt: str,
|
|
resolution: str,
|
|
aspect_ratio: str,
|
|
duration: int,
|
|
seed: int,
|
|
image: Input.Image | None = None,
|
|
) -> IO.NodeOutput:
|
|
if resolution == "1080p" and model != "grok-imagine-video-1.5":
|
|
raise ValueError(f"1080p resolution is only available for grok-imagine-video-1.5, not '{model}'.")
|
|
image_url = None
|
|
if image is not None:
|
|
if get_number_of_images(image) != 1:
|
|
raise ValueError("Only one input image is supported.")
|
|
image_url = InputUrlObject(url=f"data:image/png;base64,{tensor_to_base64_string(image)}")
|
|
if image is None or model != "grok-imagine-video-1.5":
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
initial_response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/videos/generations", method="POST"),
|
|
data=VideoGenerationRequest(
|
|
model=_GROK_VIDEO_MODEL_API_IDS.get(model, model),
|
|
image=image_url,
|
|
prompt=prompt,
|
|
resolution=resolution,
|
|
duration=duration,
|
|
aspect_ratio=None if aspect_ratio == "auto" else aspect_ratio,
|
|
seed=seed,
|
|
),
|
|
response_model=VideoGenerationResponse,
|
|
)
|
|
response = await poll_op(
|
|
cls,
|
|
ApiEndpoint(path=f"/proxy/xai/v1/videos/{initial_response.request_id}"),
|
|
status_extractor=lambda r: r.status if r.status is not None else "complete",
|
|
response_model=VideoStatusResponse,
|
|
price_extractor=_extract_grok_video_price if model == "grok-imagine-video-1.5" else _extract_grok_price,
|
|
)
|
|
return IO.NodeOutput(await download_url_to_video_output(response.video.url))
|
|
|
|
|
|
class GrokVideoEditNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokVideoEditNode",
|
|
display_name="Grok Video Edit",
|
|
category="partner/video/Grok",
|
|
description="Edit an existing video based on a text prompt.",
|
|
inputs=[
|
|
IO.Combo.Input("model", options=["grok-imagine-video"]),
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
tooltip="Text description of the desired video.",
|
|
),
|
|
IO.Video.Input("video", tooltip="Maximum supported duration is 8.7 seconds and 50MB file size."),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
],
|
|
outputs=[
|
|
IO.Video.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
expr="""{"type":"usd","usd": 0.06, "format": {"suffix": "/sec", "approximate": true}}""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
model: str,
|
|
prompt: str,
|
|
video: Input.Video,
|
|
seed: int,
|
|
) -> IO.NodeOutput:
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
validate_video_duration(video, min_duration=1, max_duration=8.7)
|
|
video_stream = video.get_stream_source()
|
|
video_size = get_fs_object_size(video_stream)
|
|
if video_size > 50 * 1024 * 1024:
|
|
raise ValueError(f"Video size ({video_size / 1024 / 1024:.1f}MB) exceeds 50MB limit.")
|
|
initial_response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/videos/edits", method="POST"),
|
|
data=VideoEditRequest(
|
|
model=model,
|
|
video=InputUrlObject(url=await upload_video_to_comfyapi(cls, video)),
|
|
prompt=prompt,
|
|
seed=seed,
|
|
),
|
|
response_model=VideoGenerationResponse,
|
|
)
|
|
response = await poll_op(
|
|
cls,
|
|
ApiEndpoint(path=f"/proxy/xai/v1/videos/{initial_response.request_id}"),
|
|
status_extractor=lambda r: r.status if r.status is not None else "complete",
|
|
response_model=VideoStatusResponse,
|
|
price_extractor=_extract_grok_price,
|
|
)
|
|
return IO.NodeOutput(await download_url_to_video_output(response.video.url))
|
|
|
|
|
|
class GrokVideoReferenceNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokVideoReferenceNode",
|
|
display_name="Grok Reference-to-Video",
|
|
category="partner/video/Grok",
|
|
description="Generate video guided by reference images, with optional preset voice references.",
|
|
inputs=[
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
tooltip="Text description of the desired video.",
|
|
),
|
|
IO.DynamicCombo.Input(
|
|
"model",
|
|
options=[
|
|
IO.DynamicCombo.Option(
|
|
"grok-imagine-video-1.5",
|
|
[
|
|
IO.Autogrow.Input(
|
|
"reference_images",
|
|
template=IO.Autogrow.TemplateNames(
|
|
IO.Image.Input("image"),
|
|
names=[f"reference_{i}" for i in range(1, 8)],
|
|
min=1,
|
|
),
|
|
tooltip="Up to 7 reference images to guide the video generation. "
|
|
"Refer to them in the prompt as @Image1 ... @Image7, numbered "
|
|
"in input order; a batched input counts once per image.",
|
|
),
|
|
IO.Combo.Input(
|
|
"voice_1",
|
|
options=_GROK_VOICE_OPTIONS,
|
|
tooltip="Optional preset voice reference; refer to it in the prompt as @Audio1. "
|
|
"The API supports only these preset voices, not custom audio.",
|
|
),
|
|
IO.Combo.Input(
|
|
"voice_2",
|
|
options=_GROK_VOICE_OPTIONS,
|
|
tooltip="Optional second voice reference; @Audio2 in the prompt.",
|
|
),
|
|
IO.Combo.Input(
|
|
"voice_3",
|
|
options=_GROK_VOICE_OPTIONS,
|
|
tooltip="Optional third voice reference; @Audio3 in the prompt.",
|
|
),
|
|
IO.Combo.Input(
|
|
"resolution",
|
|
options=["480p", "720p"],
|
|
tooltip="The resolution of the output video.",
|
|
),
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=["16:9", "4:3", "3:2", "1:1", "2:3", "3:4", "9:16"],
|
|
tooltip="The aspect ratio of the output video.",
|
|
),
|
|
IO.Int.Input(
|
|
"duration",
|
|
default=6,
|
|
min=1,
|
|
max=15,
|
|
step=1,
|
|
tooltip="The duration of the output video in seconds.",
|
|
display_mode=IO.NumberDisplay.slider,
|
|
),
|
|
],
|
|
),
|
|
IO.DynamicCombo.Option(
|
|
"grok-imagine-video",
|
|
[
|
|
IO.Autogrow.Input(
|
|
"reference_images",
|
|
template=IO.Autogrow.TemplatePrefix(
|
|
IO.Image.Input("image"),
|
|
prefix="reference_",
|
|
min=1,
|
|
max=7,
|
|
),
|
|
tooltip="Up to 7 reference images to guide the video generation.",
|
|
),
|
|
IO.Combo.Input(
|
|
"resolution",
|
|
options=["480p", "720p"],
|
|
tooltip="The resolution of the output video.",
|
|
),
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=["16:9", "4:3", "3:2", "1:1", "2:3", "3:4", "9:16"],
|
|
tooltip="The aspect ratio of the output video.",
|
|
),
|
|
IO.Int.Input(
|
|
"duration",
|
|
default=6,
|
|
min=2,
|
|
max=10,
|
|
step=1,
|
|
tooltip="The duration of the output video in seconds.",
|
|
display_mode=IO.NumberDisplay.slider,
|
|
),
|
|
],
|
|
),
|
|
],
|
|
tooltip="The model to use for video generation.",
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
],
|
|
outputs=[
|
|
IO.Video.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
depends_on=IO.PriceBadgeDepends(
|
|
widgets=["model", "model.duration", "model.resolution"],
|
|
input_groups=["model.reference_images"],
|
|
),
|
|
expr="""
|
|
(
|
|
$is15 := $contains(widgets.model, "1.5");
|
|
$res := $lookup(widgets, "model.resolution");
|
|
$dur := $lookup(widgets, "model.duration");
|
|
$refs := $lookup(inputGroups, "model.reference_images");
|
|
$rate := $is15
|
|
? ($res = "720p" ? 0.14 : 0.08)
|
|
: ($res = "720p" ? 0.07 : 0.05);
|
|
$imgCost := $is15 ? 0.01 : 0.002;
|
|
$price := ($rate * $dur + $imgCost * $refs) * 1.43;
|
|
{"type":"usd","usd": $price}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
prompt: str,
|
|
model: dict,
|
|
seed: int,
|
|
) -> IO.NodeOutput:
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
total_images = sum(get_number_of_images(t) for t in model["reference_images"].values())
|
|
if total_images > 7:
|
|
raise ValueError(f"A maximum of 7 reference images is supported; {total_images} are connected.")
|
|
reference_audios = None
|
|
if model["model"] == "grok-imagine-video-1.5":
|
|
voices = [model.get(f"voice_{i}", "none") for i in range(1, 4)]
|
|
reference_audios = [VoiceReferenceObject(voice_id=v) for v in voices if v != "none"] or None
|
|
prompt = _normalize_grok_reference_prompt(prompt, total_images=total_images, voices=voices)
|
|
ref_image_urls = await upload_images_to_comfyapi(
|
|
cls,
|
|
list(model["reference_images"].values()),
|
|
mime_type="image/png",
|
|
wait_label="Uploading base images",
|
|
max_images=7,
|
|
)
|
|
initial_response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/videos/generations", method="POST"),
|
|
data=VideoGenerationRequest(
|
|
model=_GROK_VIDEO_MODEL_API_IDS.get(model["model"], model["model"]),
|
|
reference_images=[InputUrlObject(url=i) for i in ref_image_urls],
|
|
reference_audios=reference_audios,
|
|
prompt=prompt,
|
|
resolution=model["resolution"],
|
|
duration=model["duration"],
|
|
aspect_ratio=model["aspect_ratio"],
|
|
seed=seed,
|
|
),
|
|
response_model=VideoGenerationResponse,
|
|
)
|
|
response = await poll_op(
|
|
cls,
|
|
ApiEndpoint(path=f"/proxy/xai/v1/videos/{initial_response.request_id}"),
|
|
status_extractor=lambda r: r.status if r.status is not None else "complete",
|
|
response_model=VideoStatusResponse,
|
|
price_extractor=_extract_grok_video_price,
|
|
)
|
|
return IO.NodeOutput(await download_url_to_video_output(response.video.url))
|
|
|
|
|
|
class GrokVideoExtendNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return IO.Schema(
|
|
node_id="GrokVideoExtendNode",
|
|
display_name="Grok Video Extend",
|
|
category="partner/video/Grok",
|
|
description="Extend an existing video with a seamless continuation based on a text prompt.",
|
|
inputs=[
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
tooltip="Text description of what should happen next in the video.",
|
|
),
|
|
IO.Video.Input("video", tooltip="Source video to extend. MP4 format, 2-15 seconds."),
|
|
IO.DynamicCombo.Input(
|
|
"model",
|
|
options=[
|
|
IO.DynamicCombo.Option(
|
|
"grok-imagine-video",
|
|
[
|
|
IO.Int.Input(
|
|
"duration",
|
|
default=8,
|
|
min=2,
|
|
max=10,
|
|
step=1,
|
|
tooltip="Length of the extension in seconds.",
|
|
display_mode=IO.NumberDisplay.slider,
|
|
),
|
|
],
|
|
),
|
|
],
|
|
tooltip="The model to use for video extension.",
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=0,
|
|
min=0,
|
|
max=2147483647,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.number,
|
|
control_after_generate=True,
|
|
tooltip="Seed to determine if node should re-run; "
|
|
"actual results are nondeterministic regardless of seed.",
|
|
),
|
|
],
|
|
outputs=[
|
|
IO.Video.Output(),
|
|
],
|
|
hidden=[
|
|
IO.Hidden.auth_token_comfy_org,
|
|
IO.Hidden.api_key_comfy_org,
|
|
IO.Hidden.unique_id,
|
|
],
|
|
is_api_node=True,
|
|
price_badge=IO.PriceBadge(
|
|
depends_on=IO.PriceBadgeDepends(widgets=["model.duration"]),
|
|
expr="""
|
|
(
|
|
$dur := $lookup(widgets, "model.duration");
|
|
{
|
|
"type": "range_usd",
|
|
"min_usd": (0.02 + 0.05 * $dur) * 1.43,
|
|
"max_usd": (0.15 + 0.05 * $dur) * 1.43
|
|
}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
prompt: str,
|
|
video: Input.Video,
|
|
model: dict,
|
|
seed: int,
|
|
) -> IO.NodeOutput:
|
|
validate_string(prompt, strip_whitespace=True, min_length=1)
|
|
validate_video_duration(video, min_duration=2, max_duration=15)
|
|
video_size = get_fs_object_size(video.get_stream_source())
|
|
if video_size > 50 * 1024 * 1024:
|
|
raise ValueError(f"Video size ({video_size / 1024 / 1024:.1f}MB) exceeds 50MB limit.")
|
|
initial_response = await sync_op(
|
|
cls,
|
|
ApiEndpoint(path="/proxy/xai/v1/videos/extensions", method="POST"),
|
|
data=VideoExtensionRequest(
|
|
prompt=prompt,
|
|
video=InputUrlObject(url=await upload_video_to_comfyapi(cls, video)),
|
|
duration=model["duration"],
|
|
),
|
|
response_model=VideoGenerationResponse,
|
|
)
|
|
response = await poll_op(
|
|
cls,
|
|
ApiEndpoint(path=f"/proxy/xai/v1/videos/{initial_response.request_id}"),
|
|
status_extractor=lambda r: r.status if r.status is not None else "complete",
|
|
response_model=VideoStatusResponse,
|
|
price_extractor=_extract_grok_video_price,
|
|
)
|
|
return IO.NodeOutput(await download_url_to_video_output(response.video.url))
|
|
|
|
|
|
class GrokExtension(ComfyExtension):
|
|
@override
|
|
async def get_node_list(self) -> list[type[IO.ComfyNode]]:
|
|
return [
|
|
GrokImageNode,
|
|
GrokImageEditNode,
|
|
GrokImageEditNodeV2,
|
|
GrokVideoNode,
|
|
GrokVideoReferenceNode,
|
|
GrokVideoEditNode,
|
|
GrokVideoExtendNode,
|
|
]
|
|
|
|
|
|
async def comfy_entrypoint() -> GrokExtension:
|
|
return GrokExtension()
|