Custom-Advanced-VACE-Node/nodes_utility.py

import torch
import numpy as np
from comfy.utils import common_upscale
from .utils import log
from einops import rearrange

try:
    from server import PromptServer
except:
    PromptServer = None

VAE_STRIDE = (4, 8, 8)
PATCH_SIZE = (1, 2, 2)

class WanVideoImageResizeToClosest:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
            "image": ("IMAGE", {"tooltip": "Image to resize"}),
            "generation_width": ("INT", {"default": 832, "min": 64, "max": 8096, "step": 8, "tooltip": "Width of the image to encode"}),
            "generation_height": ("INT", {"default": 480, "min": 64, "max": 8096, "step": 8, "tooltip": "Height of the image to encode"}),
            "aspect_ratio_preservation": (["keep_input", "stretch_to_new", "crop_to_new"],),
            },
        }

    RETURN_TYPES = ("IMAGE", "INT", "INT", )
    RETURN_NAMES = ("image","width","height",)
    FUNCTION = "process"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Resizes image to the closest supported resolution based on aspect ratio and max pixels, according to the original code"

    def process(self, image, generation_width, generation_height, aspect_ratio_preservation ):
    
        H, W = image.shape[1], image.shape[2]
        max_area = generation_width * generation_height

        crop = "disabled"

        if aspect_ratio_preservation == "keep_input":
            aspect_ratio = H / W
        elif aspect_ratio_preservation == "stretch_to_new" or aspect_ratio_preservation == "crop_to_new":
            aspect_ratio = generation_height / generation_width
            if aspect_ratio_preservation == "crop_to_new":
                crop = "center"
                
        lat_h = round(
        np.sqrt(max_area * aspect_ratio) // VAE_STRIDE[1] //
        PATCH_SIZE[1] * PATCH_SIZE[1])
        lat_w = round(
            np.sqrt(max_area / aspect_ratio) // VAE_STRIDE[2] //
            PATCH_SIZE[2] * PATCH_SIZE[2])
        h = lat_h * VAE_STRIDE[1]
        w = lat_w * VAE_STRIDE[2]

        resized_image = common_upscale(image.movedim(-1, 1), w, h, "lanczos", crop).movedim(1, -1)

        return (resized_image, w, h)

class ExtractStartFramesForContinuations:
    @classmethod
    def INPUT_TYPES(s):
        return {
            "required": {
                "input_video_frames": ("IMAGE", {"tooltip": "Input video frames to extract the start frames from."}),
                "num_frames": ("INT", {"default": 10, "min": 1, "max": 1024, "step": 1, "tooltip": "Number of frames to get from the start of the video."}),
            },
        }

    RETURN_TYPES = ("IMAGE",)
    RETURN_NAMES = ("start_frames",)
    FUNCTION = "get_start_frames"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Extracts the first N frames from a video sequence for continuations."

    def get_start_frames(self, input_video_frames, num_frames):
        if input_video_frames is None or input_video_frames.shape[0] == 0:
            log.warning("Input video frames are empty. Returning an empty tensor.")
            if input_video_frames is not None:
                return (torch.empty((0,) + input_video_frames.shape[1:], dtype=input_video_frames.dtype),)
            else:
                # Return a tensor with 4 dimensions, as expected for an IMAGE type.
                return (torch.empty((0, 64, 64, 3), dtype=torch.float32),)

        total_frames = input_video_frames.shape[0]
        num_to_get = min(num_frames, total_frames)

        if num_to_get < num_frames:
            log.warning(f"Requested {num_frames} frames, but input video only has {total_frames} frames. Returning first {num_to_get} frames.")

        start_frames = input_video_frames[:num_to_get]

        return (start_frames.cpu().float(),)

class WanVideoVACEStartToEndFrame:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
            "num_frames": ("INT", {"default": 81, "min": 1, "max": 10000, "step": 4, "tooltip": "Number of frames to encode"}),
            "empty_frame_level": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "White level of empty frame to use"}),
            },
            "optional": {
                "start_image": ("IMAGE",),
                "end_image": ("IMAGE",),
                "control_images": ("IMAGE",),
                "inpaint_mask": ("MASK", {"tooltip": "Inpaint mask to use for the empty frames"}),
                "start_index": ("INT", {"default": 0, "min": 0, "max": 10000, "step": 1, "tooltip": "Index to start from"}),
                "end_index": ("INT", {"default": -1, "min": -10000, "max": 10000, "step": 1, "tooltip": "Index to end at"}),
                "control_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01, "tooltip": "How much does the control images apply?"}),
                "control_ease": ("INT", {"default": 0.0, "min": 0.0, "max": 100.0, "step": 1, "tooltip": "How many frames to ease in the control video?"}),
            },
        }

    RETURN_TYPES = ("IMAGE", "MASK", )
    RETURN_NAMES = ("images", "masks",)
    FUNCTION = "process"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Helper node to create start/end frame batch and masks for VACE"

    def process(self, num_frames, empty_frame_level, start_image=None, end_image=None, control_images=None, inpaint_mask=None, start_index=0, end_index=-1, control_strength=1.0, control_ease=0):

        device = start_image.device if start_image is not None else end_image.device
        B, H, W, C = start_image.shape if start_image is not None else end_image.shape
        
        if control_images is not None:
            # weaken the control images?
            if control_strength < 1.0:
                # strength happens at much smaller number
                control_strength *= 2.0
                control_strength = control_strength * control_strength / 8.0
                control_images = torch.lerp(torch.ones((control_images.shape[0], control_images.shape[1], control_images.shape[2], control_images.shape[3])) * empty_frame_level, control_images, control_strength)
                
            # ease in control stuff?
            if num_frames > control_ease and control_ease > 0:
                empty_frame = torch.ones((1, control_images.shape[1], control_images.shape[2], control_images.shape[3])) * empty_frame_level
                if start_image is not None:
                    for i in range(1, control_ease + 1):
                        control_images[i] = torch.lerp(control_images[i], empty_frame, (control_ease - i) / (1 + control_ease))
                else:
                    for i in range(num_frames - control_ease - 1, num_frames - 1):
                        control_images[i] = torch.lerp(control_images[i], empty_frame, i / (1 + control_ease))

        if start_image is None and end_image is None and control_images is not None:
            if control_images.shape[0] >= num_frames:
                control_images = control_images[:num_frames]
            elif control_images.shape[0] < num_frames:
                # padd with empty_frame_level frames
                padding = torch.ones((num_frames - control_images.shape[0], control_images.shape[1], control_images.shape[2], control_images.shape[3]), device=control_images.device) * empty_frame_level
                control_images = torch.cat([control_images, padding], dim=0)
            return (control_images.cpu().float(), torch.zeros_like(control_images[:, :, :, 0]).cpu().float())

        # Convert negative end_index to positive
        if end_index < 0:
            end_index = num_frames + end_index
        
        # Create output batch with empty frames
        out_batch = torch.ones((num_frames, H, W, 3), device=device) * empty_frame_level
        
        # Create mask tensor with proper dimensions
        masks = torch.ones((num_frames, H, W), device=device)
        
        # Pre-process all images at once to avoid redundant work
        if end_image is not None and (end_image.shape[1] != H or end_image.shape[2] != W):
            end_image = common_upscale(end_image.movedim(-1, 1), W, H, "lanczos", "disabled").movedim(1, -1)
        
        if control_images is not None and (control_images.shape[1] != H or control_images.shape[2] != W):
            control_images = common_upscale(control_images.movedim(-1, 1), W, H, "lanczos", "disabled").movedim(1, -1)
        
        # Place start image at start_index
        if start_image is not None:
            frames_to_copy = min(start_image.shape[0], num_frames - start_index)
            if frames_to_copy > 0:
                out_batch[start_index:start_index + frames_to_copy] = start_image[:frames_to_copy]
                masks[start_index:start_index + frames_to_copy] = 0
        
        # Place end image at end_index
        if end_image is not None:
            # Calculate where to start placing end images
            end_start = end_index - end_image.shape[0] + 1
            if end_start < 0:  # Handle case where end images won't all fit
                end_image = end_image[abs(end_start):]
                end_start = 0
                
            frames_to_copy = min(end_image.shape[0], num_frames - end_start)
            if frames_to_copy > 0:
                out_batch[end_start:end_start + frames_to_copy] = end_image[:frames_to_copy]
                masks[end_start:end_start + frames_to_copy] = 0
        
        # Apply control images to remaining frames that don't have start or end images
        if control_images is not None:
            # Create a mask of frames that are still empty (mask == 1)
            empty_frames = masks.sum(dim=(1, 2)) > 0.5 * H * W
            
            if empty_frames.any():
                # Only apply control images where they exist
                control_length = control_images.shape[0]
                for frame_idx in range(num_frames):
                    if empty_frames[frame_idx] and frame_idx < control_length:
                        out_batch[frame_idx] = control_images[frame_idx]
        
        # Apply inpaint mask if provided
        if inpaint_mask is not None:
            inpaint_mask = common_upscale(inpaint_mask.unsqueeze(1), W, H, "nearest-exact", "disabled").squeeze(1).to(device)
            
            # Handle different mask lengths efficiently
            if inpaint_mask.shape[0] > num_frames:
                inpaint_mask = inpaint_mask[:num_frames]
            elif inpaint_mask.shape[0] < num_frames:
                repeat_factor = (num_frames + inpaint_mask.shape[0] - 1) // inpaint_mask.shape[0]  # Ceiling division
                inpaint_mask = inpaint_mask.repeat(repeat_factor, 1, 1)[:num_frames]

            # Apply mask in one operation
            masks = inpaint_mask * masks

        return (out_batch.cpu().float(), masks.cpu().float())


class CreateCFGScheduleFloatList:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
            "steps": ("INT", {"default": 30, "min": 2, "max": 1000, "step": 1, "tooltip": "Number of steps to schedule cfg for"} ),
            "cfg_scale_start": ("FLOAT", {"default": 5.0, "min": 0.0, "max": 30.0, "step": 0.01, "round": 0.01, "tooltip": "CFG scale to use for the steps"}),
            "cfg_scale_end": ("FLOAT", {"default": 5.0, "min": 0.0, "max": 30.0, "step": 0.01, "round": 0.01, "tooltip": "CFG scale to use for the steps"}),
            "interpolation": (["linear", "ease_in", "ease_out"], {"default": "linear", "tooltip": "Interpolation method to use for the cfg scale"}),
            "start_percent": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01,"tooltip": "Start percent of the steps to apply cfg"}),
            "end_percent": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01,"tooltip": "End percent of the steps to apply cfg"}),
            },
            "hidden": {
                "unique_id": "UNIQUE_ID",
            },
        }

    RETURN_TYPES = ("FLOAT", )
    RETURN_NAMES = ("float_list",)
    FUNCTION = "process"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Helper node to generate a list of floats that can be used to schedule cfg scale for the steps, outside the set range cfg is set to 1.0"

    def process(self, steps, cfg_scale_start, cfg_scale_end, interpolation, start_percent, end_percent, unique_id):

        # Create a list of floats for the cfg schedule
        cfg_list = [1.0] * steps
        start_idx = min(int(steps * start_percent), steps - 1)
        end_idx = min(int(steps * end_percent), steps - 1)
        
        for i in range(start_idx, end_idx + 1):
            if i >= steps:
                break
                
            if end_idx == start_idx:
                t = 0
            else:
                t = (i - start_idx) / (end_idx - start_idx)
            
            if interpolation == "linear":
                factor = t
            elif interpolation == "ease_in":
                factor = t * t
            elif interpolation == "ease_out":
                factor = t * (2 - t)
            
            cfg_list[i] = round(cfg_scale_start + factor * (cfg_scale_end - cfg_scale_start), 2)
        
        # If start_percent > 0, always include the first step
        if start_percent > 0:
            cfg_list[0] = 1.0

        if unique_id and PromptServer is not None:
            try:                
                PromptServer.instance.send_progress_text(
                    f"{cfg_list}",
                    unique_id
                )
            except:
                pass

        return (cfg_list,)
    
class CreateScheduleFloatList:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
            "steps": ("INT", {"default": 30, "min": 2, "max": 1000, "step": 1, "tooltip": "Number of steps to schedule cfg for"} ),
            "start_value": ("FLOAT", {"default": 5.0, "min": 0.0, "max": 100.0, "step": 0.01, "round": 0.01, "tooltip": "CFG scale to use for the steps"}),
            "end_value": ("FLOAT", {"default": 5.0, "min": 0.0, "max": 100.0, "step": 0.01, "round": 0.01, "tooltip": "CFG scale to use for the steps"}),
            "default_value": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 0.01, "round": 0.01, "tooltip": "Default value to use for the steps"}),
            "interpolation": (["linear", "ease_in", "ease_out"], {"default": "linear", "tooltip": "Interpolation method to use for the cfg scale"}),
            "start_percent": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01,"tooltip": "Start percent of the steps to apply cfg"}),
            "end_percent": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01,"tooltip": "End percent of the steps to apply cfg"}),
            },
            "hidden": {
                "unique_id": "UNIQUE_ID",
            },
        }

    RETURN_TYPES = ("FLOAT", )
    RETURN_NAMES = ("float_list",)
    FUNCTION = "process"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Helper node to generate a list of floats that can be used to schedule things like cfg and lora scale per step"

    def process(self, steps, start_value, end_value, default_value,interpolation, start_percent, end_percent, unique_id):

        # Create a list of floats for the cfg schedule
        cfg_list = [default_value] * steps
        start_idx = min(int(steps * start_percent), steps - 1)
        end_idx = min(int(steps * end_percent), steps - 1)
        
        for i in range(start_idx, end_idx + 1):
            if i >= steps:
                break
                
            if end_idx == start_idx:
                t = 0
            else:
                t = (i - start_idx) / (end_idx - start_idx)
            
            if interpolation == "linear":
                factor = t
            elif interpolation == "ease_in":
                factor = t * t
            elif interpolation == "ease_out":
                factor = t * (2 - t)

            cfg_list[i] = round(start_value + factor * (end_value - start_value), 2)

        # If start_percent > 0, always include the first step
        if start_percent > 0:
            cfg_list[0] = default_value

        if unique_id and PromptServer is not None:
            try:                
                PromptServer.instance.send_progress_text(
                    f"{cfg_list}",
                    unique_id
                )
            except:
                pass

        return (cfg_list,)
    

class DummyComfyWanModelObject:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
            "shift": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "tooltip": "Sigma shift value"}),
            }
        }

    RETURN_TYPES = ("MODEL", )
    RETURN_NAMES = ("model",)
    FUNCTION = "create"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Helper node to create empty Wan model to use with BasicScheduler -node to get sigmas"

    def create(self, shift):
        from comfy.model_sampling import ModelSamplingDiscreteFlow
        class DummyModel:
            def get_model_object(self, name):
                if name == "model_sampling":
                    model_sampling = ModelSamplingDiscreteFlow()
                    model_sampling.set_parameters(shift=shift)
                    return model_sampling
                return None
        return (DummyModel(),)
    
class WanVideoLatentReScale:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
                    "samples": ("LATENT",),
                    "direction": (["comfy_to_wrapper", "wrapper_to_comfy"], {"tooltip": "Direction to rescale latents, from comfy to wrapper or vice versa"}),
                }
                }

    RETURN_TYPES = ("LATENT",)
    RETURN_NAMES = ("samples",)
    FUNCTION = "encode"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Rescale latents to match the expected range for encoding or decoding between native ComfyUI VAE and the WanVideoWrapper VAE."

    def encode(self, samples, direction):
        samples = samples.copy()
        latents = samples["samples"]

        if latents.shape[1] == 48:
            mean = [
                    -0.2289, -0.0052, -0.1323, -0.2339, -0.2799, 0.0174, 0.1838, 0.1557,
                    -0.1382, 0.0542, 0.2813, 0.0891, 0.1570, -0.0098, 0.0375, -0.1825,
                    -0.2246, -0.1207, -0.0698, 0.5109, 0.2665, -0.2108, -0.2158, 0.2502,
                    -0.2055, -0.0322, 0.1109, 0.1567, -0.0729, 0.0899, -0.2799, -0.1230,
                    -0.0313, -0.1649, 0.0117, 0.0723, -0.2839, -0.2083, -0.0520, 0.3748,
                    0.0152, 0.1957, 0.1433, -0.2944, 0.3573, -0.0548, -0.1681, -0.0667,
                ]
            std = [
                    0.4765, 1.0364, 0.4514, 1.1677, 0.5313, 0.4990, 0.4818, 0.5013,
                    0.8158, 1.0344, 0.5894, 1.0901, 0.6885, 0.6165, 0.8454, 0.4978,
                    0.5759, 0.3523, 0.7135, 0.6804, 0.5833, 1.4146, 0.8986, 0.5659,
                    0.7069, 0.5338, 0.4889, 0.4917, 0.4069, 0.4999, 0.6866, 0.4093,
                    0.5709, 0.6065, 0.6415, 0.4944, 0.5726, 1.2042, 0.5458, 1.6887,
                    0.3971, 1.0600, 0.3943, 0.5537, 0.5444, 0.4089, 0.7468, 0.7744
                ]
        else:
            mean = [
                -0.7571, -0.7089, -0.9113, 0.1075, -0.1745, 0.9653, -0.1517, 1.5508,
                0.4134, -0.0715, 0.5517, -0.3632, -0.1922, -0.9497, 0.2503, -0.2921
            ]
            std = [
                2.8184, 1.4541, 2.3275, 2.6558, 1.2196, 1.7708, 2.6052, 2.0743,
                3.2687, 2.1526, 2.8652, 1.5579, 1.6382, 1.1253, 2.8251, 1.9160
            ]
        mean = torch.tensor(mean).view(1, latents.shape[1], 1, 1, 1)
        std = torch.tensor(std).view(1, latents.shape[1], 1, 1, 1)
        inv_std = (1.0 / std).view(1, latents.shape[1], 1, 1, 1)
        if direction == "comfy_to_wrapper":
            latents = (latents - mean.to(latents)) * inv_std.to(latents)
        elif direction == "wrapper_to_comfy":
            latents = latents / inv_std.to(latents) + mean.to(latents)

        samples["samples"] = latents

        return (samples,)
    
class WanVideoSigmaToStep:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
                "sigma": ("FLOAT", {"default": 0.9, "min": 0.0, "max": 1.0, "step": 0.001}),
            },
        }

    RETURN_TYPES = ("INT", )
    RETURN_NAMES = ("step",)
    FUNCTION = "convert"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Simply passes a float value as an integer, used to set start/end steps with sigma threshold"

    def convert(self, sigma):
        return (sigma,)
    
class NormalizeAudioLoudness:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
            "audio": ("AUDIO",),
            "lufs": ("FLOAT", {"default": -23.0, "min": -100.0, "max": 0.0, "step": 0.1, "tool": "Loudness Units relative to Full Scale, higher LUFS values (closer to 0) mean louder audio. Lower LUFS values (more negative) mean quieter audio."}),
           },
        }

    RETURN_TYPES = ("AUDIO", )
    RETURN_NAMES = ("audio", )
    FUNCTION = "normalize"
    CATEGORY = "WanVideoWrapper"

    def normalize(self, audio, lufs):       
        audio_input = audio["waveform"]
        sample_rate = audio["sample_rate"]
        if audio_input.dim() == 3:
            audio_input = audio_input.squeeze(0) 
        audio_input_np = audio_input.detach().transpose(0, 1).numpy().astype(np.float32)
        audio_input_np = np.ascontiguousarray(audio_input_np)
        normalized_audio = self.loudness_norm(audio_input_np, sr=sample_rate, lufs=lufs)

        out_audio = {"waveform": torch.from_numpy(normalized_audio).transpose(0, 1).unsqueeze(0).float(), "sample_rate": sample_rate}

        return (out_audio, )
    
    def loudness_norm(self, audio_array, sr=16000, lufs=-23):
        try:
            import pyloudnorm
        except:
            raise ImportError("pyloudnorm package is not installed")
        meter = pyloudnorm.Meter(sr)
        loudness = meter.integrated_loudness(audio_array)
        if abs(loudness) > 100:
            return audio_array
        normalized_audio = pyloudnorm.normalize.loudness(audio_array, loudness, lufs)
        return normalized_audio
    
class WanVideoPassImagesFromSamples:
    @classmethod
    def INPUT_TYPES(s):
        return {"required": {
                    "samples": ("LATENT",),
                }
                }

    RETURN_TYPES = ("IMAGE", "STRING",)
    RETURN_NAMES = ("images", "output_path",)
    OUTPUT_TOOLTIPS = ("Decoded images from the samples dictionary", "Output path if provided in the samples dictionary",)
    FUNCTION = "decode"
    CATEGORY = "WanVideoWrapper"
    DESCRIPTION = "Gets possible already decoded images from the samples dictionary, used with Multi/InfiniteTalk sampling"

    def decode(self, samples):
        video = samples.get("video", None)
        video.clamp_(-1.0, 1.0)
        video.add_(1.0).div_(2.0)
        return video.cpu().float(), samples.get("output_path", "")


class FaceMaskFromPoseKeypoints:
    @classmethod
    def INPUT_TYPES(s):
        input_types = {
            "required": {
                "pose_kps": ("POSE_KEYPOINT",),
                "person_index": ("INT", {"default": 0, "min": 0, "max": 100, "step": 1, "tooltip": "Index of the person to start with"}),
            }
        }
        return input_types
    RETURN_TYPES = ("MASK",)
    FUNCTION = "createmask"
    CATEGORY = "ControlNet Preprocessors/Pose Keypoint Postprocess"

    def createmask(self, pose_kps, person_index):
        pose_frames = pose_kps
        prev_center = None
        np_frames = []
        for i, pose_frame in enumerate(pose_frames):
            selected_idx, prev_center = self.select_closest_person(pose_frame, person_index if i == 0 else prev_center)
            np_frames.append(self.draw_kps(pose_frame, selected_idx))
        
        if not np_frames:
            # Handle case where no frames were processed
            log.warning("No valid pose frames found, returning empty mask")
            return (torch.zeros((1, 64, 64), dtype=torch.float32),)
            
        np_frames = np.stack(np_frames, axis=0)
        tensor = torch.from_numpy(np_frames).float() / 255.
        print("tensor.shape:", tensor.shape)
        tensor = tensor[:, :, :, 0]
        return (tensor,)

    def select_closest_person(self, pose_frame, prev_center_or_index):
        people = pose_frame["people"]
        if not people:
            return -1, None
        
        centers = []
        valid_people_indices = []
        
        for idx, person in enumerate(people):
            # Check if face keypoints exist and are valid
            if "face_keypoints_2d" not in person or not person["face_keypoints_2d"]:
                continue
                
            kps = np.array(person["face_keypoints_2d"])
            if len(kps) == 0:
                continue
                
            n = len(kps) // 3
            if n == 0:
                continue
                
            facial_kps = rearrange(kps, "(n c) -> n c", n=n, c=3)[:, :2]
            
            # Check if we have valid coordinates (not all zeros)
            if np.all(facial_kps == 0):
                continue
                
            center = facial_kps.mean(axis=0)
            
            # Check if center is valid (not NaN or infinite)
            if np.isnan(center).any() or np.isinf(center).any():
                continue
                
            centers.append(center)
            valid_people_indices.append(idx)
        
        if not centers:
            return -1, None
            
        if isinstance(prev_center_or_index, (int, np.integer)):
            # First frame: use person_index, but map to valid people
            if 0 <= prev_center_or_index < len(valid_people_indices):
                idx = valid_people_indices[prev_center_or_index]
                return idx, centers[prev_center_or_index]
            elif valid_people_indices:
                # Fallback to first valid person
                idx = valid_people_indices[0]
                return idx, centers[0]
            else:
                return -1, None
        elif prev_center_or_index is not None:
            # Find closest to previous center
            prev_center = np.array(prev_center_or_index)
            dists = [np.linalg.norm(center - prev_center) for center in centers]
            min_idx = int(np.argmin(dists))
            actual_idx = valid_people_indices[min_idx]
            return actual_idx, centers[min_idx]
        else:
            # prev_center_or_index is None, fallback to first valid person
            if valid_people_indices:
                idx = valid_people_indices[0]
                return idx, centers[0]
            else:
                return -1, None

    def draw_kps(self, pose_frame, person_index):
        import cv2
        width, height = pose_frame["canvas_width"], pose_frame["canvas_height"]
        canvas = np.zeros((height, width, 3), dtype=np.uint8)
        people = pose_frame["people"]
        
        if person_index < 0 or person_index >= len(people):
            return canvas  # Out of bounds, return blank
            
        person = people[person_index]
        
        # Check if face keypoints exist and are valid
        if "face_keypoints_2d" not in person or not person["face_keypoints_2d"]:
            return canvas  # No face keypoints, return blank
            
        face_kps_data = person["face_keypoints_2d"]
        if len(face_kps_data) == 0:
            return canvas  # Empty keypoints, return blank
            
        n = len(face_kps_data) // 3
        if n < 17:  # Need at least 17 points for outer contour
            return canvas  # Not enough keypoints, return blank
            
        facial_kps = rearrange(np.array(face_kps_data), "(n c) -> n c", n=n, c=3)[:, :2]
        
        # Check if we have valid coordinates (not all zeros)
        if np.all(facial_kps == 0):
            return canvas  # All keypoints are zero, return blank
            
        # Check for NaN or infinite values
        if np.isnan(facial_kps).any() or np.isinf(facial_kps).any():
            return canvas  # Invalid coordinates, return blank
        
        # Check for negative coordinates or coordinates that would create streaks
        if np.any(facial_kps < 0):
            return canvas  # Negative coordinates, likely bad detection
            
        # Check if coordinates are reasonable (not too close to edges which might indicate bad detection)
        min_margin = 5  # Minimum distance from edges
        if (np.any(facial_kps[:, 0] < min_margin) or 
            np.any(facial_kps[:, 1] < min_margin) or 
            np.any(facial_kps[:, 0] > width - min_margin) or 
            np.any(facial_kps[:, 1] > height - min_margin)):
            # Check if this looks like a streak to corner (many points near 0,0)
            corner_points = np.sum((facial_kps[:, 0] < min_margin) & (facial_kps[:, 1] < min_margin))
            if corner_points > 3:  # Too many points near corner, likely bad detection
                return canvas
                
        facial_kps = facial_kps.astype(np.int32)
        
        # Ensure coordinates are within canvas bounds
        facial_kps[:, 0] = np.clip(facial_kps[:, 0], 0, width - 1)
        facial_kps[:, 1] = np.clip(facial_kps[:, 1], 0, height - 1)
        
        part_color = (255, 255, 255)
        outer_contour = facial_kps[:17]
        
        # Additional validation for the contour before drawing
        # Check if contour points are too spread out (indicating bad detection)
        if len(outer_contour) >= 3:
            # Calculate bounding box of the contour
            min_x, min_y = np.min(outer_contour, axis=0)
            max_x, max_y = np.max(outer_contour, axis=0)
            contour_width = max_x - min_x
            contour_height = max_y - min_y
            
            # If contour spans more than 80% of canvas, likely bad detection
            if (contour_width > 0.8 * width or contour_height > 0.8 * height):
                return canvas
                
            # Check if we have a valid contour (at least 3 unique points)
            unique_points = np.unique(outer_contour, axis=0)
            if len(unique_points) >= 3:
                # Final check: ensure the contour is reasonable
                # Calculate area to see if it's too large or too small
                contour_area = cv2.contourArea(outer_contour)
                canvas_area = width * height
                
                # If contour is less than 0.1% or more than 50% of canvas, skip
                if 0.001 * canvas_area <= contour_area <= 0.5 * canvas_area:
                    cv2.fillPoly(canvas, pts=[outer_contour], color=part_color)
            
        return canvas
    
NODE_CLASS_MAPPINGS = {
    "WanVideoImageResizeToClosest": WanVideoImageResizeToClosest,
    "WanVideoVACEStartToEndFrame": WanVideoVACEStartToEndFrame,
    "ExtractStartFramesForContinuations": ExtractStartFramesForContinuations,
    "CreateCFGScheduleFloatList": CreateCFGScheduleFloatList,
    "DummyComfyWanModelObject": DummyComfyWanModelObject,
    "WanVideoLatentReScale": WanVideoLatentReScale,
    "CreateScheduleFloatList": CreateScheduleFloatList,
    "WanVideoSigmaToStep": WanVideoSigmaToStep,
    "NormalizeAudioLoudness": NormalizeAudioLoudness,
    "WanVideoPassImagesFromSamples": WanVideoPassImagesFromSamples,
    "FaceMaskFromPoseKeypoints": FaceMaskFromPoseKeypoints,
}
NODE_DISPLAY_NAME_MAPPINGS = {
    "WanVideoImageResizeToClosest": "WanVideo Image Resize To Closest",
    "WanVideoVACEStartToEndFrame": "WanVideo VACE Start To End Frame",
    "ExtractStartFramesForContinuations": "Extract Start Frames For Continuations",
    "CreateCFGScheduleFloatList": "Create CFG Schedule Float List",
    "DummyComfyWanModelObject": "Dummy Comfy Wan Model Object",
    "WanVideoLatentReScale": "WanVideo Latent ReScale",
    "CreateScheduleFloatList": "Create Schedule Float List",
    "WanVideoSigmaToStep": "WanVideo Sigma To Step",
    "NormalizeAudioLoudness": "Normalize Audio Loudness",
    "WanVideoPassImagesFromSamples": "WanVideo Pass Images From Samples",
    "FaceMaskFromPoseKeypoints": "Face Mask From Pose Keypoints",
}