from inspect import cleandoc from comfy_api_nodes.apis.pixverse_api import ( PixverseDto_V2OpenAPIT2VReq, PixverseController_ResponseData, PixverseGenerationStatusResponse, PixverseAspectRatio, PixverseQuality, PixverseDuration, PixverseMotionMode, PixverseStatus, PixverseIO, pixverse_templates, ) 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 PixverseTemplateNode: """ Select template for Pixverse Video generation. """ RETURN_TYPES = (PixverseIO.TEMPLATE,) RETURN_NAMES = ("pixverse_template",) FUNCTION = "create_template" CATEGORY = "api node/video/Pixverse" @classmethod def INPUT_TYPES(s): return { "required": { "template": (list(pixverse_templates.keys()), ), } } def create_template(self, template: str): template_id = pixverse_templates.get(template, None) if template_id is None: raise Exception(f"Template '{template}' is not recognized.") # just return the integer return (template_id,) 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.", }, ), "pixverse_template": ( PixverseIO.TEMPLATE, { "tooltip": "An optional template to influence style of generation, created by the Pixverse Template node." } ) }, "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, pixverse_template: int=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, template_id=pixverse_template, 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, "PixverseTemplateNode": PixverseTemplateNode, } NODE_DISPLAY_NAME_MAPPINGS = { "PixverseTextToVideoNode": "Pixverse Text to Video", "PixverseTemplateNode": "Pixverse Template", }