
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







@skip_mps
class 					lowerCamelCase		(  A_							,			A_							,			unittest.TestCase  ):
     UpperCAmelCase__    :							List[Any]									=		StableDiffusionPanoramaPipeline
     UpperCAmelCase__    :							List[Any]									=		TEXT_TO_IMAGE_PARAMS
     UpperCAmelCase__    :							str									=		TEXT_TO_IMAGE_BATCH_PARAMS
     UpperCAmelCase__    :							Dict									=		TEXT_TO_IMAGE_IMAGE_PARAMS
     UpperCAmelCase__    :							List[str]									=		TEXT_TO_IMAGE_IMAGE_PARAMS





     def 		UpperCAmelCase(self				:   List[str]       )						->						Union[str, Any]:
         torch.manual_seed(0       )
         snake_case								= UNetaDConditionModel(
             block_out_channels=(3_2, 6_4)    ,   layers_per_block=1    ,   sample_size=3_2    ,   in_channels=4    ,   out_channels=4    ,   down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")    ,   up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")    ,   cross_attention_dim=3_2    ,   )
         snake_case								= DDIMScheduler()
         torch.manual_seed(0       )
         snake_case								= AutoencoderKL(
             block_out_channels=[3_2, 6_4]    ,   in_channels=3    ,   out_channels=3    ,   down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]    ,   up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]    ,   latent_channels=4    ,   )
         torch.manual_seed(0       )
         snake_case								= CLIPTextConfig(
             bos_token_id=0    ,   eos_token_id=2    ,   hidden_size=3_2    ,   intermediate_size=3_7    ,   layer_norm_eps=1E-05    ,   num_attention_heads=4    ,   num_hidden_layers=5    ,   pad_token_id=1    ,   vocab_size=1_0_0_0    ,   )
         snake_case								= CLIPTextModel(_A       )
         snake_case								= CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       )

         snake_case								= {
             "unet": unet,
             "scheduler": scheduler,
             "vae": vae,
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "safety_checker": None,
             "feature_extractor": None,
         }
         return components





     def 		UpperCAmelCase(self				:   Optional[Any]    ,   _A				:   Optional[Any]    ,   _A				:   str=0       )						->						Dict:
         snake_case								= torch.manual_seed(_A       )
         snake_case								= {
             "prompt": "a photo of the dolomites",
             "generator": generator,
             # Setting height and width to None to prevent OOMs on CPU.
             "height": None,
             "width": None,
             "num_inference_steps": 1,
             "guidance_scale": 6.0,
             "output_type": "numpy",
         }
         return inputs





     def 		UpperCAmelCase(self				:   List[str]       )						->						List[str]:
         snake_case								= "cpu"  # ensure determinism for the device-dependent torch.Generator
         snake_case								= self.get_dummy_components()
         snake_case								= StableDiffusionPanoramaPipeline(**_A       )
         snake_case								= sd_pipe.to(_A       )
         sd_pipe.set_progress_bar_config(disable=_A       )

         snake_case								= self.get_dummy_inputs(_A       )
         snake_case								= sd_pipe(**_A       ).images
         snake_case								= image[0, -3:, -3:, -1]
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case								= np.array([0.61_86, 0.53_74, 0.49_15, 0.41_35, 0.41_14, 0.45_63, 0.51_28, 0.49_77, 0.47_57]       )

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





     def 		UpperCAmelCase(self				:   str       )						->						Dict:
         super().test_inference_batch_consistent(batch_sizes=[1, 2]       )





     def 		UpperCAmelCase(self				:   str       )						->						Any:
         super().test_inference_batch_single_identical(batch_size=2    ,   expected_max_diff=3.25E-3       )





     def 		UpperCAmelCase(self				:   Dict       )						->						Any:
         snake_case								= "cpu"  # ensure determinism for the device-dependent torch.Generator
         snake_case								= self.get_dummy_components()
         snake_case								= StableDiffusionPanoramaPipeline(**_A       )
         snake_case								= sd_pipe.to(_A       )
         sd_pipe.set_progress_bar_config(disable=_A       )

         snake_case								= self.get_dummy_inputs(_A       )
         snake_case								= "french fries"
         snake_case								= sd_pipe(**_A    ,   negative_prompt=_A       )
         snake_case								= output.images
         snake_case								= image[0, -3:, -3:, -1]

         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case								= np.array([0.61_87, 0.53_75, 0.49_15, 0.41_36, 0.41_14, 0.45_63, 0.51_28, 0.49_76, 0.47_57]       )

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





     def 		UpperCAmelCase(self				:   Tuple       )						->						str:
         snake_case								= "cpu"  # ensure determinism for the device-dependent torch.Generator
         snake_case								= self.get_dummy_components()
         snake_case								= StableDiffusionPanoramaPipeline(**_A       )
         snake_case								= sd_pipe.to(_A       )
         sd_pipe.set_progress_bar_config(disable=_A       )

         snake_case								= self.get_dummy_inputs(_A       )
         snake_case								= sd_pipe(**_A    ,   view_batch_size=2       )
         snake_case								= output.images
         snake_case								= image[0, -3:, -3:, -1]

         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case								= np.array([0.61_87, 0.53_75, 0.49_15, 0.41_36, 0.41_14, 0.45_63, 0.51_28, 0.49_76, 0.47_57]       )

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





     def 		UpperCAmelCase(self				:   int       )						->						List[Any]:
         snake_case								= "cpu"  # ensure determinism for the device-dependent torch.Generator
         snake_case								= self.get_dummy_components()
         snake_case								= EulerAncestralDiscreteScheduler(
             beta_start=0.0_00_85    ,   beta_end=0.0_12    ,   beta_schedule="scaled_linear"       )
         snake_case								= StableDiffusionPanoramaPipeline(**_A       )
         snake_case								= sd_pipe.to(_A       )
         sd_pipe.set_progress_bar_config(disable=_A       )

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

         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case								= np.array([0.40_24, 0.65_10, 0.49_01, 0.53_78, 0.58_13, 0.56_22, 0.47_95, 0.44_67, 0.49_52]       )

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





     def 		UpperCAmelCase(self				:   Tuple       )						->						Optional[Any]:
         snake_case								= "cpu"  # ensure determinism for the device-dependent torch.Generator
         snake_case								= self.get_dummy_components()
         snake_case								= PNDMScheduler(
             beta_start=0.0_00_85    ,   beta_end=0.0_12    ,   beta_schedule="scaled_linear"    ,   skip_prk_steps=_A       )
         snake_case								= StableDiffusionPanoramaPipeline(**_A       )
         snake_case								= sd_pipe.to(_A       )
         sd_pipe.set_progress_bar_config(disable=_A       )

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

         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case								= np.array([0.63_91, 0.62_91, 0.48_61, 0.51_34, 0.55_52, 0.45_78, 0.50_32, 0.50_23, 0.45_39]       )

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







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





     def 		UpperCAmelCase(self				:   Any       )						->						Optional[Any]:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





     def 		UpperCAmelCase(self				:   List[Any]    ,   _A				:   Any=0       )						->						Optional[int]:
         snake_case								= torch.manual_seed(_A       )
         snake_case								= {
             "prompt": "a photo of the dolomites",
             "generator": generator,
             "num_inference_steps": 3,
             "guidance_scale": 7.5,
             "output_type": "numpy",
         }
         return inputs





     def 		UpperCAmelCase(self				:   Tuple       )						->						List[Any]:
         snake_case								= "stabilityai/stable-diffusion-2-base"
         snake_case								= DDIMScheduler.from_pretrained(_A    ,   subfolder="scheduler"       )
         snake_case								= StableDiffusionPanoramaPipeline.from_pretrained(_A    ,   scheduler=_A    ,   safety_checker=_A       )
         pipe.to(_A       )
         pipe.set_progress_bar_config(disable=_A       )
         pipe.enable_attention_slicing()

         snake_case								= self.get_inputs()
         snake_case								= pipe(**_A       ).images
         snake_case								= image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_1_2, 2_0_4_8, 3)

         snake_case								= np.array(
             [
                 0.36_96_83_92,
                 0.27_02_53_72,
                 0.32_44_67_66,
                 0.28_37_93_87,
                 0.36_36_32_74,
                 0.30_73_33_47,
                 0.27_10_00_27,
                 0.27_05_41_25,
                 0.25_53_60_96,
             ]       )

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





     def 		UpperCAmelCase(self				:   int       )						->						Any:
         snake_case								= StableDiffusionPanoramaPipeline.from_pretrained(
             "stabilityai/stable-diffusion-2-base"    ,   safety_checker=_A       )
         snake_case								= LMSDiscreteScheduler.from_config(pipe.scheduler.config       )
         pipe.to(_A       )
         pipe.set_progress_bar_config(disable=_A       )
         pipe.enable_attention_slicing()

         snake_case								= self.get_inputs()
         snake_case								= pipe(**_A       ).images
         snake_case								= image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_1_2, 2_0_4_8, 3)

         snake_case								= np.array(
             [
                 [
                     0.0,
                     0.0,
                     0.0,
                     0.0,
                     0.0,
                     0.0,
                     0.0,
                     0.0,
                     0.0,
                 ]
             ]       )

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





     def 		UpperCAmelCase(self				:   Tuple       )						->						Optional[Any]:
         snake_case								= 0

         def callback_fn(_A				:   int    ,   _A				:   int    ,   _A				:   torch.FloatTensor       ) -> None:
             snake_case								= True
             nonlocal number_of_steps
             number_of_steps += 1
             if step == 1:
                 snake_case								= latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 6_4, 2_5_6)
                 snake_case								= latents[0, -3:, -3:, -1]

                 snake_case								= np.array(
                     [
                         0.18_68_18_69,
                         0.33_90_78_16,
                         0.5_36_12_76,
                         0.14_43_28_65,
                         -0.02_85_66_11,
                         -0.73_94_11_23,
                         0.23_39_79_87,
                         0.47_32_26_82,
                         -0.37_82_31_64,
                     ]       )
                 assert np.abs(latents_slice.flatten() - expected_slice       ).max() < 5E-2
             elif step == 2:
                 snake_case								= latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 6_4, 2_5_6)
                 snake_case								= latents[0, -3:, -3:, -1]

                 snake_case								= np.array(
                     [
                         0.18_53_96_45,
                         0.33_98_72_48,
                         0.5_37_85_59,
                         0.14_43_71_42,
                         -0.02_45_52_61,
                         -0.7_33_83_17,
                         0.23_99_07_55,
                         0.47_35_62_72,
                         -0.3_78_65_05,
                     ]       )

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

         snake_case								= False

         snake_case								= "stabilityai/stable-diffusion-2-base"
         snake_case								= DDIMScheduler.from_pretrained(_A    ,   subfolder="scheduler"       )
         snake_case								= StableDiffusionPanoramaPipeline.from_pretrained(_A    ,   scheduler=_A    ,   safety_checker=_A       )
         snake_case								= pipe.to(_A       )
         pipe.set_progress_bar_config(disable=_A       )
         pipe.enable_attention_slicing()

         snake_case								= self.get_inputs()
         pipe(**_A    ,   callback=_A    ,   callback_steps=1       )
         assert callback_fn.has_been_called
         assert number_of_steps == 3





     def 		UpperCAmelCase(self				:   Dict       )						->						str:
         torch.cuda.empty_cache()
         torch.cuda.reset_max_memory_allocated()
         torch.cuda.reset_peak_memory_stats()

         snake_case								= "stabilityai/stable-diffusion-2-base"
         snake_case								= DDIMScheduler.from_pretrained(_A    ,   subfolder="scheduler"       )
         snake_case								= StableDiffusionPanoramaPipeline.from_pretrained(_A    ,   scheduler=_A    ,   safety_checker=_A       )
         snake_case								= pipe.to(_A       )
         pipe.set_progress_bar_config(disable=_A       )
         pipe.enable_attention_slicing(1       )
         pipe.enable_sequential_cpu_offload()

         snake_case								= self.get_inputs()
         snake_case								= pipe(**_A       )

         snake_case								= torch.cuda.max_memory_allocated()
         # make sure that less than 5.2 GB is allocated
         assert mem_bytes < 5.5 * 1_0**9



from operator import delitem, getitem, setitem

import pytest

from data_structures.hashing.hash_map import HashMap



def 						lowercase_      (       A__		)							->							str:
    """simple docstring"""

    return getitem, k



def 						lowercase_      (       A__			,		A__		)							->							str:
    """simple docstring"""

    return setitem, k, v



def 						lowercase_      (       A__		)							->							List[Any]:
    """simple docstring"""

    return delitem, k



def 						lowercase_      (       A__			,		A__			,		*A__		)							->							str:
    """simple docstring"""

    try:
        return fun(A__			,		*A__		), None
    except Exception as e:
        return None, e


_A   =			(
    _set("key_a", "val_a"),
    _set("key_b", "val_b"),
)

_A   =			[
    _set("key_a", "val_a"),
    _set("key_a", "val_b"),
]

_A   =			[
    _set("key_a", "val_a"),
    _set("key_b", "val_b"),
    _del("key_a"),
    _del("key_b"),
    _set("key_a", "val_a"),
    _del("key_a"),
]

_A   =			[
    _get("key_a"),
    _del("key_a"),
    _set("key_a", "val_a"),
    _del("key_a"),
    _del("key_a"),
    _get("key_a"),
]

_A   =			[
    *[_set(x, x) for x in range(5)],  # guaranteed upsize
]

_A   =			[
    *[_set(x, x) for x in range(5)],  # guaranteed upsize
    *[_del(x) for x in range(5)],
    _set("key_a", "val_b"),
]



@pytest.mark.parametrize(
    "operations"			,		(
        pytest.param(_add_items			,		id="add items"		),
        pytest.param(_overwrite_items			,		id="overwrite items"		),
        pytest.param(_delete_items			,		id="delete items"		),
        pytest.param(_access_absent_items			,		id="access absent items"		),
        pytest.param(_add_with_resize_up			,		id="add with resize up"		),
        pytest.param(_add_with_resize_down			,		id="add with resize down"		),
    )			,		)
def 						lowercase_      (       A__		)							->							List[Any]:
    """simple docstring"""

    snake_case								= HashMap(initial_block_size=4		)
    snake_case								= {}
    for _, (fun, *args) in enumerate(A__		):
        snake_case					,							snake_case								= _run_operation(A__			,		A__			,		*A__		)
        snake_case					,							snake_case								= _run_operation(A__			,		A__			,		*A__		)
        assert my_res == py_res
        assert str(A__		) == str(A__		)
        assert set(A__		) == set(A__		)
        assert len(A__		) == len(A__		)
        assert set(my.items()		) == set(py.items()		)





def 						lowercase_      (       )							->							Optional[int]:
    """simple docstring"""

    def is_public(A__		) -> bool:
        return not name.startswith("_"		)

    snake_case								= {name for name in dir({}		) if is_public(A__		)}
    snake_case								= {name for name in dir(HashMap()		) if is_public(A__		)}

    assert dict_public_names > hash_public_names

