xinyun/ComfyUI
1
0
Fork 0
mirror of https://git.datalinker.icu/comfyanonymous/ComfyUI synced 2025-12-14 00:14:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

convert nodes_pika.py to V3 schema (#10216)

Browse Source
This commit is contained in:
Alexander Piskun 2025-10-07 02:20:26 +03:00 committed by GitHub
parent a49007a7b0
commit e77e0a8f8f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 373 additions and 406 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

673
comfy_api_nodes/nodes_pika.py
View File

@ -5,14 +5,16 @@ Pika API docs: https://pika-827374fb.mintlify.app/api-reference
""" """
from __future__ import annotations from __future__ import annotations
import io from io import BytesIO
import logging import logging
from typing import Optional, TypeVar from typing import Optional, TypeVar
from enum import Enum
import numpy as np import numpy as np
import torch import torch
from comfy.comfy_types.node_typing import IO, ComfyNodeABC, InputTypeOptions from typing_extensions import override
from comfy_api.latest import ComfyExtension, io as comfy_io
from comfy_api.input_impl import VideoFromFile from comfy_api.input_impl import VideoFromFile
from comfy_api.input_impl.video_types import VideoCodec, VideoContainer, VideoInput from comfy_api.input_impl.video_types import VideoCodec, VideoContainer, VideoInput
from comfy_api_nodes.apinode_utils import ( from comfy_api_nodes.apinode_utils import (
@ -20,7 +22,6 @@ from comfy_api_nodes.apinode_utils import (
tensor_to_bytesio, tensor_to_bytesio,
) )
from comfy_api_nodes.apis import ( from comfy_api_nodes.apis import (
IngredientsMode,
PikaBodyGenerate22C2vGenerate22PikascenesPost, PikaBodyGenerate22C2vGenerate22PikascenesPost,
PikaBodyGenerate22I2vGenerate22I2vPost, PikaBodyGenerate22I2vGenerate22I2vPost,
PikaBodyGenerate22KeyframeGenerate22PikaframesPost, PikaBodyGenerate22KeyframeGenerate22PikaframesPost,
@ -28,10 +29,7 @@ from comfy_api_nodes.apis import (
PikaBodyGeneratePikadditionsGeneratePikadditionsPost, PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
PikaBodyGeneratePikaffectsGeneratePikaffectsPost, PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
PikaBodyGeneratePikaswapsGeneratePikaswapsPost, PikaBodyGeneratePikaswapsGeneratePikaswapsPost,
PikaDurationEnum,
Pikaffect,
PikaGenerateResponse, PikaGenerateResponse,
PikaResolutionEnum,
PikaVideoResponse, PikaVideoResponse,
) )
from comfy_api_nodes.apis.client import ( from comfy_api_nodes.apis.client import (
@ -41,7 +39,6 @@ from comfy_api_nodes.apis.client import (
PollingOperation, PollingOperation,
SynchronousOperation, SynchronousOperation,
) )
from comfy_api_nodes.mapper_utils import model_field_to_node_input
R = TypeVar("R") R = TypeVar("R")
@ -58,6 +55,35 @@ PATH_PIKASCENES = f"/proxy/pika/generate/{PIKA_API_VERSION}/pikascenes"
PATH_VIDEO_GET = "/proxy/pika/videos" PATH_VIDEO_GET = "/proxy/pika/videos"
class PikaDurationEnum(int, Enum):
integer_5 = 5
integer_10 = 10
class PikaResolutionEnum(str, Enum):
field_1080p = "1080p"
field_720p = "720p"
class Pikaffect(str, Enum):
Cake_ify = "Cake-ify"
Crumble = "Crumble"
Crush = "Crush"
Decapitate = "Decapitate"
Deflate = "Deflate"
Dissolve = "Dissolve"
Explode = "Explode"
Eye_pop = "Eye-pop"
Inflate = "Inflate"
Levitate = "Levitate"
Melt = "Melt"
Peel = "Peel"
Poke = "Poke"
Squish = "Squish"
Ta_da = "Ta-da"
Tear = "Tear"
class PikaApiError(Exception): class PikaApiError(Exception):
"""Exception for Pika API errors.""" """Exception for Pika API errors."""
@ -74,60 +100,11 @@ def is_valid_initial_response(response: PikaGenerateResponse) -> bool:
return hasattr(response, "video_id") and response.video_id is not None return hasattr(response, "video_id") and response.video_id is not None
class PikaNodeBase(ComfyNodeABC): async def poll_for_task_status(
"""Base class for Pika nodes."""
@classmethod
def get_base_inputs_types(
cls, request_model
) -> dict[str, tuple[IO, InputTypeOptions]]:
"""Get the base required inputs types common to all Pika nodes."""
return {
"prompt_text": model_field_to_node_input(
IO.STRING,
request_model,
"promptText",
multiline=True,
),
"negative_prompt": model_field_to_node_input(
IO.STRING,
request_model,
"negativePrompt",
multiline=True,
),
"seed": model_field_to_node_input(
IO.INT,
request_model,
"seed",
min=0,
max=0xFFFFFFFF,
control_after_generate=True,
),
"resolution": model_field_to_node_input(
IO.COMBO,
request_model,
"resolution",
enum_type=PikaResolutionEnum,
),
"duration": model_field_to_node_input(
IO.COMBO,
request_model,
"duration",
enum_type=PikaDurationEnum,
),
}
CATEGORY = "api node/video/Pika"
API_NODE = True
FUNCTION = "api_call"
RETURN_TYPES = ("VIDEO",)
async def poll_for_task_status(
self,
task_id: str, task_id: str,
auth_kwargs: Optional[dict[str, str]] = None, auth_kwargs: Optional[dict[str, str]] = None,
node_id: Optional[str] = None, node_id: Optional[str] = None,
) -> PikaGenerateResponse: ) -> PikaGenerateResponse:
polling_operation = PollingOperation( polling_operation = PollingOperation(
poll_endpoint=ApiEndpoint( poll_endpoint=ApiEndpoint(
path=f"{PATH_VIDEO_GET}/{task_id}", path=f"{PATH_VIDEO_GET}/{task_id}",
@ -154,12 +131,12 @@ class PikaNodeBase(ComfyNodeABC):
) )
return await polling_operation.execute() return await polling_operation.execute()
async def execute_task(
self, async def execute_task(
initial_operation: SynchronousOperation[R, PikaGenerateResponse], initial_operation: SynchronousOperation[R, PikaGenerateResponse],
auth_kwargs: Optional[dict[str, str]] = None, auth_kwargs: Optional[dict[str, str]] = None,
node_id: Optional[str] = None, node_id: Optional[str] = None,
) -> tuple[VideoFromFile]: ) -> tuple[VideoFromFile]:
"""Executes the initial operation then polls for the task status until it is completed. """Executes the initial operation then polls for the task status until it is completed.
Args: Args:
@ -176,7 +153,7 @@ class PikaNodeBase(ComfyNodeABC):
raise PikaApiError(error_msg) raise PikaApiError(error_msg)
task_id = initial_response.video_id task_id = initial_response.video_id
final_response = await self.poll_for_task_status(task_id, auth_kwargs) final_response = await poll_for_task_status(task_id, auth_kwargs, node_id=node_id)
if not is_valid_video_response(final_response): if not is_valid_video_response(final_response):
error_msg = ( error_msg = (
f"Pika task {task_id} succeeded but no video data found in response." f"Pika task {task_id} succeeded but no video data found in response."
@ -190,39 +167,54 @@ class PikaNodeBase(ComfyNodeABC):
return (await download_url_to_video_output(video_url),) return (await download_url_to_video_output(video_url),)
class PikaImageToVideoV2_2(PikaNodeBase): def get_base_inputs_types() -> list[comfy_io.Input]:
"""Get the base required inputs types common to all Pika nodes."""
return [
comfy_io.String.Input("prompt_text", multiline=True),
comfy_io.String.Input("negative_prompt", multiline=True),
comfy_io.Int.Input("seed", min=0, max=0xFFFFFFFF, control_after_generate=True),
comfy_io.Combo.Input(
"resolution", options=[resolution.value for resolution in PikaResolutionEnum], default="1080p"
),
comfy_io.Combo.Input(
"duration", options=[duration.value for duration in PikaDurationEnum], default=5
),
]
class PikaImageToVideoV2_2(comfy_io.ComfyNode):
"""Pika 2.2 Image to Video Node.""" """Pika 2.2 Image to Video Node."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
return { return comfy_io.Schema(
"required": { node_id="PikaImageToVideoNode2_2",
"image": ( display_name="Pika Image to Video",
IO.IMAGE, description="Sends an image and prompt to the Pika API v2.2 to generate a video.",
{"tooltip": "The image to convert to video"}, category="api node/video/Pika",
), inputs=[
**cls.get_base_inputs_types(PikaBodyGenerate22I2vGenerate22I2vPost), comfy_io.Image.Input("image", tooltip="The image to convert to video"),
}, *get_base_inputs_types(),
"hidden": { ],
"auth_token": "AUTH_TOKEN_COMFY_ORG", outputs=[comfy_io.Video.Output()],
"comfy_api_key": "API_KEY_COMFY_ORG", hidden=[
"unique_id": "UNIQUE_ID", comfy_io.Hidden.auth_token_comfy_org,
}, comfy_io.Hidden.api_key_comfy_org,
} comfy_io.Hidden.unique_id,
],
is_api_node=True,
)
DESCRIPTION = "Sends an image and prompt to the Pika API v2.2 to generate a video." @classmethod
async def execute(
async def api_call( cls,
self,
image: torch.Tensor, image: torch.Tensor,
prompt_text: str, prompt_text: str,
negative_prompt: str, negative_prompt: str,
seed: int, seed: int,
resolution: str, resolution: str,
duration: int, duration: int,
unique_id: str, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[VideoFromFile]:
# Convert image to BytesIO # Convert image to BytesIO
image_bytes_io = tensor_to_bytesio(image) image_bytes_io = tensor_to_bytesio(image)
image_bytes_io.seek(0) image_bytes_io.seek(0)
@ -237,7 +229,10 @@ class PikaImageToVideoV2_2(PikaNodeBase):
resolution=resolution, resolution=resolution,
duration=duration, duration=duration,
) )
auth = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_IMAGE_TO_VIDEO, path=PATH_IMAGE_TO_VIDEO,
@ -248,50 +243,55 @@ class PikaImageToVideoV2_2(PikaNodeBase):
request=pika_request_data, request=pika_request_data,
files=pika_files, files=pika_files,
content_type="multipart/form-data", content_type="multipart/form-data",
auth_kwargs=kwargs, auth_kwargs=auth,
) )
return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id)
class PikaTextToVideoNodeV2_2(PikaNodeBase): class PikaTextToVideoNodeV2_2(comfy_io.ComfyNode):
"""Pika Text2Video v2.2 Node.""" """Pika Text2Video v2.2 Node."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
return { return comfy_io.Schema(
"required": { node_id="PikaTextToVideoNode2_2",
**cls.get_base_inputs_types(PikaBodyGenerate22T2vGenerate22T2vPost), display_name="Pika Text to Video",
"aspect_ratio": model_field_to_node_input( description="Sends a text prompt to the Pika API v2.2 to generate a video.",
IO.FLOAT, category="api node/video/Pika",
PikaBodyGenerate22T2vGenerate22T2vPost, inputs=[
"aspectRatio", *get_base_inputs_types(),
comfy_io.Float.Input(
"aspect_ratio",
step=0.001, step=0.001,
min=0.4, min=0.4,
max=2.5, max=2.5,
default=1.7777777777777777, default=1.7777777777777777,
), tooltip="Aspect ratio (width / height)",
}, )
"hidden": { ],
"auth_token": "AUTH_TOKEN_COMFY_ORG", outputs=[comfy_io.Video.Output()],
"comfy_api_key": "API_KEY_COMFY_ORG", hidden=[
"unique_id": "UNIQUE_ID", comfy_io.Hidden.auth_token_comfy_org,
}, comfy_io.Hidden.api_key_comfy_org,
} comfy_io.Hidden.unique_id,
],
is_api_node=True,
)
DESCRIPTION = "Sends a text prompt to the Pika API v2.2 to generate a video." @classmethod
async def execute(
async def api_call( cls,
self,
prompt_text: str, prompt_text: str,
negative_prompt: str, negative_prompt: str,
seed: int, seed: int,
resolution: str, resolution: str,
duration: int, duration: int,
aspect_ratio: float, aspect_ratio: float,
unique_id: str, ) -> comfy_io.NodeOutput:
**kwargs, auth = {
) -> tuple[VideoFromFile]: "auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_TEXT_TO_VIDEO, path=PATH_TEXT_TO_VIDEO,
@ -307,62 +307,75 @@ class PikaTextToVideoNodeV2_2(PikaNodeBase):
duration=duration, duration=duration,
aspectRatio=aspect_ratio, aspectRatio=aspect_ratio,
), ),
auth_kwargs=kwargs, auth_kwargs=auth,
content_type="application/x-www-form-urlencoded", content_type="application/x-www-form-urlencoded",
) )
return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id)
class PikaScenesV2_2(PikaNodeBase): class PikaScenesV2_2(comfy_io.ComfyNode):
"""PikaScenes v2.2 Node.""" """PikaScenes v2.2 Node."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
image_ingredient_input = ( return comfy_io.Schema(
IO.IMAGE, node_id="PikaScenesV2_2",
{"tooltip": "Image that will be used as ingredient to create a video."}, display_name="Pika Scenes (Video Image Composition)",
) description="Combine your images to create a video with the objects in them. Upload multiple images as ingredients and generate a high-quality video that incorporates all of them.",
return { category="api node/video/Pika",
"required": { inputs=[
**cls.get_base_inputs_types( *get_base_inputs_types(),
PikaBodyGenerate22C2vGenerate22PikascenesPost, comfy_io.Combo.Input(
), "ingredients_mode",
"ingredients_mode": model_field_to_node_input( options=["creative", "precise"],
IO.COMBO,
PikaBodyGenerate22C2vGenerate22PikascenesPost,
"ingredientsMode",
enum_type=IngredientsMode,
default="creative", default="creative",
), ),
"aspect_ratio": model_field_to_node_input( comfy_io.Float.Input(
IO.FLOAT, "aspect_ratio",
PikaBodyGenerate22C2vGenerate22PikascenesPost,
"aspectRatio",
step=0.001, step=0.001,
min=0.4, min=0.4,
max=2.5, max=2.5,
default=1.7777777777777777, default=1.7777777777777777,
tooltip="Aspect ratio (width / height)",
), ),
}, comfy_io.Image.Input(
"optional": { "image_ingredient_1",
"image_ingredient_1": image_ingredient_input, optional=True,
"image_ingredient_2": image_ingredient_input, tooltip="Image that will be used as ingredient to create a video.",
"image_ingredient_3": image_ingredient_input, ),
"image_ingredient_4": image_ingredient_input, comfy_io.Image.Input(
"image_ingredient_5": image_ingredient_input, "image_ingredient_2",
}, optional=True,
"hidden": { tooltip="Image that will be used as ingredient to create a video.",
"auth_token": "AUTH_TOKEN_COMFY_ORG", ),
"comfy_api_key": "API_KEY_COMFY_ORG", comfy_io.Image.Input(
"unique_id": "UNIQUE_ID", "image_ingredient_3",
}, optional=True,
} tooltip="Image that will be used as ingredient to create a video.",
),
comfy_io.Image.Input(
"image_ingredient_4",
optional=True,
tooltip="Image that will be used as ingredient to create a video.",
),
comfy_io.Image.Input(
"image_ingredient_5",
optional=True,
tooltip="Image that will be used as ingredient to create a video.",
),
],
outputs=[comfy_io.Video.Output()],
hidden=[
comfy_io.Hidden.auth_token_comfy_org,
comfy_io.Hidden.api_key_comfy_org,
comfy_io.Hidden.unique_id,
],
is_api_node=True,
)
DESCRIPTION = "Combine your images to create a video with the objects in them. Upload multiple images as ingredients and generate a high-quality video that incorporates all of them." @classmethod
async def execute(
async def api_call( cls,
self,
prompt_text: str, prompt_text: str,
negative_prompt: str, negative_prompt: str,
seed: int, seed: int,
@ -370,14 +383,12 @@ class PikaScenesV2_2(PikaNodeBase):
duration: int, duration: int,
ingredients_mode: str, ingredients_mode: str,
aspect_ratio: float, aspect_ratio: float,
unique_id: str,
image_ingredient_1: Optional[torch.Tensor] = None, image_ingredient_1: Optional[torch.Tensor] = None,
image_ingredient_2: Optional[torch.Tensor] = None, image_ingredient_2: Optional[torch.Tensor] = None,
image_ingredient_3: Optional[torch.Tensor] = None, image_ingredient_3: Optional[torch.Tensor] = None,
image_ingredient_4: Optional[torch.Tensor] = None, image_ingredient_4: Optional[torch.Tensor] = None,
image_ingredient_5: Optional[torch.Tensor] = None, image_ingredient_5: Optional[torch.Tensor] = None,
**kwargs, ) -> comfy_io.NodeOutput:
) -> tuple[VideoFromFile]:
# Convert all passed images to BytesIO # Convert all passed images to BytesIO
all_image_bytes_io = [] all_image_bytes_io = []
for image in [ for image in [
@ -406,7 +417,10 @@ class PikaScenesV2_2(PikaNodeBase):
duration=duration, duration=duration,
aspectRatio=aspect_ratio, aspectRatio=aspect_ratio,
) )
auth = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_PIKASCENES, path=PATH_PIKASCENES,
@ -417,63 +431,54 @@ class PikaScenesV2_2(PikaNodeBase):
request=pika_request_data, request=pika_request_data,
files=pika_files, files=pika_files,
content_type="multipart/form-data", content_type="multipart/form-data",
auth_kwargs=kwargs, auth_kwargs=auth,
) )
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id) return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
class PikAdditionsNode(PikaNodeBase): class PikAdditionsNode(comfy_io.ComfyNode):
"""Pika Pikadditions Node. Add an image into a video.""" """Pika Pikadditions Node. Add an image into a video."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
return { return comfy_io.Schema(
"required": { node_id="Pikadditions",
"video": (IO.VIDEO, {"tooltip": "The video to add an image to."}), display_name="Pikadditions (Video Object Insertion)",
"image": (IO.IMAGE, {"tooltip": "The image to add to the video."}), description="Add any object or image into your video. Upload a video and specify what you'd like to add to create a seamlessly integrated result.",
"prompt_text": model_field_to_node_input( category="api node/video/Pika",
IO.STRING, inputs=[
PikaBodyGeneratePikadditionsGeneratePikadditionsPost, comfy_io.Video.Input("video", tooltip="The video to add an image to."),
"promptText", comfy_io.Image.Input("image", tooltip="The image to add to the video."),
multiline=True, comfy_io.String.Input("prompt_text", multiline=True),
), comfy_io.String.Input("negative_prompt", multiline=True),
"negative_prompt": model_field_to_node_input( comfy_io.Int.Input(
IO.STRING,
PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
"negativePrompt",
multiline=True,
),
"seed": model_field_to_node_input(
IO.INT,
PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
"seed", "seed",
min=0, min=0,
max=0xFFFFFFFF, max=0xFFFFFFFF,
control_after_generate=True, control_after_generate=True,
), ),
}, ],
"hidden": { outputs=[comfy_io.Video.Output()],
"auth_token": "AUTH_TOKEN_COMFY_ORG", hidden=[
"comfy_api_key": "API_KEY_COMFY_ORG", comfy_io.Hidden.auth_token_comfy_org,
"unique_id": "UNIQUE_ID", comfy_io.Hidden.api_key_comfy_org,
}, comfy_io.Hidden.unique_id,
} ],
is_api_node=True,
)
DESCRIPTION = "Add any object or image into your video. Upload a video and specify what you'd like to add to create a seamlessly integrated result." @classmethod
async def execute(
async def api_call( cls,
self,
video: VideoInput, video: VideoInput,
image: torch.Tensor, image: torch.Tensor,
prompt_text: str, prompt_text: str,
negative_prompt: str, negative_prompt: str,
seed: int, seed: int,
unique_id: str, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[VideoFromFile]:
# Convert video to BytesIO # Convert video to BytesIO
video_bytes_io = io.BytesIO() video_bytes_io = BytesIO()
video.save_to(video_bytes_io, format=VideoContainer.MP4, codec=VideoCodec.H264) video.save_to(video_bytes_io, format=VideoContainer.MP4, codec=VideoCodec.H264)
video_bytes_io.seek(0) video_bytes_io.seek(0)
@ -492,7 +497,10 @@ class PikAdditionsNode(PikaNodeBase):
negativePrompt=negative_prompt, negativePrompt=negative_prompt,
seed=seed, seed=seed,
) )
auth = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_PIKADDITIONS, path=PATH_PIKADDITIONS,
@ -503,74 +511,51 @@ class PikAdditionsNode(PikaNodeBase):
request=pika_request_data, request=pika_request_data,
files=pika_files, files=pika_files,
content_type="multipart/form-data", content_type="multipart/form-data",
auth_kwargs=kwargs, auth_kwargs=auth,
) )
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id) return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
class PikaSwapsNode(PikaNodeBase): class PikaSwapsNode(comfy_io.ComfyNode):
"""Pika Pikaswaps Node.""" """Pika Pikaswaps Node."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
return { return comfy_io.Schema(
"required": { node_id="Pikaswaps",
"video": (IO.VIDEO, {"tooltip": "The video to swap an object in."}), display_name="Pika Swaps (Video Object Replacement)",
"image": ( description="Swap out any object or region of your video with a new image or object. Define areas to replace either with a mask or coordinates.",
IO.IMAGE, category="api node/video/Pika",
{ inputs=[
"tooltip": "The image used to replace the masked object in the video." comfy_io.Video.Input("video", tooltip="The video to swap an object in."),
}, comfy_io.Image.Input("image", tooltip="The image used to replace the masked object in the video."),
), comfy_io.Mask.Input("mask", tooltip="Use the mask to define areas in the video to replace"),
"mask": ( comfy_io.String.Input("prompt_text", multiline=True),
IO.MASK, comfy_io.String.Input("negative_prompt", multiline=True),
{"tooltip": "Use the mask to define areas in the video to replace"}, comfy_io.Int.Input("seed", min=0, max=0xFFFFFFFF, control_after_generate=True),
), ],
"prompt_text": model_field_to_node_input( outputs=[comfy_io.Video.Output()],
IO.STRING, hidden=[
PikaBodyGeneratePikaswapsGeneratePikaswapsPost, comfy_io.Hidden.auth_token_comfy_org,
"promptText", comfy_io.Hidden.api_key_comfy_org,
multiline=True, comfy_io.Hidden.unique_id,
), ],
"negative_prompt": model_field_to_node_input( is_api_node=True,
IO.STRING, )
PikaBodyGeneratePikaswapsGeneratePikaswapsPost,
"negativePrompt",
multiline=True,
),
"seed": model_field_to_node_input(
IO.INT,
PikaBodyGeneratePikaswapsGeneratePikaswapsPost,
"seed",
min=0,
max=0xFFFFFFFF,
control_after_generate=True,
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
"comfy_api_key": "API_KEY_COMFY_ORG",
"unique_id": "UNIQUE_ID",
},
}
DESCRIPTION = "Swap out any object or region of your video with a new image or object. Define areas to replace either with a mask or coordinates." @classmethod
RETURN_TYPES = ("VIDEO",) async def execute(
cls,
async def api_call(
self,
video: VideoInput, video: VideoInput,
image: torch.Tensor, image: torch.Tensor,
mask: torch.Tensor, mask: torch.Tensor,
prompt_text: str, prompt_text: str,
negative_prompt: str, negative_prompt: str,
seed: int, seed: int,
unique_id: str, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[VideoFromFile]:
# Convert video to BytesIO # Convert video to BytesIO
video_bytes_io = io.BytesIO() video_bytes_io = BytesIO()
video.save_to(video_bytes_io, format=VideoContainer.MP4, codec=VideoCodec.H264) video.save_to(video_bytes_io, format=VideoContainer.MP4, codec=VideoCodec.H264)
video_bytes_io.seek(0) video_bytes_io.seek(0)
@ -579,7 +564,7 @@ class PikaSwapsNode(PikaNodeBase):
mask = mask.repeat(1, 3, 1, 1) mask = mask.repeat(1, 3, 1, 1)
# Convert 3-channel binary mask to BytesIO # Convert 3-channel binary mask to BytesIO
mask_bytes_io = io.BytesIO() mask_bytes_io = BytesIO()
mask_bytes_io.write(mask.numpy().astype(np.uint8)) mask_bytes_io.write(mask.numpy().astype(np.uint8))
mask_bytes_io.seek(0) mask_bytes_io.seek(0)
@ -599,7 +584,10 @@ class PikaSwapsNode(PikaNodeBase):
negativePrompt=negative_prompt, negativePrompt=negative_prompt,
seed=seed, seed=seed,
) )
auth = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_PIKADDITIONS, path=PATH_PIKADDITIONS,
@ -610,71 +598,52 @@ class PikaSwapsNode(PikaNodeBase):
request=pika_request_data, request=pika_request_data,
files=pika_files, files=pika_files,
content_type="multipart/form-data", content_type="multipart/form-data",
auth_kwargs=kwargs, auth_kwargs=auth,
) )
return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id)
class PikaffectsNode(PikaNodeBase): class PikaffectsNode(comfy_io.ComfyNode):
"""Pika Pikaffects Node.""" """Pika Pikaffects Node."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
return { return comfy_io.Schema(
"required": { node_id="Pikaffects",
"image": ( display_name="Pikaffects (Video Effects)",
IO.IMAGE, description="Generate a video with a specific Pikaffect. Supported Pikaffects: Cake-ify, Crumble, Crush, Decapitate, Deflate, Dissolve, Explode, Eye-pop, Inflate, Levitate, Melt, Peel, Poke, Squish, Ta-da, Tear",
{"tooltip": "The reference image to apply the Pikaffect to."}, category="api node/video/Pika",
inputs=[
comfy_io.Image.Input("image", tooltip="The reference image to apply the Pikaffect to."),
comfy_io.Combo.Input(
"pikaffect", options=[pikaffect.value for pikaffect in Pikaffect], default="Cake-ify"
), ),
"pikaffect": model_field_to_node_input( comfy_io.String.Input("prompt_text", multiline=True),
IO.COMBO, comfy_io.String.Input("negative_prompt", multiline=True),
PikaBodyGeneratePikaffectsGeneratePikaffectsPost, comfy_io.Int.Input("seed", min=0, max=0xFFFFFFFF, control_after_generate=True),
"pikaffect", ],
enum_type=Pikaffect, outputs=[comfy_io.Video.Output()],
default="Cake-ify", hidden=[
), comfy_io.Hidden.auth_token_comfy_org,
"prompt_text": model_field_to_node_input( comfy_io.Hidden.api_key_comfy_org,
IO.STRING, comfy_io.Hidden.unique_id,
PikaBodyGeneratePikaffectsGeneratePikaffectsPost, ],
"promptText", is_api_node=True,
multiline=True, )
),
"negative_prompt": model_field_to_node_input(
IO.STRING,
PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
"negativePrompt",
multiline=True,
),
"seed": model_field_to_node_input(
IO.INT,
PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
"seed",
min=0,
max=0xFFFFFFFF,
control_after_generate=True,
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
"comfy_api_key": "API_KEY_COMFY_ORG",
"unique_id": "UNIQUE_ID",
},
}
DESCRIPTION = "Generate a video with a specific Pikaffect. Supported Pikaffects: Cake-ify, Crumble, Crush, Decapitate, Deflate, Dissolve, Explode, Eye-pop, Inflate, Levitate, Melt, Peel, Poke, Squish, Ta-da, Tear" @classmethod
async def execute(
async def api_call( cls,
self,
image: torch.Tensor, image: torch.Tensor,
pikaffect: str, pikaffect: str,
prompt_text: str, prompt_text: str,
negative_prompt: str, negative_prompt: str,
seed: int, seed: int,
unique_id: str, ) -> comfy_io.NodeOutput:
**kwargs, auth = {
) -> tuple[VideoFromFile]: "auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_PIKAFFECTS, path=PATH_PIKAFFECTS,
@ -690,36 +659,38 @@ class PikaffectsNode(PikaNodeBase):
), ),
files={"image": ("image.png", tensor_to_bytesio(image), "image/png")}, files={"image": ("image.png", tensor_to_bytesio(image), "image/png")},
content_type="multipart/form-data", content_type="multipart/form-data",
auth_kwargs=kwargs, auth_kwargs=auth,
) )
return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id)
class PikaStartEndFrameNode2_2(PikaNodeBase): class PikaStartEndFrameNode2_2(comfy_io.ComfyNode):
"""PikaFrames v2.2 Node.""" """PikaFrames v2.2 Node."""
@classmethod @classmethod
def INPUT_TYPES(cls): def define_schema(cls) -> comfy_io.Schema:
return { return comfy_io.Schema(
"required": { node_id="PikaStartEndFrameNode2_2",
"image_start": (IO.IMAGE, {"tooltip": "The first image to combine."}), display_name="Pika Start and End Frame to Video",
"image_end": (IO.IMAGE, {"tooltip": "The last image to combine."}), description="Generate a video by combining your first and last frame. Upload two images to define the start and end points, and let the AI create a smooth transition between them.",
**cls.get_base_inputs_types( category="api node/video/Pika",
PikaBodyGenerate22KeyframeGenerate22PikaframesPost inputs=[
), comfy_io.Image.Input("image_start", tooltip="The first image to combine."),
}, comfy_io.Image.Input("image_end", tooltip="The last image to combine."),
"hidden": { *get_base_inputs_types(),
"auth_token": "AUTH_TOKEN_COMFY_ORG", ],
"comfy_api_key": "API_KEY_COMFY_ORG", outputs=[comfy_io.Video.Output()],
"unique_id": "UNIQUE_ID", hidden=[
}, comfy_io.Hidden.auth_token_comfy_org,
} comfy_io.Hidden.api_key_comfy_org,
comfy_io.Hidden.unique_id,
],
is_api_node=True,
)
DESCRIPTION = "Generate a video by combining your first and last frame. Upload two images to define the start and end points, and let the AI create a smooth transition between them." @classmethod
async def execute(
async def api_call( cls,
self,
image_start: torch.Tensor, image_start: torch.Tensor,
image_end: torch.Tensor, image_end: torch.Tensor,
prompt_text: str, prompt_text: str,
@ -727,15 +698,15 @@ class PikaStartEndFrameNode2_2(PikaNodeBase):
seed: int, seed: int,
resolution: str, resolution: str,
duration: int, duration: int,
unique_id: str, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[VideoFromFile]:
pika_files = [ pika_files = [
("keyFrames", ("image_start.png", tensor_to_bytesio(image_start), "image/png")), ("keyFrames", ("image_start.png", tensor_to_bytesio(image_start), "image/png")),
("keyFrames", ("image_end.png", tensor_to_bytesio(image_end), "image/png")), ("keyFrames", ("image_end.png", tensor_to_bytesio(image_end), "image/png")),
] ]
auth = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_PIKAFRAMES, path=PATH_PIKAFRAMES,
@ -752,28 +723,24 @@ class PikaStartEndFrameNode2_2(PikaNodeBase):
), ),
files=pika_files, files=pika_files,
content_type="multipart/form-data", content_type="multipart/form-data",
auth_kwargs=kwargs, auth_kwargs=auth,
) )
return await execute_task(initial_operation, auth_kwargs=auth, node_id=cls.hidden.unique_id)
return await self.execute_task(initial_operation, auth_kwargs=kwargs, node_id=unique_id)
NODE_CLASS_MAPPINGS = { class PikaApiNodesExtension(ComfyExtension):
"PikaImageToVideoNode2_2": PikaImageToVideoV2_2, @override
"PikaTextToVideoNode2_2": PikaTextToVideoNodeV2_2, async def get_node_list(self) -> list[type[comfy_io.ComfyNode]]:
"PikaScenesV2_2": PikaScenesV2_2, return [
"Pikadditions": PikAdditionsNode, PikaImageToVideoV2_2,
"Pikaswaps": PikaSwapsNode, PikaTextToVideoNodeV2_2,
"Pikaffects": PikaffectsNode, PikaScenesV2_2,
"PikaStartEndFrameNode2_2": PikaStartEndFrameNode2_2, PikAdditionsNode,
} PikaSwapsNode,
PikaffectsNode,
PikaStartEndFrameNode2_2,
]
NODE_DISPLAY_NAME_MAPPINGS = {
"PikaImageToVideoNode2_2": "Pika Image to Video", async def comfy_entrypoint() -> PikaApiNodesExtension:
"PikaTextToVideoNode2_2": "Pika Text to Video", return PikaApiNodesExtension()
"PikaScenesV2_2": "Pika Scenes (Video Image Composition)",
"Pikadditions": "Pikadditions (Video Object Insertion)",
"Pikaswaps": "Pika Swaps (Video Object Replacement)",
"Pikaffects": "Pikaffects (Video Effects)",
"PikaStartEndFrameNode2_2": "Pika Start and End Frame to Video",
}