Add Pixverse nodes (#46)

comfy_api_nodes/apis/pixverse_api.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class PixverseStatus(int, Enum):
+    successful = 1
+    generating = 5
+    contents_moderation = 7
+    failed = 8
+
+
+class PixverseAspectRatio(str, Enum):
+    ratio_16_9 = "16:9"
+    ratio_4_3 = "4:3"
+    ratio_1_1 = "1:1"
+    ratio_3_4 = "3:4"
+    ratio_9_16 = "9:16"
+
+
+class PixverseQuality(str, Enum):
+    res_360p = "360p"
+    res_540p = "540p"
+    res_720p = "720p"
+    res_1080p = "1080p"
+
+
+class PixverseDuration(int, Enum):
+    dur_5 = 5
+    dur_8 = 8
+
+
+class PixverseMotionMode(str, Enum):
+    normal = "normal"
+    fast = "fast"
+
+
+class PixverseStyle(str, Enum):
+    anime = "anime"
+    animation_3d = "3d_animation"
+    clay = "clay"
+    comic = "comic"
+    cyberpunk = "cyberpunk"
+
+
+# NOTE: forgoing descriptions for now in return for dev speed
+class PixverseDto_V2OpenAPIT2VReq(BaseModel):
+    aspect_ratio: PixverseAspectRatio = Field(...)
+    quality: PixverseQuality = Field(...)
+    duration: PixverseDuration = Field(...)
+    model: Optional[str] = Field("v3.5")
+    motion_mode: Optional[PixverseMotionMode] = Field(PixverseMotionMode.normal)
+    prompt: str = Field(...)
+    negative_prompt: Optional[str] = Field(None)
+    seed: Optional[int] = Field(None)
+    style: Optional[str] = Field(None)
+    template_id: Optional[str] = Field(None)
+    water_mark: Optional[bool] = Field(None)
+
+
+class PixverseController_ResponseData(BaseModel):
+    ErrCode: Optional[int] = Field(None)
+    ErrMsg: Optional[str] = Field(None)
+    Resp: Optional[PixverseDto_V2OpenAPII2VResp] = Field(None)
+
+
+class PixverseDto_V2OpenAPII2VResp(BaseModel):
+    video_id: int = Field(..., description='Video_id')
+
+
+class PixverseGenerationStatusResponse(BaseModel):
+    ErrCode: Optional[int] = Field(None)
+    ErrMsg: Optional[str] = Field(None)
+    Resp: Optional[PixverseDto_GetOpenapiMediaDetailResp] = Field(None)
+
+
+class PixverseDto_GetOpenapiMediaDetailResp(BaseModel):
+    create_time: Optional[str] = Field(None)
+    id: Optional[int] = Field(None)
+    modify_time: Optional[str] = Field(None)
+    negative_prompt: Optional[str] = Field(None)
+    outputHeight: Optional[int] = Field(None)
+    outputWidth: Optional[int] = Field(None)
+    prompt: Optional[str] = Field(None)
+    resolution_ratio: Optional[int] = Field(None)
+    seed: Optional[int] = Field(None)
+    size: Optional[int] = Field(None)
+    status: Optional[int] = Field(None)
+    style: Optional[str] = Field(None)
+    url: Optional[str] = Field(None)

comfy_api_nodes/nodes_pixverse.py

@@ -0,0 +1,154 @@
+from inspect import cleandoc
+
+from comfy_api_nodes.apis.pixverse_api import (
+    PixverseDto_V2OpenAPIT2VReq,
+    PixverseController_ResponseData,
+    PixverseGenerationStatusResponse,
+    PixverseAspectRatio,
+    PixverseQuality,
+    PixverseDuration,
+    PixverseMotionMode,
+    PixverseStatus,
+)
+from comfy_api_nodes.apis.client import (
+    ApiEndpoint,
+    HttpMethod,
+    SynchronousOperation,
+    PollingOperation,
+    EmptyRequest,
+)
+from comfy.comfy_types.node_typing import IO, ComfyNodeABC
+from comfy_api.input_impl import VideoFromFile
+
+import requests
+from io import BytesIO
+
+
+class PixverseTextToVideoNode(ComfyNodeABC):
+    """
+    Generates videos synchronously based on prompt and output_size.
+    """
+
+    RETURN_TYPES = (IO.VIDEO,)
+    DESCRIPTION = cleandoc(__doc__ or "")  # Handle potential None value
+    FUNCTION = "api_call"
+    API_NODE = True
+    CATEGORY = "api node/video/Pixverse"
+
+    @classmethod
+    def INPUT_TYPES(s):
+        return {
+            "required": {
+                "prompt": (
+                    IO.STRING,
+                    {
+                        "multiline": True,
+                        "default": "",
+                        "tooltip": "Prompt for the video generation",
+                    },
+                ),
+                "aspect_ratio": (
+                    [ratio.value for ratio in PixverseAspectRatio],
+                ),
+                "quality": (
+                    [resolution.value for resolution in PixverseQuality],
+                    {
+                        "default": PixverseQuality.res_540p,
+                    },
+                ),
+                "duration_seconds": ([dur.value for dur in PixverseDuration],),
+                "motion_mode": ([mode.value for mode in PixverseMotionMode],),
+                "seed": (
+                    IO.INT,
+                    {
+                        "default": 0,
+                        "min": 0,
+                        "max": 2147483647,
+                        "control_after_generate": True,
+                        "tooltip": "Seed for video generation.",
+                    },
+                ),
+            },
+            "optional": {
+                "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,
+        prompt: str,
+        aspect_ratio: str,
+        quality: str,
+        duration_seconds: int,
+        motion_mode: str,
+        seed,
+        negative_prompt: str=None,
+        auth_token=None,
+        **kwargs,
+    ):
+        # 1080p is limited to 5 seconds duration
+        # only normal motion_mode supported for 1080p or for non-5 second duration
+        if quality == PixverseQuality.res_1080p:
+            motion_mode = PixverseMotionMode.normal
+            duration_seconds = PixverseDuration.dur_5
+        elif duration_seconds != PixverseDuration.dur_5:
+            motion_mode = PixverseMotionMode.normal
+
+        operation = SynchronousOperation(
+            endpoint=ApiEndpoint(
+                path="/proxy/pixverse/video/text/generate",
+                method=HttpMethod.POST,
+                request_model=PixverseDto_V2OpenAPIT2VReq,
+                response_model=PixverseController_ResponseData,
+            ),
+            request=PixverseDto_V2OpenAPIT2VReq(
+                prompt=prompt,
+                aspect_ratio=aspect_ratio,
+                quality=quality,
+                duration=duration_seconds,
+                motion_mode=motion_mode,
+                negative_prompt=negative_prompt if negative_prompt else None,
+                seed=seed,
+            ),
+            auth_token=auth_token,
+        )
+        response_api = operation.execute()
+
+        if response_api.Resp is None:
+            raise Exception(f"Pixverse request failed: '{response_api.ErrMsg}'")
+
+        operation = PollingOperation(
+            poll_endpoint=ApiEndpoint(
+                path=f"/proxy/pixverse/video/result/{response_api.Resp.video_id}",
+                method=HttpMethod.GET,
+                request_model=EmptyRequest,
+                response_model=PixverseGenerationStatusResponse,
+            ),
+            completed_statuses=[PixverseStatus.successful],
+            failed_statuses=[PixverseStatus.contents_moderation, PixverseStatus.failed],
+            status_extractor=lambda x: x.Resp.status,
+            auth_token=auth_token,
+        )
+        response_poll = operation.execute()
+
+        vid_response = requests.get(response_poll.Resp.url)
+        return (VideoFromFile(BytesIO(vid_response.content)),)
+
+
+NODE_CLASS_MAPPINGS = {
+    "PixverseTextToVideoNode": PixverseTextToVideoNode
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "PixverseTextToVideoNode": "Pixverse Text to Video"
+}

nodes.py

@@ -2268,6 +2268,7 @@ def init_builtin_extra_nodes():
         "nodes_runway.py",
         "nodes_luma.py",
         "nodes_recraft.py",
+        "nodes_pixverse.py",
     ]
 
     import_failed = []