



import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()







@skip_mps
class 		snake_case						(	SCREAMING_SNAKE_CASE_					,unittest.TestCase  ):



   '''simple docstring'''






   snake_case_					: List[str]          =		VideoToVideoSDPipeline
   snake_case_					: Optional[Any]          =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}  ) - {"""image""", """width""", """height"""}
   snake_case_					: str          =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}  ) - {"""image"""}
   snake_case_					: List[str]          =		PipelineTesterMixin.required_optional_params - {"""latents"""}
   snake_case_					: Union[str, Any]          =		False

   # No `output_type`.
   snake_case_					: List[Any]          =		frozenset(
       [
           """num_inference_steps""",
           """generator""",
           """latents""",
           """return_dict""",
           """callback""",
           """callback_steps""",
       ]  )

   def      UpperCamelCase_  (					self							:			Union[str, Any])				->				List[Any]:

    """simple docstring"""







    torch.manual_seed(0)
    _snake_case    :  Optional[int]					=					UNetaDConditionModel(
        block_out_channels=(32, 64, 64, 64)				,		layers_per_block=2				,		sample_size=32				,		in_channels=4				,		out_channels=4				,		down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""")				,		up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""")				,		cross_attention_dim=32				,		attention_head_dim=4				,		)
    _snake_case    :  Union[str, Any]					=					DDIMScheduler(
        beta_start=0.00_085				,		beta_end=0.012				,		beta_schedule="""scaled_linear"""				,		clip_sample=lowerCAmelCase				,		set_alpha_to_one=lowerCAmelCase				,		)
    torch.manual_seed(0)
    _snake_case    :  int					=					AutoencoderKL(
        block_out_channels=[32, 64]				,		in_channels=3				,		out_channels=3				,		down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]				,		up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]				,		latent_channels=4				,		sample_size=128				,		)
    torch.manual_seed(0)
    _snake_case    :  int					=					CLIPTextConfig(
        bos_token_id=0				,		eos_token_id=2				,		hidden_size=32				,		intermediate_size=37				,		layer_norm_eps=1E-05				,		num_attention_heads=4				,		num_hidden_layers=5				,		pad_token_id=1				,		vocab_size=1000				,		hidden_act="""gelu"""				,		projection_dim=512				,		)
    _snake_case    :  int					=					CLIPTextModel(lowerCAmelCase)
    _snake_case    :  Optional[Any]					=					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

    _snake_case    :  Optional[int]					=					{
        """unet""": unet,
        """scheduler""": scheduler,
        """vae""": vae,
        """text_encoder""": text_encoder,
        """tokenizer""": tokenizer,
    }
    return components

   def      UpperCamelCase_  (					self							:			Union[str, Any]				,		lowerCAmelCase							:			Optional[int]				,		lowerCAmelCase							:			str=0)				->				Dict:

    """simple docstring"""







    _snake_case    :  Optional[Any]					=					floats_tensor((1, 3, 3, 32, 32)				,		rng=random.Random(lowerCAmelCase)).to(lowerCAmelCase)

    if str(lowerCAmelCase).startswith("""mps"""):
     _snake_case    :  int					=					torch.manual_seed(lowerCAmelCase)
    else:
     _snake_case    :  Optional[Any]					=					torch.Generator(device=lowerCAmelCase).manual_seed(lowerCAmelCase)
    _snake_case    :  Optional[int]					=					{
        """prompt""": """A painting of a squirrel eating a burger""",
        """video""": video,
        """generator""": generator,
        """num_inference_steps""": 2,
        """guidance_scale""": 6.0,
        """output_type""": """pt""",
    }
    return inputs

   def      UpperCamelCase_  (					self							:			Any)				->				List[str]:

    """simple docstring"""







    _snake_case    :  List[Any]					=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
    _snake_case    :  List[Any]					=					self.get_dummy_components()
    _snake_case    :  Dict					=					VideoToVideoSDPipeline(**lowerCAmelCase)
    _snake_case    :  Dict					=					sd_pipe.to(lowerCAmelCase)
    sd_pipe.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case    :  Optional[Any]					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  Any					=					"""np"""
    _snake_case    :  Tuple					=					sd_pipe(**lowerCAmelCase).frames
    _snake_case    :  Union[str, Any]					=					frames[0][-3:, -3:, -1]

    assert frames[0].shape == (32, 32, 3)
    _snake_case    :  Dict					=					np.array([106, 117, 113, 174, 137, 112, 148, 151, 131])

    assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2

   @unittest.skipIf(
       torch_device != """cuda""" or not is_xformers_available()				,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""				,		)
   def      UpperCamelCase_  (					self							:			List[Any])				->				str:

    """simple docstring"""







    self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCAmelCase				,		expected_max_diff=5E-3)

   @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline.""")
   def      UpperCamelCase_  (					self							:			List[Any])				->				List[Any]:

    """simple docstring"""







    pass

   @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline.""")
   def      UpperCamelCase_  (					self							:			int)				->				Optional[int]:

    """simple docstring"""







    pass

   @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline.""")
   def      UpperCamelCase_  (					self							:			Dict)				->				List[Any]:

    """simple docstring"""







    pass


   def      UpperCamelCase_  (					self							:			Union[str, Any])				->				Any:

    """simple docstring"""







    return super().test_progress_bar()







@slow
@skip_mps
class 		snake_case						(	unittest.TestCase  ):



   '''simple docstring'''






   def      UpperCamelCase_  (					self							:			Tuple)				->				List[str]:

    """simple docstring"""







    _snake_case    :  List[str]					=					VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""				,		torch_dtype=torch.floataa)
    pipe.enable_model_cpu_offload()

    # 10 frames
    _snake_case    :  Dict					=					torch.Generator(device="""cpu""").manual_seed(0)
    _snake_case    :  Union[str, Any]					=					torch.randn((1, 10, 3, 1024, 576)				,		generator=lowerCAmelCase)
    _snake_case    :  Tuple					=					video.to("""cuda""")

    _snake_case    :  Optional[Any]					=					"""Spiderman is surfing"""

    _snake_case    :  Dict					=					pipe(lowerCAmelCase				,		video=lowerCAmelCase				,		generator=lowerCAmelCase				,		num_inference_steps=3				,		output_type="""pt""").frames

    _snake_case    :  int					=					np.array([-1.0_458_984, -1.1_279_297, -0.9_663_086, -0.91_503_906, -0.75_097_656])
    assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array).sum() < 1E-2





from .constants import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SAFE_WEIGHTS_INDEX_NAME,
    SAFE_WEIGHTS_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    TORCH_LAUNCH_PARAMS,
    WEIGHTS_INDEX_NAME,
    WEIGHTS_NAME,
)
from .dataclasses import (
    BnbQuantizationConfig,
    ComputeEnvironment,
    CustomDtype,
    DeepSpeedPlugin,
    DistributedDataParallelKwargs,
    DistributedType,
    DynamoBackend,
    FPaRecipeKwargs,
    FullyShardedDataParallelPlugin,
    GradientAccumulationPlugin,
    GradScalerKwargs,
    InitProcessGroupKwargs,
    KwargsHandler,
    LoggerType,
    MegatronLMPlugin,
    PrecisionType,
    ProjectConfiguration,
    RNGType,
    SageMakerDistributedType,
    TensorInformation,
    TorchDynamoPlugin,
)
from .environment import get_int_from_env, parse_choice_from_env, parse_flag_from_env
from .imports import (
    get_ccl_version,
    is_abit_bnb_available,
    is_abit_bnb_available,
    is_aim_available,
    is_bfaa_available,
    is_bnb_available,
    is_botoa_available,
    is_ccl_available,
    is_comet_ml_available,
    is_datasets_available,
    is_deepspeed_available,
    is_fpa_available,
    is_ipex_available,
    is_megatron_lm_available,
    is_mlflow_available,
    is_mps_available,
    is_npu_available,
    is_rich_available,
    is_safetensors_available,
    is_sagemaker_available,
    is_tensorboard_available,
    is_tpu_available,
    is_transformers_available,
    is_wandb_available,
    is_xpu_available,
)
from .modeling import (
    check_device_map,
    check_tied_parameters_in_config,
    check_tied_parameters_on_same_device,
    compute_module_sizes,
    convert_file_size_to_int,
    dtype_byte_size,
    find_tied_parameters,
    get_balanced_memory,
    get_max_layer_size,
    get_max_memory,
    get_mixed_precision_context_manager,
    id_tensor_storage,
    infer_auto_device_map,
    load_checkpoint_in_model,
    load_offloaded_weights,
    load_state_dict,
    named_module_tensors,
    retie_parameters,
    set_module_tensor_to_device,
    shard_checkpoint,
)
from .offload import (
    OffloadedWeightsLoader,
    PrefixedDataset,
    extract_submodules_state_dict,
    load_offloaded_weight,
    offload_state_dict,
    offload_weight,
    save_offload_index,
)
from .operations import (
    broadcast,
    broadcast_object_list,
    concatenate,
    convert_outputs_to_fpaa,
    convert_to_fpaa,
    find_batch_size,
    find_device,
    gather,
    gather_object,
    get_data_structure,
    honor_type,
    initialize_tensors,
    is_namedtuple,
    is_tensor_information,
    is_torch_tensor,
    listify,
    pad_across_processes,
    recursively_apply,
    reduce,
    send_to_device,
    slice_tensors,
)
from .versions import compare_versions, is_torch_version


if is_deepspeed_available():
  from .deepspeed import (
      DeepSpeedEngineWrapper,
      DeepSpeedOptimizerWrapper,
      DeepSpeedSchedulerWrapper,
      DummyOptim,
      DummyScheduler,
      HfDeepSpeedConfig,
  )

from .bnb import has_abit_bnb_layers, load_and_quantize_model
from .fsdp_utils import load_fsdp_model, load_fsdp_optimizer, save_fsdp_model, save_fsdp_optimizer
from .launch import (
    PrepareForLaunch,
    _filter_args,
    prepare_deepspeed_cmd_env,
    prepare_multi_gpu_env,
    prepare_sagemager_args_inputs,
    prepare_simple_launcher_cmd_env,
    prepare_tpu,
)
from .megatron_lm import (
    AbstractTrainStep,
    BertTrainStep,
    GPTTrainStep,
    MegatronEngine,
    MegatronLMDummyDataLoader,
    MegatronLMDummyScheduler,
    MegatronLMOptimizerWrapper,
    MegatronLMSchedulerWrapper,
    TaTrainStep,
    avg_losses_across_data_parallel_group,
    gather_across_data_parallel_groups,
)
from .megatron_lm import initialize as megatron_lm_initialize
from .megatron_lm import prepare_data_loader as megatron_lm_prepare_data_loader
from .megatron_lm import prepare_model as megatron_lm_prepare_model
from .megatron_lm import prepare_optimizer as megatron_lm_prepare_optimizer
from .megatron_lm import prepare_scheduler as megatron_lm_prepare_scheduler
from .memory import find_executable_batch_size, release_memory
from .other import (
    extract_model_from_parallel,
    get_pretty_name,
    is_port_in_use,
    merge_dicts,
    patch_environment,
    save,
    wait_for_everyone,
    write_basic_config,
)
from .random import set_seed, synchronize_rng_state, synchronize_rng_states
from .torch_xla import install_xla
from .tqdm import tqdm
from .transformer_engine import convert_model, has_transformer_engine_layers
