




import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class        UpperCAmelCase__				(							UpperCAmelCase__      ,		UpperCAmelCase__      ,		UpperCAmelCase__      ,		unittest.TestCase				):



     '''simple docstring'''
     _SCREAMING_SNAKE_CASE      :					List[str]     					=    StableDiffusionImageVariationPipeline
     _SCREAMING_SNAKE_CASE      :					Optional[int]     					=    IMAGE_VARIATION_PARAMS
     _SCREAMING_SNAKE_CASE      :					str     					=    IMAGE_VARIATION_BATCH_PARAMS
     _SCREAMING_SNAKE_CASE      :					int     					=    frozenset([]				)
     # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     _SCREAMING_SNAKE_CASE      :					List[str]     					=    frozenset([]				)
     def        lowerCAmelCase__  (				self	):
            torch.manual_seed(0	)
            a			   =UNetaDConditionModel(
                block_out_channels=(32, 64) ,    layers_per_block=2 ,    sample_size=32 ,    in_channels=4 ,    out_channels=4 ,    down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""") ,    up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""") ,    cross_attention_dim=32 ,    )
            a			   =PNDMScheduler(skip_prk_steps=_lowerCAmelCase	)
            torch.manual_seed(0	)
            a			   =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 ,    )
            torch.manual_seed(0	)
            a			   =CLIPVisionConfig(
                hidden_size=32 ,    projection_dim=32 ,    intermediate_size=37 ,    layer_norm_eps=1E-05 ,    num_attention_heads=4 ,    num_hidden_layers=5 ,    image_size=32 ,    patch_size=4 ,    )
            a			   =CLIPVisionModelWithProjection(_lowerCAmelCase	)
            a			   =CLIPImageProcessor(crop_size=32 ,    size=32	)

            a			   ={
                """unet""": unet,
                """scheduler""": scheduler,
                """vae""": vae,
                """image_encoder""": image_encoder,
                """feature_extractor""": feature_extractor,
                """safety_checker""": None,
            }
            return components
     def        lowerCAmelCase__  (				self ,    _lowerCAmelCase ,    _lowerCAmelCase=0	):
            a			   =floats_tensor((1, 3, 32, 32) ,    rng=random.Random(_lowerCAmelCase	)	)
            a			   =image.cpu().permute(0 ,    2 ,    3 ,    1	)[0]
            a			   =Image.fromarray(np.uinta(_lowerCAmelCase	)	).convert("""RGB"""	).resize((32, 32)	)
            if str(_lowerCAmelCase	).startswith("""mps"""	):
                   a			   =torch.manual_seed(_lowerCAmelCase	)
            else:
                   a			   =torch.Generator(device=_lowerCAmelCase	).manual_seed(_lowerCAmelCase	)
            a			   ={
                """image""": image,
                """generator""": generator,
                """num_inference_steps""": 2,
                """guidance_scale""": 6.0,
                """output_type""": """numpy""",
            }
            return inputs
     def        lowerCAmelCase__  (				self	):
            a			   ="""cpu"""  # ensure determinism for the device-dependent torch.Generator
            a			   =self.get_dummy_components()
            a			   =StableDiffusionImageVariationPipeline(**_lowerCAmelCase	)
            a			   =sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

            a			   =self.get_dummy_inputs(_lowerCAmelCase	)
            a			   =sd_pipe(**_lowerCAmelCase	).images
            a			   =image[0, -3:, -3:, -1]

            assert image.shape == (1, 64, 64, 3)
            a			   =np.array([0.52_39, 0.57_23, 0.47_96, 0.50_49, 0.55_50, 0.46_85, 0.53_29, 0.48_91, 0.49_21]	)

            assert np.abs(image_slice.flatten() - expected_slice	).max() < 1E-3
     def        lowerCAmelCase__  (				self	):
            a			   ="""cpu"""  # ensure determinism for the device-dependent torch.Generator
            a			   =self.get_dummy_components()
            a			   =StableDiffusionImageVariationPipeline(**_lowerCAmelCase	)
            a			   =sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

            a			   =self.get_dummy_inputs(_lowerCAmelCase	)
            a			   =2 * [inputs["""image"""]]
            a			   =sd_pipe(**_lowerCAmelCase	)

            a			   =output.images

            a			   =image[-1, -3:, -3:, -1]

            assert image.shape == (2, 64, 64, 3)
            a			   =np.array([0.68_92, 0.56_37, 0.58_36, 0.57_71, 0.62_54, 0.64_09, 0.55_80, 0.55_69, 0.52_89]	)

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






     def        lowerCAmelCase__  (				self	):
            super().test_inference_batch_single_identical(expected_max_diff=3E-3	)



@slow
@require_torch_gpu
class        UpperCAmelCase__				(							unittest.TestCase				):



     '''simple docstring'''
     def        lowerCAmelCase__  (				self	):
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()
     def        lowerCAmelCase__  (				self ,    _lowerCAmelCase ,    _lowerCAmelCase="cpu" ,    _lowerCAmelCase=torch.floataa ,    _lowerCAmelCase=0	):
            a			   =torch.Generator(device=_lowerCAmelCase	).manual_seed(_lowerCAmelCase	)
            a			   =load_image(
                """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
                """/stable_diffusion_imgvar/input_image_vermeer.png"""	)
            a			   =np.random.RandomState(_lowerCAmelCase	).standard_normal((1, 4, 64, 64)	)
            a			   =torch.from_numpy(_lowerCAmelCase	).to(device=_lowerCAmelCase ,    dtype=_lowerCAmelCase	)
            a			   ={
                """image""": init_image,
                """latents""": latents,
                """generator""": generator,
                """num_inference_steps""": 3,
                """guidance_scale""": 7.5,
                """output_type""": """numpy""",
            }
            return inputs
     def        lowerCAmelCase__  (				self	):
            a			   =StableDiffusionImageVariationPipeline.from_pretrained(
                """lambdalabs/sd-image-variations-diffusers""" ,    safety_checker=_lowerCAmelCase	)
            a			   =sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

            a			   =self.get_inputs(_lowerCAmelCase	)
            a			   =sd_pipe(**_lowerCAmelCase	).images
            a			   =image[0, -3:, -3:, -1].flatten()

            assert image.shape == (1, 512, 512, 3)
            a			   =np.array([0.8_44_91, 0.9_07_89, 0.7_57_08, 0.7_87_34, 0.8_34_85, 0.7_00_99, 0.6_69_38, 0.6_87_27, 0.6_13_79]	)
            assert np.abs(image_slice - expected_slice	).max() < 6E-3
     def        lowerCAmelCase__  (				self	):
            a			   =0

            def callback_fn(_lowerCAmelCase ,    _lowerCAmelCase ,    _lowerCAmelCase	) -> None:
                   a			   =True
                   nonlocal number_of_steps
                   number_of_steps += 1
                   if step == 1:
                          a			   =latents.detach().cpu().numpy()
                          assert latents.shape == (1, 4, 64, 64)
                          a			   =latents[0, -3:, -3:, -1]
                          a			   =np.array(
                              [-0.16_21, 0.28_37, -0.79_79, -0.12_21, -1.30_57, 0.76_81, -2.11_91, 0.04_64, 1.63_09]	)

                          assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5E-2
                   elif step == 2:
                          a			   =latents.detach().cpu().numpy()
                          assert latents.shape == (1, 4, 64, 64)
                          a			   =latents[0, -3:, -3:, -1]
                          a			   =np.array([0.62_99, 1.75_00, 1.19_92, -2.15_82, -1.89_94, 0.73_34, -0.70_90, 1.01_37, 1.52_73]	)

                          assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5E-2

            a			   =False

            a			   =StableDiffusionImageVariationPipeline.from_pretrained(
                """fusing/sd-image-variations-diffusers""" ,    safety_checker=_lowerCAmelCase ,    torch_dtype=torch.floataa ,    )
            pipe.to(_lowerCAmelCase	)
            pipe.set_progress_bar_config(disable=_lowerCAmelCase	)
            pipe.enable_attention_slicing()

            a			   =self.get_inputs(_lowerCAmelCase ,    dtype=torch.floataa	)
            pipe(**_lowerCAmelCase ,    callback=_lowerCAmelCase ,    callback_steps=1	)
            assert callback_fn.has_been_called
            assert number_of_steps == inputs["num_inference_steps"]






     def        lowerCAmelCase__  (				self	):
            torch.cuda.empty_cache()
            torch.cuda.reset_max_memory_allocated()
            torch.cuda.reset_peak_memory_stats()

            a			   ="""fusing/sd-image-variations-diffusers"""
            a			   =StableDiffusionImageVariationPipeline.from_pretrained(
                _lowerCAmelCase ,    safety_checker=_lowerCAmelCase ,    torch_dtype=torch.floataa	)
            a			   =pipe.to(_lowerCAmelCase	)
            pipe.set_progress_bar_config(disable=_lowerCAmelCase	)
            pipe.enable_attention_slicing(1	)
            pipe.enable_sequential_cpu_offload()

            a			   =self.get_inputs(_lowerCAmelCase ,    dtype=torch.floataa	)
            a			   =pipe(**_lowerCAmelCase	)

            a			   =torch.cuda.max_memory_allocated()
            # make sure that less than 2.6 GB is allocated
            assert mem_bytes < 2.6 * 10**9



@nightly
@require_torch_gpu
class        UpperCAmelCase__				(							unittest.TestCase				):



     '''simple docstring'''
     def        lowerCAmelCase__  (				self	):
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()
     def        lowerCAmelCase__  (				self ,    _lowerCAmelCase ,    _lowerCAmelCase="cpu" ,    _lowerCAmelCase=torch.floataa ,    _lowerCAmelCase=0	):
            a			   =torch.Generator(device=_lowerCAmelCase	).manual_seed(_lowerCAmelCase	)
            a			   =load_image(
                """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
                """/stable_diffusion_imgvar/input_image_vermeer.png"""	)
            a			   =np.random.RandomState(_lowerCAmelCase	).standard_normal((1, 4, 64, 64)	)
            a			   =torch.from_numpy(_lowerCAmelCase	).to(device=_lowerCAmelCase ,    dtype=_lowerCAmelCase	)
            a			   ={
                """image""": init_image,
                """latents""": latents,
                """generator""": generator,
                """num_inference_steps""": 50,
                """guidance_scale""": 7.5,
                """output_type""": """numpy""",
            }
            return inputs
     def        lowerCAmelCase__  (				self	):
            a			   =StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""	)
            sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

            a			   =self.get_inputs(_lowerCAmelCase	)
            a			   =sd_pipe(**_lowerCAmelCase	).images[0]

            a			   =load_numpy(
                """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
                """/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"""	)
            a			   =np.abs(expected_image - image	).max()
            assert max_diff < 1E-3






     def        lowerCAmelCase__  (				self	):
            a			   =StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""	)
            a			   =DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config	)
            sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

            a			   =self.get_inputs(_lowerCAmelCase	)
            a			   =25
            a			   =sd_pipe(**_lowerCAmelCase	).images[0]

            a			   =load_numpy(
                """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
                """/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"""	)
            a			   =np.abs(expected_image - image	).max()
            assert max_diff < 1E-3






import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCamelCase							=      logging.get_logger(__name__)

_lowerCamelCase							=      {
    '''facebook/hubert-base-ls960''': '''https://huggingface.co/facebook/hubert-base-ls960/resolve/main/config.json''',
    # See all Hubert models at https://huggingface.co/models?filter=hubert
}


class        UpperCAmelCase__				(							UpperCAmelCase__				):



     '''simple docstring'''
     _SCREAMING_SNAKE_CASE      :					Dict     					=    "hubert"
     def __init__(				self ,    _lowerCAmelCase=32 ,    _lowerCAmelCase=768 ,    _lowerCAmelCase=12 ,    _lowerCAmelCase=12 ,    _lowerCAmelCase=3_072 ,    _lowerCAmelCase="gelu" ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=True ,    _lowerCAmelCase=0.0 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.02 ,    _lowerCAmelCase=1E-5 ,    _lowerCAmelCase="group" ,    _lowerCAmelCase="gelu" ,    _lowerCAmelCase=(512, 512, 512, 512, 512, 512, 512) ,    _lowerCAmelCase=(5, 2, 2, 2, 2, 2, 2) ,    _lowerCAmelCase=(10, 3, 3, 3, 3, 2, 2) ,    _lowerCAmelCase=False ,    _lowerCAmelCase=128 ,    _lowerCAmelCase=16 ,    _lowerCAmelCase=False ,    _lowerCAmelCase=True ,    _lowerCAmelCase=0.05 ,    _lowerCAmelCase=10 ,    _lowerCAmelCase=2 ,    _lowerCAmelCase=0.0 ,    _lowerCAmelCase=10 ,    _lowerCAmelCase=0 ,    _lowerCAmelCase="sum" ,    _lowerCAmelCase=False ,    _lowerCAmelCase=False ,    _lowerCAmelCase=256 ,    _lowerCAmelCase=0 ,    _lowerCAmelCase=1 ,    _lowerCAmelCase=2 ,    **_lowerCAmelCase ,    ):
            super().__init__(**_lowerCAmelCase ,    pad_token_id=_lowerCAmelCase ,    bos_token_id=_lowerCAmelCase ,    eos_token_id=_lowerCAmelCase	)
            a			   =hidden_size
            a			   =feat_extract_norm
            a			   =feat_extract_activation
            a			   =list(_lowerCAmelCase	)
            a			   =list(_lowerCAmelCase	)
            a			   =list(_lowerCAmelCase	)
            a			   =conv_bias
            a			   =num_conv_pos_embeddings
            a			   =num_conv_pos_embedding_groups
            a			   =len(self.conv_dim	)
            a			   =num_hidden_layers
            a			   =intermediate_size
            a			   =hidden_act
            a			   =num_attention_heads
            a			   =hidden_dropout
            a			   =attention_dropout
            a			   =activation_dropout
            a			   =feat_proj_layer_norm
            a			   =feat_proj_dropout
            a			   =final_dropout
            a			   =layerdrop
            a			   =layer_norm_eps
            a			   =initializer_range
            a			   =vocab_size
            a			   =do_stable_layer_norm
            a			   =use_weighted_layer_sum
            a			   =classifier_proj_size

            if (
                (len(self.conv_stride	) != self.num_feat_extract_layers)
                or (len(self.conv_kernel	) != self.num_feat_extract_layers)
                or (len(self.conv_dim	) != self.num_feat_extract_layers)
            ):
                   raise ValueError(
                       """Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =="""
                       """ `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ="""
                       F''' {len(self.conv_dim	)}`, `len(config.conv_stride) = {len(self.conv_stride	)}`,'''
                       F''' `len(config.conv_kernel) = {len(self.conv_kernel	)}`.'''	)

            # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
            a			   =apply_spec_augment
            a			   =mask_time_prob
            a			   =mask_time_length
            a			   =mask_time_min_masks
            a			   =mask_feature_prob
            a			   =mask_feature_length
            a			   =mask_feature_min_masks

            # ctc loss
            a			   =ctc_loss_reduction
            a			   =ctc_zero_infinity






     @property
     def        lowerCAmelCase__  (				self	):
            return functools.reduce(operator.mul ,    self.conv_stride ,    1	)
