musev/pipelines/pipeline_controlnet_predictor.py

import copy
from typing import Any, Callable, Dict, Iterable, Union
import PIL
import cv2
import torch
import argparse
import datetime
import logging
import inspect
import math
import os
import shutil
from typing import Dict, List, Optional, Tuple
from pprint import pformat, pprint
from collections import OrderedDict
from dataclasses import dataclass
import gc
import time

import numpy as np
from omegaconf import OmegaConf
from omegaconf import SCMode
import torch
from torch import nn
import torch.nn.functional as F
import torch.utils.checkpoint
from einops import rearrange, repeat
import pandas as pd
import h5py
from diffusers.models.autoencoder_kl import AutoencoderKL

from diffusers.models.modeling_utils import load_state_dict
from diffusers.utils import (
    logging,
    BaseOutput,
    logging,
)
from diffusers.utils.dummy_pt_objects import ConsistencyDecoderVAE
from diffusers.utils.import_utils import is_xformers_available

from mmcm.utils.seed_util import set_all_seed
from mmcm.vision.data.video_dataset import DecordVideoDataset
from mmcm.vision.process.correct_color import hist_match_video_bcthw
from mmcm.vision.process.image_process import (
    batch_dynamic_crop_resize_images,
    batch_dynamic_crop_resize_images_v2,
)
from mmcm.vision.utils.data_type_util import is_video
from mmcm.vision.feature_extractor.controlnet import load_controlnet_model

from ..schedulers import (
    EulerDiscreteScheduler,
    LCMScheduler,
    DDIMScheduler,
    DDPMScheduler,
)
from ..models.unet_3d_condition import UNet3DConditionModel
from .pipeline_controlnet import (
    MusevControlNetPipeline,
    VideoPipelineOutput as PipelineVideoPipelineOutput,
)
from ..utils.util import save_videos_grid_with_opencv
from ..utils.model_util import (
    update_pipeline_basemodel,
    update_pipeline_lora_model,
    update_pipeline_lora_models,
    update_pipeline_model_parameters,
)

logger = logging.get_logger(__name__)  # pylint: disable=invalid-name


@dataclass
class VideoPipelineOutput(BaseOutput):
    videos: Union[torch.Tensor, np.ndarray]
    latents: Union[torch.Tensor, np.ndarray]
    videos_mid: Union[torch.Tensor, np.ndarray]
    controlnet_cond: Union[torch.Tensor, np.ndarray]
    generated_videos: Union[torch.Tensor, np.ndarray]


def update_controlnet_processor_params(
    src: Union[Dict, List[Dict]], dst: Union[Dict, List[Dict]]
):
    """merge dst into src"""
    if isinstance(src, list) and not isinstance(dst, List):
        dst = [dst] * len(src)
    if isinstance(src, list) and isinstance(dst, list):
        return [
            update_controlnet_processor_params(src[i], dst[i]) for i in range(len(src))
        ]
    if src is None:
        dct = {}
    else:
        dct = copy.deepcopy(src)
    if dst is None:
        dst = {}
    dct.update(dst)
    return dct


class DiffusersPipelinePredictor(object):
    """wraper of diffusers pipeline, support generation function interface. support
    1. text2video: inputs include text, image(optional), refer_image(optional)
    2. video2video:
        1. use controlnet to control spatial
        2. or use video fuse noise to denoise
    """

    def __init__(
        self,
        sd_model_path: str,
        unet: nn.Module,
        controlnet_name: Union[str, List[str]] = None,
        controlnet: nn.Module = None,
        lora_dict: Dict[str, Dict] = None,
        requires_safety_checker: bool = False,
        device: str = "cuda",
        dtype: torch.dtype = torch.float16,
        # controlnet parameters start
        need_controlnet_processor: bool = True,
        need_controlnet: bool = True,
        image_resolution: int = 512,
        detect_resolution: int = 512,
        include_body: bool = True,
        hand_and_face: bool = None,
        include_face: bool = False,
        include_hand: bool = True,
        negative_embedding: List = None,
        # controlnet parameters end
        enable_xformers_memory_efficient_attention: bool = True,
        lcm_lora_dct: Dict = None,
        referencenet: nn.Module = None,
        ip_adapter_image_proj: nn.Module = None,
        vision_clip_extractor: nn.Module = None,
        face_emb_extractor: nn.Module = None,
        facein_image_proj: nn.Module = None,
        ip_adapter_face_emb_extractor: nn.Module = None,
        ip_adapter_face_image_proj: nn.Module = None,
        vae_model: Optional[Tuple[nn.Module, str]] = None,
        pose_guider: Optional[nn.Module] = None,
        enable_zero_snr: bool = False,
    ) -> None:
        self.sd_model_path = sd_model_path
        self.unet = unet
        self.controlnet_name = controlnet_name
        self.controlnet = controlnet
        self.requires_safety_checker = requires_safety_checker
        self.device = device
        self.dtype = dtype
        self.need_controlnet_processor = need_controlnet_processor
        self.need_controlnet = need_controlnet
        self.need_controlnet_processor = need_controlnet_processor
        self.image_resolution = image_resolution
        self.detect_resolution = detect_resolution
        self.include_body = include_body
        self.hand_and_face = hand_and_face
        self.include_face = include_face
        self.include_hand = include_hand
        self.negative_embedding = negative_embedding
        self.device = device
        self.dtype = dtype
        self.lcm_lora_dct = lcm_lora_dct
        if controlnet is None and controlnet_name is not None:
            controlnet, controlnet_processor, processor_params = load_controlnet_model(
                controlnet_name,
                device=device,
                dtype=dtype,
                need_controlnet_processor=need_controlnet_processor,
                need_controlnet=need_controlnet,
                image_resolution=image_resolution,
                detect_resolution=detect_resolution,
                include_body=include_body,
                include_face=include_face,
                hand_and_face=hand_and_face,
                include_hand=include_hand,
            )
            self.controlnet_processor = controlnet_processor
            self.controlnet_processor_params = processor_params
            logger.debug(f"init controlnet controlnet_name={controlnet_name}")

        if controlnet is not None:
            controlnet = controlnet.to(device=device, dtype=dtype)
            controlnet.eval()
        if pose_guider is not None:
            pose_guider = pose_guider.to(device=device, dtype=dtype)
            pose_guider.eval()
        unet.to(device=device, dtype=dtype)
        unet.eval()
        if referencenet is not None:
            referencenet.to(device=device, dtype=dtype)
            referencenet.eval()
        if ip_adapter_image_proj is not None:
            ip_adapter_image_proj.to(device=device, dtype=dtype)
            ip_adapter_image_proj.eval()
        if vision_clip_extractor is not None:
            vision_clip_extractor.to(device=device, dtype=dtype)
            vision_clip_extractor.eval()
        if face_emb_extractor is not None:
            face_emb_extractor.to(device=device, dtype=dtype)
            face_emb_extractor.eval()
        if facein_image_proj is not None:
            facein_image_proj.to(device=device, dtype=dtype)
            facein_image_proj.eval()

        if isinstance(vae_model, str):
            # TODO: poor implementation, to improve
            if "consistency" in vae_model:
                vae = ConsistencyDecoderVAE.from_pretrained(vae_model)
            else:
                vae = AutoencoderKL.from_pretrained(vae_model)
        elif isinstance(vae_model, nn.Module):
            vae = vae_model
        else:
            vae = None
        if vae is not None:
            vae.to(device=device, dtype=dtype)
            vae.eval()
        if ip_adapter_face_emb_extractor is not None:
            ip_adapter_face_emb_extractor.to(device=device, dtype=dtype)
            ip_adapter_face_emb_extractor.eval()
        if ip_adapter_face_image_proj is not None:
            ip_adapter_face_image_proj.to(device=device, dtype=dtype)
            ip_adapter_face_image_proj.eval()
        params = {
            "pretrained_model_name_or_path": sd_model_path,
            "controlnet": controlnet,
            "unet": unet,
            "requires_safety_checker": requires_safety_checker,
            "torch_dtype": dtype,
            "torch_device": device,
            "referencenet": referencenet,
            "ip_adapter_image_proj": ip_adapter_image_proj,
            "vision_clip_extractor": vision_clip_extractor,
            "facein_image_proj": facein_image_proj,
            "face_emb_extractor": face_emb_extractor,
            "ip_adapter_face_emb_extractor": ip_adapter_face_emb_extractor,
            "ip_adapter_face_image_proj": ip_adapter_face_image_proj,
            "pose_guider": pose_guider,
        }
        if vae is not None:
            params["vae"] = vae
        pipeline = MusevControlNetPipeline.from_pretrained(**params)
        pipeline = pipeline.to(torch_device=device, torch_dtype=dtype)
        logger.debug(
            f"init pipeline from sd_model_path={sd_model_path}, device={device}, dtype={dtype}"
        )
        if (
            negative_embedding is not None
            and pipeline.text_encoder is not None
            and pipeline.tokenizer is not None
        ):
            for neg_emb_path, neg_token in negative_embedding:
                pipeline.load_textual_inversion(neg_emb_path, token=neg_token)

        # pipe.scheduler = UniPCMultistepScheduler.from_config(pipe.scheduler.config)
        # pipe.enable_model_cpu_offload()
        if not enable_zero_snr:
            pipeline.scheduler = EulerDiscreteScheduler.from_config(
                pipeline.scheduler.config
            )
            # pipeline.scheduler = DDIMScheduler.from_config(
            #     pipeline.scheduler.config,
            # 该部分会影响生成视频的亮度，不适用于首帧给定的视频生成
            # this part will change brightness of video, not suitable for image2video mode
            # rescale_betas_zero_snr affect the brightness of the generated video, not suitable for vision condition images mode
            #     # rescale_betas_zero_snr=True,
            # )
            # pipeline.scheduler = DDPMScheduler.from_config(pipeline.scheduler.config)
        else:
            # moore scheduler, just for codetest
            pipeline.scheduler = DDIMScheduler(
                beta_start=0.00085,
                beta_end=0.012,
                beta_schedule="linear",
                clip_sample=False,
                steps_offset=1,
                ### Zero-SNR params
                prediction_type="v_prediction",
                rescale_betas_zero_snr=True,
                timestep_spacing="trailing",
            )

        pipeline.enable_vae_slicing()
        self.enable_xformers_memory_efficient_attention = (
            enable_xformers_memory_efficient_attention
        )
        if enable_xformers_memory_efficient_attention:
            if is_xformers_available():
                pipeline.enable_xformers_memory_efficient_attention()
            else:
                raise ValueError(
                    "xformers is not available. Make sure it is installed correctly"
                )
        self.pipeline = pipeline
        self.unload_dict = []  # keep lora state
        if lora_dict is not None:
            self.load_lora(lora_dict=lora_dict)
            logger.debug("load lora {}".format(" ".join(list(lora_dict.keys()))))

        if lcm_lora_dct is not None:
            self.pipeline.scheduler = LCMScheduler.from_config(
                self.pipeline.scheduler.config
            )
            self.load_lora(lora_dict=lcm_lora_dct)
            logger.debug("load lcm lora {}".format(" ".join(list(lcm_lora_dct.keys()))))

        # logger.debug("Unet3Model Parameters")
        # logger.debug(pformat(self.__dict__))

    def load_lora(
        self,
        lora_dict: Dict[str, Dict],
    ):
        self.pipeline, unload_dict = update_pipeline_lora_models(
            self.pipeline, lora_dict, device=self.device
        )
        self.unload_dict += unload_dict

    def unload_lora(self):
        for layer_data in self.unload_dict:
            layer = layer_data["layer"]
            added_weight = layer_data["added_weight"]
            layer.weight.data -= added_weight
        self.unload_dict = []
        gc.collect()
        torch.cuda.empty_cache()

    def update_unet(self, unet: nn.Module):
        self.pipeline.unet = unet.to(device=self.device, dtype=self.dtype)

    def update_sd_model(self, model_path: str, text_model_path: str):
        self.pipeline = update_pipeline_basemodel(
            self.pipeline,
            model_path,
            text_sd_model_path=text_model_path,
            device=self.device,
        )

    def update_sd_model_and_unet(
        self, lora_sd_path: str, lora_path: str, sd_model_path: str = None
    ):
        self.pipeline = update_pipeline_model_parameters(
            self.pipeline,
            model_path=lora_sd_path,
            lora_path=lora_path,
            text_model_path=sd_model_path,
            device=self.device,
        )

    def update_controlnet(self, controlnet_name=Union[str, List[str]]):
        self.pipeline.controlnet = load_controlnet_model(controlnet_name).to(
            device=self.device, dtype=self.dtype
        )

    def run_pipe_text2video(
        self,
        video_length: int,
        prompt: Union[str, List[str]] = None,
        # b c t h w
        height: Optional[int] = None,
        width: Optional[int] = None,
        video_num_inference_steps: int = 50,
        video_guidance_scale: float = 7.5,
        video_guidance_scale_end: float = 3.5,
        video_guidance_scale_method: str = "linear",
        strength: float = 0.8,
        video_negative_prompt: Optional[Union[str, List[str]]] = None,
        negative_prompt: Optional[Union[str, List[str]]] = None,
        num_videos_per_prompt: Optional[int] = 1,
        eta: float = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        same_seed: Optional[Union[int, List[int]]] = None,
        # b c t(1) ho wo
        condition_latents: Optional[torch.FloatTensor] = None,
        latents: Optional[torch.FloatTensor] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        guidance_scale: float = 7.5,
        num_inference_steps: int = 50,
        output_type: Optional[str] = "tensor",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
        need_middle_latents: bool = False,
        w_ind_noise: float = 0.5,
        initial_common_latent: Optional[torch.FloatTensor] = None,
        latent_index: torch.LongTensor = None,
        vision_condition_latent_index: torch.LongTensor = None,
        n_vision_condition: int = 1,
        noise_type: str = "random",
        max_batch_num: int = 30,
        need_img_based_video_noise: bool = False,
        condition_images: torch.Tensor = None,
        fix_condition_images: bool = False,
        redraw_condition_image: bool = False,
        img_weight: float = 1e-3,
        motion_speed: float = 8.0,
        need_hist_match: bool = False,
        refer_image: Optional[
            Tuple[np.ndarray, torch.Tensor, List[str], List[np.ndarray]]
        ] = None,
        ip_adapter_image: Optional[Tuple[torch.Tensor, np.array]] = None,
        fixed_refer_image: bool = True,
        fixed_ip_adapter_image: bool = True,
        redraw_condition_image_with_ipdapter: bool = True,
        redraw_condition_image_with_referencenet: bool = True,
        refer_face_image: Optional[Tuple[torch.Tensor, np.array]] = None,
        fixed_refer_face_image: bool = True,
        redraw_condition_image_with_facein: bool = True,
        ip_adapter_scale: float = 1.0,
        redraw_condition_image_with_ip_adapter_face: bool = True,
        facein_scale: float = 1.0,
        ip_adapter_face_scale: float = 1.0,
        prompt_only_use_image_prompt: bool = False,
        # serial_denoise parameter start
        record_mid_video_noises: bool = False,
        record_mid_video_latents: bool = False,
        video_overlap: int = 1,
        # serial_denoise parameter end
        # parallel_denoise parameter start
        context_schedule="uniform",
        context_frames=12,
        context_stride=1,
        context_overlap=4,
        context_batch_size=1,
        interpolation_factor=1,
        # parallel_denoise parameter end
    ):
        """
        generate long video with end2end mode
        1. prepare vision condition image by assingning, redraw, or generation with text2image module with skip_temporal_layer=True;
        2. use image or latest of vision condition image to generate first shot;
        3. use last n (1) image or last latent of last shot as new vision condition latent to generate next shot
        4. repeat n_batch times between 2 and 3

        类似img2img pipeline
        refer_image和ip_adapter_image的来源：
        1. 输入给定；
        2. 当未输入时，纯text2video生成首帧，并赋值更新refer_image和ip_adapter_image;
        3. 当有输入，但是因为redraw更新了首帧时，也需要赋值更新refer_image和ip_adapter_image;

        refer_image和ip_adapter_image的作用：
        1. 当无首帧图像时，用于生成首帧；
        2. 用于生成视频。


        similar to diffusers img2img pipeline.
        three ways to prepare refer_image  and ip_adapter_image
        1. from input parameter
        2. when input paramter is None, use text2video to generate vis cond image, and use as refer_image and ip_adapter_image too.
        3. given from input paramter, but still redraw, update with redrawn vis cond image.
        """
        run_video_length = video_length
        # generate vision condition frame start
        # if condition_images is None, generate with refer_image, ip_adapter_image
        # if condition_images not None and need redraw, according to redraw_condition_image_with_ipdapter, redraw_condition_image_with_referencenet, refer_image, ip_adapter_image
        if n_vision_condition > 0:
            if condition_images is None and condition_latents is None:
                logger.debug("run_pipe_text2video, generate first_image")
                (
                    condition_images,
                    condition_latents,
                    _,
                    _,
                    _,
                ) = self.pipeline(
                    prompt=prompt,
                    num_inference_steps=num_inference_steps,
                    guidance_scale=guidance_scale,
                    negative_prompt=negative_prompt,
                    video_length=1,
                    height=height,
                    width=width,
                    return_dict=False,
                    skip_temporal_layer=True,
                    output_type="np",
                    generator=generator,
                    w_ind_noise=w_ind_noise,
                    need_img_based_video_noise=need_img_based_video_noise,
                    refer_image=refer_image
                    if redraw_condition_image_with_referencenet
                    else None,
                    ip_adapter_image=ip_adapter_image
                    if redraw_condition_image_with_ipdapter
                    else None,
                    refer_face_image=refer_face_image
                    if redraw_condition_image_with_facein
                    else None,
                    ip_adapter_scale=ip_adapter_scale,
                    facein_scale=facein_scale,
                    ip_adapter_face_scale=ip_adapter_face_scale,
                    ip_adapter_face_image=refer_face_image
                    if redraw_condition_image_with_ip_adapter_face
                    else None,
                    prompt_only_use_image_prompt=prompt_only_use_image_prompt,
                )
                run_video_length = video_length - 1
            elif (
                condition_images is not None
                and redraw_condition_image
                and condition_latents is None
            ):
                logger.debug("run_pipe_text2video, redraw first_image")

                (
                    condition_images,
                    condition_latents,
                    _,
                    _,
                    _,
                ) = self.pipeline(
                    prompt=prompt,
                    image=condition_images,
                    num_inference_steps=num_inference_steps,
                    guidance_scale=guidance_scale,
                    negative_prompt=negative_prompt,
                    strength=strength,
                    video_length=condition_images.shape[2],
                    height=height,
                    width=width,
                    return_dict=False,
                    skip_temporal_layer=True,
                    output_type="np",
                    generator=generator,
                    w_ind_noise=w_ind_noise,
                    need_img_based_video_noise=need_img_based_video_noise,
                    refer_image=refer_image
                    if redraw_condition_image_with_referencenet
                    else None,
                    ip_adapter_image=ip_adapter_image
                    if redraw_condition_image_with_ipdapter
                    else None,
                    refer_face_image=refer_face_image
                    if redraw_condition_image_with_facein
                    else None,
                    ip_adapter_scale=ip_adapter_scale,
                    facein_scale=facein_scale,
                    ip_adapter_face_scale=ip_adapter_face_scale,
                    ip_adapter_face_image=refer_face_image
                    if redraw_condition_image_with_ip_adapter_face
                    else None,
                    prompt_only_use_image_prompt=prompt_only_use_image_prompt,
                )
        else:
            condition_images = None
            condition_latents = None
        # generate vision condition frame end

        # refer_image and ip_adapter_image, update mode from 2 and 3 as mentioned above start
        if (
            refer_image is not None
            and redraw_condition_image
            and condition_images is not None
        ):
            refer_image = condition_images * 255.0
            logger.debug(f"update refer_image because of redraw_condition_image")
        elif (
            refer_image is None
            and self.pipeline.referencenet is not None
            and condition_images is not None
        ):
            refer_image = condition_images * 255.0
            logger.debug(f"update refer_image because of generate first_image")

        # ipadapter_image
        if (
            ip_adapter_image is not None
            and redraw_condition_image
            and condition_images is not None
        ):
            ip_adapter_image = condition_images * 255.0
            logger.debug(f"update ip_adapter_image because of redraw_condition_image")
        elif (
            ip_adapter_image is None
            and self.pipeline.ip_adapter_image_proj is not None
            and condition_images is not None
        ):
            ip_adapter_image = condition_images * 255.0
            logger.debug(f"update ip_adapter_image because of generate first_image")
        # refer_image and ip_adapter_image, update mode from 2 and 3 as mentioned above end

        # refer_face_image, update mode from 2 and 3 as mentioned above start
        if (
            refer_face_image is not None
            and redraw_condition_image
            and condition_images is not None
        ):
            refer_face_image = condition_images * 255.0
            logger.debug(f"update refer_face_image because of redraw_condition_image")
        elif (
            refer_face_image is None
            and self.pipeline.facein_image_proj is not None
            and condition_images is not None
        ):
            refer_face_image = condition_images * 255.0
            logger.debug(f"update face_image because of generate first_image")
            # refer_face_image, update mode from 2 and 3 as mentioned above end

        last_mid_video_noises = None
        last_mid_video_latents = None
        initial_common_latent = None

        out_videos = []
        for i_batch in range(max_batch_num):
            logger.debug(f"sd_pipeline_predictor, run_pipe_text2video: {i_batch}")
            if max_batch_num is not None and i_batch == max_batch_num:
                break

            if i_batch == 0:
                result_overlap = 0
            else:
                if n_vision_condition > 0:
                    # ignore condition_images if condition_latents is not None in pipeline
                    if not fix_condition_images:
                        logger.debug(f"{i_batch}, update condition_latents")
                        condition_latents = out_latents_batch[
                            :, :, -n_vision_condition:, :, :
                        ]
                    else:
                        logger.debug(f"{i_batch}, do not update condition_latents")
                result_overlap = n_vision_condition

                if not fixed_refer_image and n_vision_condition > 0:
                    logger.debug("ref_image use last frame of last generated out video")
                    refer_image = out_batch[:, :, -n_vision_condition:, :, :] * 255.0
                else:
                    logger.debug("use given fixed ref_image")

                if not fixed_ip_adapter_image and n_vision_condition > 0:
                    logger.debug(
                        "ip_adapter_image use last frame of last generated out video"
                    )
                    ip_adapter_image = (
                        out_batch[:, :, -n_vision_condition:, :, :] * 255.0
                    )
                else:
                    logger.debug("use given fixed ip_adapter_image")

                if not fixed_refer_face_image and n_vision_condition > 0:
                    logger.debug(
                        "refer_face_image use last frame of last generated out video"
                    )
                    refer_face_image = (
                        out_batch[:, :, -n_vision_condition:, :, :] * 255.0
                    )
                else:
                    logger.debug("use given fixed ip_adapter_image")

                run_video_length = video_length
            if same_seed is not None:
                _, generator = set_all_seed(same_seed)

            out = self.pipeline(
                video_length=run_video_length,  # int
                prompt=prompt,
                num_inference_steps=video_num_inference_steps,
                height=height,
                width=width,
                generator=generator,
                condition_images=condition_images,
                condition_latents=condition_latents,  # b co t(1) ho wo
                skip_temporal_layer=False,
                output_type="np",
                noise_type=noise_type,
                negative_prompt=video_negative_prompt,
                guidance_scale=video_guidance_scale,
                guidance_scale_end=video_guidance_scale_end,
                guidance_scale_method=video_guidance_scale_method,
                w_ind_noise=w_ind_noise,
                need_img_based_video_noise=need_img_based_video_noise,
                img_weight=img_weight,
                motion_speed=motion_speed,
                vision_condition_latent_index=vision_condition_latent_index,
                refer_image=refer_image,
                ip_adapter_image=ip_adapter_image,
                refer_face_image=refer_face_image,
                ip_adapter_scale=ip_adapter_scale,
                facein_scale=facein_scale,
                ip_adapter_face_scale=ip_adapter_face_scale,
                ip_adapter_face_image=refer_face_image,
                prompt_only_use_image_prompt=prompt_only_use_image_prompt,
                initial_common_latent=initial_common_latent,
                # serial_denoise parameter start
                record_mid_video_noises=record_mid_video_noises,
                last_mid_video_noises=last_mid_video_noises,
                record_mid_video_latents=record_mid_video_latents,
                last_mid_video_latents=last_mid_video_latents,
                video_overlap=video_overlap,
                # serial_denoise parameter end
                # parallel_denoise parameter start
                context_schedule=context_schedule,
                context_frames=context_frames,
                context_stride=context_stride,
                context_overlap=context_overlap,
                context_batch_size=context_batch_size,
                interpolation_factor=interpolation_factor,
                # parallel_denoise parameter end
            )
            logger.debug(
                f"run_pipe_text2video, out.videos.shape, i_batch={i_batch}, videos={out.videos.shape}, result_overlap={result_overlap}"
            )
            out_batch = out.videos[:, :, result_overlap:, :, :]
            out_latents_batch = out.latents[:, :, result_overlap:, :, :]
            out_videos.append(out_batch)

        out_videos = np.concatenate(out_videos, axis=2)
        if need_hist_match:
            out_videos[:, :, 1:, :, :] = hist_match_video_bcthw(
                out_videos[:, :, 1:, :, :], out_videos[:, :, :1, :, :], value=255.0
            )
        return out_videos

    def run_pipe_with_latent_input(
        self,
    ):
        pass

    def run_pipe_middle2video_with_middle(self, middle: Tuple[str, Iterable]):
        pass

    def run_pipe_video2video(
        self,
        video: Tuple[str, Iterable],
        time_size: int = None,
        sample_rate: int = None,
        overlap: int = None,
        step: int = None,
        prompt: Union[str, List[str]] = None,
        # b c t h w
        height: Optional[int] = None,
        width: Optional[int] = None,
        num_inference_steps: int = 50,
        video_num_inference_steps: int = 50,
        guidance_scale: float = 7.5,
        video_guidance_scale: float = 7.5,
        video_guidance_scale_end: float = 3.5,
        video_guidance_scale_method: str = "linear",
        video_negative_prompt: Optional[Union[str, List[str]]] = None,
        negative_prompt: Optional[Union[str, List[str]]] = None,
        num_videos_per_prompt: Optional[int] = 1,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        eta: float = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        controlnet_latents: Union[torch.FloatTensor, np.ndarray] = None,
        # b c t(1) hi wi
        controlnet_condition_images: Optional[torch.FloatTensor] = None,
        # b c t(1) ho wo
        controlnet_condition_latents: Optional[torch.FloatTensor] = None,
        # b c t(1) ho wo
        condition_latents: Optional[torch.FloatTensor] = None,
        condition_images: Optional[torch.FloatTensor] = None,
        fix_condition_images: bool = False,
        latents: Optional[torch.FloatTensor] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        output_type: Optional[str] = "tensor",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
        controlnet_conditioning_scale: Union[float, List[float]] = 1.0,
        guess_mode: bool = False,
        control_guidance_start: Union[float, List[float]] = 0.0,
        control_guidance_end: Union[float, List[float]] = 1.0,
        need_middle_latents: bool = False,
        w_ind_noise: float = 0.5,
        img_weight: float = 0.001,
        initial_common_latent: Optional[torch.FloatTensor] = None,
        latent_index: torch.LongTensor = None,
        vision_condition_latent_index: torch.LongTensor = None,
        noise_type: str = "random",
        controlnet_processor_params: Dict = None,
        need_return_videos: bool = False,
        need_return_condition: bool = False,
        max_batch_num: int = 30,
        strength: float = 0.8,
        video_strength: float = 0.8,
        need_video2video: bool = False,
        need_img_based_video_noise: bool = False,
        need_hist_match: bool = False,
        end_to_end: bool = True,
        refer_image: Optional[
            Tuple[np.ndarray, torch.Tensor, List[str], List[np.ndarray]]
        ] = None,
        ip_adapter_image: Optional[Tuple[torch.Tensor, np.array]] = None,
        fixed_refer_image: bool = True,
        fixed_ip_adapter_image: bool = True,
        redraw_condition_image: bool = False,
        redraw_condition_image_with_ipdapter: bool = True,
        redraw_condition_image_with_referencenet: bool = True,
        refer_face_image: Optional[Tuple[torch.Tensor, np.array]] = None,
        fixed_refer_face_image: bool = True,
        redraw_condition_image_with_facein: bool = True,
        ip_adapter_scale: float = 1.0,
        facein_scale: float = 1.0,
        ip_adapter_face_scale: float = 1.0,
        redraw_condition_image_with_ip_adapter_face: bool = False,
        n_vision_condition: int = 1,
        prompt_only_use_image_prompt: bool = False,
        motion_speed: float = 8.0,
        # serial_denoise parameter start
        record_mid_video_noises: bool = False,
        record_mid_video_latents: bool = False,
        video_overlap: int = 1,
        # serial_denoise parameter end
        # parallel_denoise parameter start
        context_schedule="uniform",
        context_frames=12,
        context_stride=1,
        context_overlap=4,
        context_batch_size=1,
        interpolation_factor=1,
        # parallel_denoise parameter end
        # 支持 video_path 时多种输入
        # TODO:// video_has_condition =False，当且仅支持 video_is_middle=True, 待后续重构
        # TODO:// when video_has_condition =False, video_is_middle should be True.
        video_is_middle: bool = False,
        video_has_condition: bool = True,
    ):
        """
        类似controlnet text2img pipeline。 输入视频，用视频得到controlnet condition。
        目前仅支持time_size == step，overlap=0
        输出视频长度=输入视频长度

        similar to controlnet text2image pipeline, generate video with controlnet condition from given video.
        By now, sliding window only support time_size == step, overlap = 0.
        """
        if isinstance(video, str):
            video_reader = DecordVideoDataset(
                video,
                time_size=time_size,
                step=step,
                overlap=overlap,
                sample_rate=sample_rate,
                device="cpu",
                data_type="rgb",
                channels_order="c t h w",
                drop_last=True,
            )
        else:
            video_reader = video
        videos = [] if need_return_videos else None
        out_videos = []
        out_condition = (
            []
            if need_return_condition and self.pipeline.controlnet is not None
            else None
        )
        # crop resize images
        if condition_images is not None:
            logger.debug(
                f"center crop resize condition_images={condition_images.shape}, to height={height}, width={width}"
            )
            condition_images = batch_dynamic_crop_resize_images_v2(
                condition_images,
                target_height=height,
                target_width=width,
            )
        if refer_image is not None:
            logger.debug(
                f"center crop resize refer_image to height={height}, width={width}"
            )
            refer_image = batch_dynamic_crop_resize_images_v2(
                refer_image,
                target_height=height,
                target_width=width,
            )
        if ip_adapter_image is not None:
            logger.debug(
                f"center crop resize ip_adapter_image to height={height}, width={width}"
            )
            ip_adapter_image = batch_dynamic_crop_resize_images_v2(
                ip_adapter_image,
                target_height=height,
                target_width=width,
            )
        if refer_face_image is not None:
            logger.debug(
                f"center crop resize refer_face_image to height={height}, width={width}"
            )
            refer_face_image = batch_dynamic_crop_resize_images_v2(
                refer_face_image,
                target_height=height,
                target_width=width,
            )
        first_image = None
        last_mid_video_noises = None
        last_mid_video_latents = None
        initial_common_latent = None
        # initial_common_latent = torch.randn((1, 4, 1, 112, 64)).to(
        #     device=self.device, dtype=self.dtype
        # )

        for i_batch, item in enumerate(video_reader):
            logger.debug(f"\n sd_pipeline_predictor, run_pipe_video2video: {i_batch}")
            if max_batch_num is not None and i_batch == max_batch_num:
                break
            # read and prepare video batch
            batch = item.data
            batch = batch_dynamic_crop_resize_images(
                batch,
                target_height=height,
                target_width=width,
            )

            batch = batch[np.newaxis, ...]
            batch_size, channel, video_length, video_height, video_width = batch.shape
            # extract controlnet middle
            if self.pipeline.controlnet is not None:
                batch = rearrange(batch, "b c t h w-> (b t) h w c")
                controlnet_processor_params = update_controlnet_processor_params(
                    src=self.controlnet_processor_params,
                    dst=controlnet_processor_params,
                )
                if not video_is_middle:
                    batch_condition = self.controlnet_processor(
                        data=batch,
                        data_channel_order="b h w c",
                        target_height=height,
                        target_width=width,
                        return_type="np",
                        return_data_channel_order="b c h w",
                        input_rgb_order="rgb",
                        processor_params=controlnet_processor_params,
                    )
                else:
                    # TODO: 临时用于可视化输入的 controlnet middle 序列，后续待拆到 middl2video中，也可以增加参数支持
                    # TODO: only use video_path is controlnet middle output, to improved
                    batch_condition = rearrange(
                        copy.deepcopy(batch), " b h w c-> b c h w"
                    )

                # 当前仅当 输入是 middle、condition_image的pose在middle首帧之前，需要重新生成condition_images的pose并绑定到middle_batch上
                # when video_path is middle seq and condition_image is not aligned with middle seq,
                # regenerate codntion_images pose, and then concat into middle_batch,
                if (
                    i_batch == 0
                    and not video_has_condition
                    and video_is_middle
                    and condition_images is not None
                ):
                    condition_images_reshape = rearrange(
                        condition_images, "b c t h w-> (b t) h w c"
                    )
                    condition_images_condition = self.controlnet_processor(
                        data=condition_images_reshape,
                        data_channel_order="b h w c",
                        target_height=height,
                        target_width=width,
                        return_type="np",
                        return_data_channel_order="b c h w",
                        input_rgb_order="rgb",
                        processor_params=controlnet_processor_params,
                    )
                    condition_images_condition = rearrange(
                        condition_images_condition,
                        "(b t) c h w-> b c t h w",
                        b=batch_size,
                    )
                else:
                    condition_images_condition = None
                if not isinstance(batch_condition, list):
                    batch_condition = rearrange(
                        batch_condition, "(b t) c h w-> b c t h w", b=batch_size
                    )
                    if condition_images_condition is not None:
                        batch_condition = np.concatenate(
                            [
                                condition_images_condition,
                                batch_condition,
                            ],
                            axis=2,
                        )
                        # 此时 batch_condition 比 batch 多了一帧，为了最终视频能 concat 存储，替换下
                        # 当前仅适用于  condition_images_condition 不为None
                        # when condition_images_condition is not None,  batch_condition has more frames than batch
                        batch = rearrange(batch_condition, "b c t h w ->(b t) h w c")
                else:
                    batch_condition = [
                        rearrange(x, "(b t) c h w-> b c t h w", b=batch_size)
                        for x in batch_condition
                    ]
                    if condition_images_condition is not None:
                        batch_condition = [
                            np.concatenate(
                                [condition_images_condition, batch_condition_tmp],
                                axis=2,
                            )
                            for batch_condition_tmp in batch_condition
                        ]
                batch = rearrange(batch, "(b t) h w c -> b c t h w", b=batch_size)
            else:
                batch_condition = None
            # condition [0,255]
            # latent: [0,1]
            # 按需求生成多个片段，
            # generate multi video_shot
            # 第一个片段 会特殊处理，需要生成首帧
            # first shot is special because of first frame.
            # 后续片段根据拿前一个片段结果，首尾相连的方式生成。
            # use last frame of last shot as the first frame of the current shot
            # TODO: 当前独立拆开实现，待后续融合到一起实现
            # TODO: to optimize  implementation way
            if n_vision_condition == 0:
                actual_video_length = video_length
                control_image = batch_condition
                first_image_controlnet_condition = None
                first_image_latents = None
                if need_video2video:
                    video = batch
                else:
                    video = None
                result_overlap = 0
            else:
                if i_batch == 0:
                    if self.pipeline.controlnet is not None:
                        if not isinstance(batch_condition, list):
                            first_image_controlnet_condition = batch_condition[
                                :, :, :1, :, :
                            ]
                        else:
                            first_image_controlnet_condition = [
                                x[:, :, :1, :, :] for x in batch_condition
                            ]
                    else:
                        first_image_controlnet_condition = None
                    if need_video2video:
                        if condition_images is None:
                            video = batch[:, :, :1, :, :]
                        else:
                            video = condition_images
                    else:
                        video = None
                    if condition_images is not None and not redraw_condition_image:
                        first_image = condition_images
                        first_image_latents = None
                    else:
                        (
                            first_image,
                            first_image_latents,
                            _,
                            _,
                            _,
                        ) = self.pipeline(
                            prompt=prompt,
                            image=video,
                            control_image=first_image_controlnet_condition,
                            num_inference_steps=num_inference_steps,
                            video_length=1,
                            height=height,
                            width=width,
                            return_dict=False,
                            skip_temporal_layer=True,
                            output_type="np",
                            generator=generator,
                            negative_prompt=negative_prompt,
                            controlnet_conditioning_scale=controlnet_conditioning_scale,
                            control_guidance_start=control_guidance_start,
                            control_guidance_end=control_guidance_end,
                            w_ind_noise=w_ind_noise,
                            strength=strength,
                            refer_image=refer_image
                            if redraw_condition_image_with_referencenet
                            else None,
                            ip_adapter_image=ip_adapter_image
                            if redraw_condition_image_with_ipdapter
                            else None,
                            refer_face_image=refer_face_image
                            if redraw_condition_image_with_facein
                            else None,
                            ip_adapter_scale=ip_adapter_scale,
                            facein_scale=facein_scale,
                            ip_adapter_face_scale=ip_adapter_face_scale,
                            ip_adapter_face_image=refer_face_image
                            if redraw_condition_image_with_ip_adapter_face
                            else None,
                            prompt_only_use_image_prompt=prompt_only_use_image_prompt,
                        )
                        if refer_image is not None:
                            refer_image = first_image * 255.0
                        if ip_adapter_image is not None:
                            ip_adapter_image = first_image * 255.0
                        # 首帧用于后续推断可以直接用first_image_latent不需要 first_image了
                        first_image = None
                    if self.pipeline.controlnet is not None:
                        if not isinstance(batch_condition, list):
                            control_image = batch_condition[:, :, 1:, :, :]
                            logger.debug(f"control_image={control_image.shape}")
                        else:
                            control_image = [x[:, :, 1:, :, :] for x in batch_condition]
                    else:
                        control_image = None

                    actual_video_length = time_size - int(video_has_condition)
                    if need_video2video:
                        video = batch[:, :, 1:, :, :]
                    else:
                        video = None

                    result_overlap = 0
                else:
                    actual_video_length = time_size
                    if self.pipeline.controlnet is not None:
                        if not fix_condition_images:
                            logger.debug(
                                f"{i_batch}, update first_image_controlnet_condition"
                            )

                            if not isinstance(last_batch_condition, list):
                                first_image_controlnet_condition = last_batch_condition[
                                    :, :, -1:, :, :
                                ]
                            else:
                                first_image_controlnet_condition = [
                                    x[:, :, -1:, :, :] for x in last_batch_condition
                                ]
                        else:
                            logger.debug(
                                f"{i_batch}, do not update first_image_controlnet_condition"
                            )
                        control_image = batch_condition
                    else:
                        control_image = None
                        first_image_controlnet_condition = None
                    if not fix_condition_images:
                        logger.debug(f"{i_batch}, update condition_images")
                        first_image_latents = out_latents_batch[:, :, -1:, :, :]
                    else:
                        logger.debug(f"{i_batch}, do not update condition_images")

                    if need_video2video:
                        video = batch
                    else:
                        video = None
                    result_overlap = 1

                    # 更新 ref_image和 ipadapter_image
                    if not fixed_refer_image:
                        logger.debug(
                            "ref_image use last frame of last generated out video"
                        )
                        refer_image = (
                            out_batch[:, :, -n_vision_condition:, :, :] * 255.0
                        )
                    else:
                        logger.debug("use given fixed ref_image")

                    if not fixed_ip_adapter_image:
                        logger.debug(
                            "ip_adapter_image use last frame of last generated out video"
                        )
                        ip_adapter_image = (
                            out_batch[:, :, -n_vision_condition:, :, :] * 255.0
                        )
                    else:
                        logger.debug("use given fixed ip_adapter_image")

                    # face image
                    if not fixed_ip_adapter_image:
                        logger.debug(
                            "refer_face_image use last frame of last generated out video"
                        )
                        refer_face_image = (
                            out_batch[:, :, -n_vision_condition:, :, :] * 255.0
                        )
                    else:
                        logger.debug("use given fixed ip_adapter_image")

            out = self.pipeline(
                video_length=actual_video_length,  # int
                prompt=prompt,
                num_inference_steps=video_num_inference_steps,
                height=height,
                width=width,
                generator=generator,
                image=video,
                control_image=control_image,  # b ci(3) t hi wi
                controlnet_condition_images=first_image_controlnet_condition,  # b ci(3) t(1) hi wi
                # controlnet_condition_images=np.zeros_like(
                #     first_image_controlnet_condition
                # ),  # b ci(3) t(1) hi wi
                condition_images=first_image,
                condition_latents=first_image_latents,  # b co t(1) ho wo
                skip_temporal_layer=False,
                output_type="np",
                noise_type=noise_type,
                negative_prompt=video_negative_prompt,
                need_img_based_video_noise=need_img_based_video_noise,
                controlnet_conditioning_scale=controlnet_conditioning_scale,
                control_guidance_start=control_guidance_start,
                control_guidance_end=control_guidance_end,
                w_ind_noise=w_ind_noise,
                img_weight=img_weight,
                motion_speed=video_reader.sample_rate,
                guidance_scale=video_guidance_scale,
                guidance_scale_end=video_guidance_scale_end,
                guidance_scale_method=video_guidance_scale_method,
                strength=video_strength,
                refer_image=refer_image,
                ip_adapter_image=ip_adapter_image,
                refer_face_image=refer_face_image,
                ip_adapter_scale=ip_adapter_scale,
                facein_scale=facein_scale,
                ip_adapter_face_scale=ip_adapter_face_scale,
                ip_adapter_face_image=refer_face_image,
                prompt_only_use_image_prompt=prompt_only_use_image_prompt,
                initial_common_latent=initial_common_latent,
                # serial_denoise parameter start
                record_mid_video_noises=record_mid_video_noises,
                last_mid_video_noises=last_mid_video_noises,
                record_mid_video_latents=record_mid_video_latents,
                last_mid_video_latents=last_mid_video_latents,
                video_overlap=video_overlap,
                # serial_denoise parameter end
                # parallel_denoise parameter start
                context_schedule=context_schedule,
                context_frames=context_frames,
                context_stride=context_stride,
                context_overlap=context_overlap,
                context_batch_size=context_batch_size,
                interpolation_factor=interpolation_factor,
                # parallel_denoise parameter end
            )
            last_batch = batch
            last_batch_condition = batch_condition
            last_mid_video_latents = out.mid_video_latents
            last_mid_video_noises = out.mid_video_noises
            out_batch = out.videos[:, :, result_overlap:, :, :]
            out_latents_batch = out.latents[:, :, result_overlap:, :, :]
            out_videos.append(out_batch)
            if need_return_videos:
                videos.append(batch)
            if out_condition is not None:
                out_condition.append(batch_condition)

        out_videos = np.concatenate(out_videos, axis=2)
        if need_return_videos:
            videos = np.concatenate(videos, axis=2)
        if out_condition is not None:
            if not isinstance(out_condition[0], list):
                out_condition = np.concatenate(out_condition, axis=2)
            else:
                out_condition = [
                    [out_condition[j][i] for j in range(len(out_condition))]
                    for i in range(len(out_condition[0]))
                ]
                out_condition = [np.concatenate(x, axis=2) for x in out_condition]
        if need_hist_match:
            videos[:, :, 1:, :, :] = hist_match_video_bcthw(
                videos[:, :, 1:, :, :], videos[:, :, :1, :, :], value=255.0
            )
        return out_videos, out_condition, videos