Add remaining Pika nodes (#119)

comfy_api_nodes/apis/__init__.py

@@ -1,6 +1,6 @@
 # generated by datamodel-codegen:
 #   filename:  filtered-openapi.yaml
-#   timestamp: 2025-05-03T21:40:30+00:00
+#   timestamp: 2025-05-04T04:12:39+00:00
 
 from __future__ import annotations
 
@@ -2125,9 +2125,28 @@ class CustomerStorageResourceResponse(BaseModel):
     )
 
 
+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 PikaBodyGeneratePikaffectsGeneratePikaffectsPost(BaseModel):
-    image: StrictBytes = Field(..., title='Image')
+    image: Optional[StrictBytes] = Field(None, title='Image')
-    pikaffect: Optional[str] = Field(None, title='Pikaffect')
+    pikaffect: Optional[Pikaffect] = Field(None, title='Pikaffect')
     promptText: Optional[str] = Field(None, title='Prompttext')
     negativePrompt: Optional[str] = Field(None, title='Negativeprompt')
     seed: Optional[int] = Field(None, title='Seed')
@@ -2138,15 +2157,15 @@ class PikaGenerateResponse(BaseModel):
 
 
 class PikaBodyGeneratePikadditionsGeneratePikadditionsPost(BaseModel):
-    video: StrictBytes = Field(..., title='Video')
+    video: Optional[StrictBytes] = Field(None, title='Video')
-    image: StrictBytes = Field(..., title='Image')
+    image: Optional[StrictBytes] = Field(None, title='Image')
     promptText: Optional[str] = Field(None, title='Prompttext')
     negativePrompt: Optional[str] = Field(None, title='Negativeprompt')
     seed: Optional[int] = Field(None, title='Seed')
 
 
 class PikaBodyGeneratePikaswapsGeneratePikaswapsPost(BaseModel):
-    video: StrictBytes = Field(..., title='Video')
+    video: Optional[StrictBytes] = Field(None, title='Video')
     image: Optional[StrictBytes] = Field(None, title='Image')
     promptText: Optional[str] = Field(None, title='Prompttext')
     modifyRegionMask: Optional[StrictBytes] = Field(
@@ -2179,7 +2198,7 @@ class AspectRatio1(RootModel[float]):
 
 
 class PikaBodyGenerate22C2vGenerate22PikascenesPost(BaseModel):
-    images: List[StrictBytes] = Field(..., title='Images')
+    images: Optional[List[StrictBytes]] = Field(None, title='Images')
     ingredientsMode: IngredientsMode = Field(..., title='Ingredientsmode')
     promptText: Optional[str] = Field(None, title='Prompttext')
     negativePrompt: Optional[str] = Field(None, title='Negativeprompt')
@@ -3629,8 +3648,8 @@ class PikaBodyGenerate22I2vGenerate22I2vPost(BaseModel):
 
 
 class PikaBodyGenerate22KeyframeGenerate22PikaframesPost(BaseModel):
-    keyFrames: List[StrictBytes] = Field(
+    keyFrames: Optional[List[StrictBytes]] = Field(
-        ..., description='Array of keyframe images', title='Keyframes'
+        None, description='Array of keyframe images', title='Keyframes'
     )
     promptText: str = Field(..., title='Prompttext')
     negativePrompt: Optional[str] = Field(None, title='Negativeprompt')

comfy_api_nodes/nodes_pika.py

@@ -1,8 +1,14 @@
-"""Pika API docs: https://pika-827374fb.mintlify.app/api-reference"""
+"""
+Pika x ComfyUI API Nodes
 
+Pika API docs: https://pika-827374fb.mintlify.app/api-reference
+"""
+
+import io
 from typing import Optional, TypeVar
 import logging
 import torch
+import numpy as np
 from comfy_api_nodes.apis import (
     PikaBodyGenerate22T2vGenerate22T2vPost,
     PikaGenerateResponse,
@@ -12,6 +18,11 @@ from comfy_api_nodes.apis import (
     IngredientsMode,
     PikaDurationEnum,
     PikaResolutionEnum,
+    PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
+    PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
+    PikaBodyGeneratePikaswapsGeneratePikaswapsPost,
+    PikaBodyGenerate22KeyframeGenerate22PikaframesPost,
+    Pikaffect,
 )
 from comfy_api_nodes.apis.client import (
     ApiEndpoint,
@@ -25,16 +36,22 @@ from comfy_api_nodes.apinode_utils import (
     download_url_to_video_output,
 )
 from comfy_api_nodes.mapper_utils import model_field_to_node_input
-from comfy.comfy_types.node_typing import IO, ComfyNodeABC, InputTypeOptions
+from comfy_api.input_impl.video_types import VideoInput, VideoContainer, VideoCodec
 from comfy_api.input_impl import VideoFromFile
+from comfy.comfy_types.node_typing import IO, ComfyNodeABC, InputTypeOptions
 
 R = TypeVar("R")
 
+PATH_PIKADDITIONS = "/proxy/pika/generate/pikadditions"
+PATH_PIKASWAPS = "/proxy/pika/generate/pikaswaps"
+PATH_PIKAFFECTS = "/proxy/pika/generate/pikaffects"
+
 PIKA_API_VERSION = "2.2"
 PATH_TEXT_TO_VIDEO = f"/proxy/pika/generate/{PIKA_API_VERSION}/t2v"
 PATH_IMAGE_TO_VIDEO = f"/proxy/pika/generate/{PIKA_API_VERSION}/i2v"
 PATH_PIKAFRAMES = f"/proxy/pika/generate/{PIKA_API_VERSION}/pikaframes"
 PATH_PIKASCENES = f"/proxy/pika/generate/{PIKA_API_VERSION}/pikascenes"
+
 PATH_VIDEO_GET = "/proxy/pika/videos"
 
 
@@ -100,11 +117,11 @@ class PikaNodeBase(ComfyNodeABC):
     CATEGORY = "api node/video/Pika"
     API_NODE = True
     FUNCTION = "api_call"
+    RETURN_TYPES = ("VIDEO",)
 
     def poll_for_task_status(
         self, task_id: str, auth_token: str
     ) -> PikaGenerateResponse:
-        """Polls the Pika API endpoint until the task reaches a terminal state."""
         polling_operation = PollingOperation(
             poll_endpoint=ApiEndpoint(
                 path=f"{PATH_VIDEO_GET}/{task_id}",
@@ -180,7 +197,6 @@ class PikaImageToVideoV2_2(PikaNodeBase):
         }
 
     DESCRIPTION = "Sends an image and prompt to the Pika API v2.2 to generate a video."
-    RETURN_TYPES = ("VIDEO",)
 
     def api_call(
         self,
@@ -192,15 +208,13 @@ class PikaImageToVideoV2_2(PikaNodeBase):
         duration: int,
         auth_token: Optional[str] = None,
     ) -> tuple[VideoFromFile]:
-        """API call for Pika 2.2 Image to Video."""
         # Convert image to BytesIO
         image_bytes_io = tensor_to_bytesio(image)
-        image_bytes_io.seek(0)  # Reset stream position
+        image_bytes_io.seek(0)
 
-        # Prepare file data for multipart upload
         pika_files = {"image": ("image.png", image_bytes_io, "image/png")}
 
-        # Prepare non-file data using the Pydantic model
+        # Prepare non-file data
         pika_request_data = PikaBodyGenerate22I2vGenerate22I2vPost(
             promptText=prompt_text,
             negativePrompt=negative_prompt,
@@ -226,7 +240,7 @@ class PikaImageToVideoV2_2(PikaNodeBase):
 
 
 class PikaTextToVideoNodeV2_2(PikaNodeBase):
-    """Pika 2.2 Text to Video Node."""
+    """Pika Text2Video v2.2 Node."""
 
     @classmethod
     def INPUT_TYPES(cls):
@@ -248,7 +262,6 @@ class PikaTextToVideoNodeV2_2(PikaNodeBase):
             },
         }
 
-    RETURN_TYPES = ("VIDEO",)
     DESCRIPTION = "Sends a text prompt to the Pika API v2.2 to generate a video."
 
     def api_call(
@@ -261,7 +274,6 @@ class PikaTextToVideoNodeV2_2(PikaNodeBase):
         aspect_ratio: float,
         auth_token: Optional[str] = None,
     ) -> tuple[VideoFromFile]:
-        """API call for Pika 2.2 Text to Video."""
         initial_operation = SynchronousOperation(
             endpoint=ApiEndpoint(
                 path=PATH_TEXT_TO_VIDEO,
@@ -285,7 +297,7 @@ class PikaTextToVideoNodeV2_2(PikaNodeBase):
 
 
 class PikaScenesV2_2(PikaNodeBase):
-    """Pika 2.2 Scenes Node."""
+    """PikaScenes v2.2 Node."""
 
     @classmethod
     def INPUT_TYPES(cls):
@@ -328,7 +340,6 @@ class PikaScenesV2_2(PikaNodeBase):
         }
 
     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_TYPES = ("VIDEO",)
 
     def api_call(
         self,
@@ -346,7 +357,7 @@ class PikaScenesV2_2(PikaNodeBase):
         image_ingredient_5: Optional[torch.Tensor] = None,
         auth_token: Optional[str] = None,
     ) -> tuple[VideoFromFile]:
-        """API call for Pika Scenes 2.2."""
+        # Convert all passed images to BytesIO
         all_image_bytes_io = []
         for image in [
             image_ingredient_1,
@@ -360,13 +371,11 @@ class PikaScenesV2_2(PikaNodeBase):
                 image_bytes_io.seek(0)
                 all_image_bytes_io.append(image_bytes_io)
 
-        # Prepare files data for multipart upload
         pika_files = [
             ("images", (f"image_{i}.png", image_bytes_io, "image/png"))
             for i, image_bytes_io in enumerate(all_image_bytes_io)
         ]
 
-        # Prepare non-file data using the Pydantic model
         pika_request_data = PikaBodyGenerate22C2vGenerate22PikascenesPost(
             ingredientsMode=ingredients_mode,
             promptText=prompt_text,
@@ -393,14 +402,348 @@ class PikaScenesV2_2(PikaNodeBase):
         return self.execute_task(initial_operation, auth_token)
 
 
+class PikAdditionsNode(PikaNodeBase):
+    """Pika Pikadditions Node. Add an image into a video."""
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "video": (IO.VIDEO, {"tooltip": "The video to add an image to."}),
+                "image": (IO.IMAGE, {"tooltip": "The image to add to the video."}),
+                "prompt_text": model_field_to_node_input(
+                    IO.STRING,
+                    PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
+                    "promptText",
+                    multiline=True,
+                ),
+                "negative_prompt": model_field_to_node_input(
+                    IO.STRING,
+                    PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
+                    "negativePrompt",
+                    multiline=True,
+                ),
+                "seed": model_field_to_node_input(
+                    IO.INT,
+                    PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
+                    "seed",
+                    min=0,
+                    max=0xFFFFFFFF,
+                    control_after_generate=True,
+                ),
+            },
+            "hidden": {
+                "auth_token": "AUTH_TOKEN_COMFY_ORG",
+            },
+        }
+
+    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."
+
+    def api_call(
+        self,
+        video: VideoInput,
+        image: torch.Tensor,
+        prompt_text: str,
+        negative_prompt: str,
+        seed: int,
+        auth_token: Optional[str] = None,
+    ) -> tuple[VideoFromFile]:
+        # Convert video to BytesIO
+        video_bytes_io = io.BytesIO()
+        video.save_to(video_bytes_io, format=VideoContainer.MP4, codec=VideoCodec.H264)
+        video_bytes_io.seek(0)
+
+        # Convert image to BytesIO
+        image_bytes_io = tensor_to_bytesio(image)
+        image_bytes_io.seek(0)
+
+        pika_files = [
+            ("video", ("video.mp4", video_bytes_io, "video/mp4")),
+            ("image", ("image.png", image_bytes_io, "image/png")),
+        ]
+
+        # Prepare non-file data
+        pika_request_data = PikaBodyGeneratePikadditionsGeneratePikadditionsPost(
+            promptText=prompt_text,
+            negativePrompt=negative_prompt,
+            seed=seed,
+        )
+
+        initial_operation = SynchronousOperation(
+            endpoint=ApiEndpoint(
+                path=PATH_PIKADDITIONS,
+                method=HttpMethod.POST,
+                request_model=PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
+                response_model=PikaGenerateResponse,
+            ),
+            request=pika_request_data,
+            files=pika_files,
+            content_type="multipart/form-data",
+            auth_token=auth_token,
+        )
+
+        return self.execute_task(initial_operation, auth_token)
+
+
+class PikaSwapsNode(PikaNodeBase):
+    """Pika Pikaswaps Node."""
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "video": (IO.VIDEO, {"tooltip": "The video to swap an object in."}),
+                "image": (
+                    IO.IMAGE,
+                    {
+                        "tooltip": "The image used to replace the masked object in the video."
+                    },
+                ),
+                "mask": (
+                    IO.MASK,
+                    {"tooltip": "Use the mask to define areas in the video to replace"},
+                ),
+                "prompt_text": model_field_to_node_input(
+                    IO.STRING,
+                    PikaBodyGeneratePikaswapsGeneratePikaswapsPost,
+                    "promptText",
+                    multiline=True,
+                ),
+                "negative_prompt": model_field_to_node_input(
+                    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",
+            },
+        }
+
+    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."
+    RETURN_TYPES = ("VIDEO",)
+
+    def api_call(
+        self,
+        video: VideoInput,
+        image: torch.Tensor,
+        mask: torch.Tensor,
+        prompt_text: str,
+        negative_prompt: str,
+        seed: int,
+        auth_token: Optional[str] = None,
+    ) -> tuple[VideoFromFile]:
+        # Convert video to BytesIO
+        video_bytes_io = io.BytesIO()
+        video.save_to(video_bytes_io, format=VideoContainer.MP4, codec=VideoCodec.H264)
+        video_bytes_io.seek(0)
+
+        # Convert mask to binary mask with three channels
+        mask = torch.round(mask)
+        mask = mask.repeat(1, 3, 1, 1)
+
+        # Convert 3-channel binary mask to BytesIO
+        mask_bytes_io = io.BytesIO()
+        mask_bytes_io.write(mask.numpy().astype(np.uint8))
+        mask_bytes_io.seek(0)
+
+        # Convert image to BytesIO
+        image_bytes_io = tensor_to_bytesio(image)
+        image_bytes_io.seek(0)
+
+        pika_files = [
+            ("video", ("video.mp4", video_bytes_io, "video/mp4")),
+            ("image", ("image.png", image_bytes_io, "image/png")),
+            ("modifyRegionMask", ("mask.png", mask_bytes_io, "image/png")),
+        ]
+
+        # Prepare non-file data
+        pika_request_data = PikaBodyGeneratePikaswapsGeneratePikaswapsPost(
+            promptText=prompt_text,
+            negativePrompt=negative_prompt,
+            seed=seed,
+        )
+
+        initial_operation = SynchronousOperation(
+            endpoint=ApiEndpoint(
+                path=PATH_PIKADDITIONS,
+                method=HttpMethod.POST,
+                request_model=PikaBodyGeneratePikadditionsGeneratePikadditionsPost,
+                response_model=PikaGenerateResponse,
+            ),
+            request=pika_request_data,
+            files=pika_files,
+            content_type="multipart/form-data",
+            auth_token=auth_token,
+        )
+
+        return self.execute_task(initial_operation, auth_token)
+
+
+class PikaffectsNode(PikaNodeBase):
+    """Pika Pikaffects Node."""
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "image": (
+                    IO.IMAGE,
+                    {"tooltip": "The reference image to apply the Pikaffect to."},
+                ),
+                "pikaffect": model_field_to_node_input(
+                    IO.COMBO,
+                    PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
+                    "pikaffect",
+                    enum_type=Pikaffect,
+                    default="Cake-ify",
+                ),
+                "prompt_text": model_field_to_node_input(
+                    IO.STRING,
+                    PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
+                    "promptText",
+                    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",
+            },
+        }
+
+    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"
+
+    def api_call(
+        self,
+        image: torch.Tensor,
+        pikaffect: str,
+        prompt_text: str,
+        negative_prompt: str,
+        seed: int,
+        auth_token: Optional[str] = None,
+    ) -> tuple[VideoFromFile]:
+
+        initial_operation = SynchronousOperation(
+            endpoint=ApiEndpoint(
+                path=PATH_PIKAFFECTS,
+                method=HttpMethod.POST,
+                request_model=PikaBodyGeneratePikaffectsGeneratePikaffectsPost,
+                response_model=PikaGenerateResponse,
+            ),
+            request=PikaBodyGeneratePikaffectsGeneratePikaffectsPost(
+                pikaffect=pikaffect,
+                promptText=prompt_text,
+                negativePrompt=negative_prompt,
+                seed=seed,
+            ),
+            files={"image": ("image.png", tensor_to_bytesio(image), "image/png")},
+            content_type="multipart/form-data",
+            auth_token=auth_token,
+        )
+
+        return self.execute_task(initial_operation, auth_token)
+
+
+class PikaStartEndFrameNode2_2(PikaNodeBase):
+    """PikaFrames v2.2 Node."""
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "image_start": (IO.IMAGE, {"tooltip": "The first image to combine."}),
+                "image_end": (IO.IMAGE, {"tooltip": "The last image to combine."}),
+                **cls.get_base_inputs_types(
+                    PikaBodyGenerate22KeyframeGenerate22PikaframesPost
+                ),
+            },
+            "hidden": {
+                "auth_token": "AUTH_TOKEN_COMFY_ORG",
+            },
+        }
+
+    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."
+
+    def api_call(
+        self,
+        image_start: torch.Tensor,
+        image_end: torch.Tensor,
+        prompt_text: str,
+        negative_prompt: str,
+        seed: int,
+        resolution: str,
+        duration: int,
+        auth_token: Optional[str] = None,
+    ) -> tuple[VideoFromFile]:
+
+        pika_files = [
+            (
+                "keyFrames",
+                ("image_start.png", tensor_to_bytesio(image_start), "image/png"),
+            ),
+            ("keyFrames", ("image_end.png", tensor_to_bytesio(image_end), "image/png")),
+        ]
+
+        initial_operation = SynchronousOperation(
+            endpoint=ApiEndpoint(
+                path=PATH_PIKAFRAMES,
+                method=HttpMethod.POST,
+                request_model=PikaBodyGenerate22KeyframeGenerate22PikaframesPost,
+                response_model=PikaGenerateResponse,
+            ),
+            request=PikaBodyGenerate22KeyframeGenerate22PikaframesPost(
+                promptText=prompt_text,
+                negativePrompt=negative_prompt,
+                seed=seed,
+                resolution=resolution,
+                duration=duration,
+            ),
+            files=pika_files,
+            content_type="multipart/form-data",
+            auth_token=auth_token,
+        )
+
+        return self.execute_task(initial_operation, auth_token)
+
+
 NODE_CLASS_MAPPINGS = {
     "PikaImageToVideoNode2_2": PikaImageToVideoV2_2,
     "PikaTextToVideoNode2_2": PikaTextToVideoNodeV2_2,
     "PikaScenesV2_2": PikaScenesV2_2,
+    "Pikadditions": PikAdditionsNode,
+    "Pikaswaps": PikaSwapsNode,
+    "Pikaffects": PikaffectsNode,
+    "PikaStartEndFrameNode2_2": PikaStartEndFrameNode2_2,
 }
 
 NODE_DISPLAY_NAME_MAPPINGS = {
-    "PikaImageToVideoNode2_2": "Pika 2.2 Image to Video",
+    "PikaImageToVideoNode2_2": "Pika Image to Video",
-    "PikaTextToVideoNode2_2": "Pika 2.2 Text to Video",
+    "PikaTextToVideoNode2_2": "Pika Text to Video",
-    "PikaScenesV2_2": "Pika 2.2 Scenes",
+    "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",
 }