(
self,
pipe: WanVideoPipeline,
vace_video, vace_video_mask, vace_reference_image, vace_scale,
height, width, num_frames,
tiled, tile_size, tile_stride
)
| 885 | ) |
| 886 | |
| 887 | def process( |
| 888 | self, |
| 889 | pipe: WanVideoPipeline, |
| 890 | vace_video, vace_video_mask, vace_reference_image, vace_scale, |
| 891 | height, width, num_frames, |
| 892 | tiled, tile_size, tile_stride |
| 893 | ): |
| 894 | if vace_video is not None or vace_video_mask is not None or vace_reference_image is not None: |
| 895 | pipe.load_models_to_device(["vae"]) |
| 896 | if vace_video is None: |
| 897 | vace_video = torch.zeros((1, 3, num_frames, height, width), dtype=pipe.torch_dtype, device=pipe.device) |
| 898 | else: |
| 899 | vace_video = pipe.preprocess_video(vace_video) |
| 900 | |
| 901 | if vace_video_mask is None: |
| 902 | vace_video_mask = torch.ones_like(vace_video) |
| 903 | else: |
| 904 | vace_video_mask = pipe.preprocess_video(vace_video_mask, min_value=0, max_value=1) |
| 905 | |
| 906 | inactive = vace_video * (1 - vace_video_mask) + 0 * vace_video_mask |
| 907 | reactive = vace_video * vace_video_mask + 0 * (1 - vace_video_mask) |
| 908 | inactive = pipe.vae.encode(inactive, device=pipe.device, tiled=tiled, tile_size=tile_size, tile_stride=tile_stride).to(dtype=pipe.torch_dtype, device=pipe.device) |
| 909 | reactive = pipe.vae.encode(reactive, device=pipe.device, tiled=tiled, tile_size=tile_size, tile_stride=tile_stride).to(dtype=pipe.torch_dtype, device=pipe.device) |
| 910 | vace_video_latents = torch.concat((inactive, reactive), dim=1) |
| 911 | |
| 912 | vace_mask_latents = rearrange(vace_video_mask[0,0], "T (H P) (W Q) -> 1 (P Q) T H W", P=8, Q=8) |
| 913 | vace_mask_latents = torch.nn.functional.interpolate(vace_mask_latents, size=((vace_mask_latents.shape[2] + 3) // 4, vace_mask_latents.shape[3], vace_mask_latents.shape[4]), mode='nearest-exact') |
| 914 | |
| 915 | if vace_reference_image is None: |
| 916 | pass |
| 917 | else: |
| 918 | if not isinstance(vace_reference_image,list): |
| 919 | vace_reference_image = [vace_reference_image] |
| 920 | |
| 921 | vace_reference_image = pipe.preprocess_video(vace_reference_image) |
| 922 | |
| 923 | bs, c, f, h, w = vace_reference_image.shape |
| 924 | new_vace_ref_images = [] |
| 925 | for j in range(f): |
| 926 | new_vace_ref_images.append(vace_reference_image[0, :, j:j+1]) |
| 927 | vace_reference_image = new_vace_ref_images |
| 928 | |
| 929 | vace_reference_latents = pipe.vae.encode(vace_reference_image, device=pipe.device, tiled=tiled, tile_size=tile_size, tile_stride=tile_stride).to(dtype=pipe.torch_dtype, device=pipe.device) |
| 930 | vace_reference_latents = torch.concat((vace_reference_latents, torch.zeros_like(vace_reference_latents)), dim=1) |
| 931 | vace_reference_latents = [u.unsqueeze(0) for u in vace_reference_latents] |
| 932 | |
| 933 | vace_video_latents = torch.concat((*vace_reference_latents, vace_video_latents), dim=2) |
| 934 | vace_mask_latents = torch.concat((torch.zeros_like(vace_mask_latents[:, :, :f]), vace_mask_latents), dim=2) |
| 935 | |
| 936 | vace_context = torch.concat((vace_video_latents, vace_mask_latents), dim=1) |
| 937 | return {"vace_context": vace_context, "vace_scale": vace_scale} |
| 938 | else: |
| 939 | return {"vace_context": None, "vace_scale": vace_scale} |
| 940 | |
| 941 | class WanVideoUnit_VAP(PipelineUnit): |
| 942 | def __init__(self): |
nothing calls this directly
no test coverage detected