


import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

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()




class A  (       UpperCamelCase_   ,							UpperCamelCase_   ,							unittest.TestCase					):
      UpperCamelCase__							:  str				=StableDiffusionXLPipeline
      UpperCamelCase__							:  Optional[int]				=TEXT_TO_IMAGE_PARAMS
      UpperCamelCase__							:  Dict				=TEXT_TO_IMAGE_BATCH_PARAMS
      UpperCamelCase__							:  List[str]				=TEXT_TO_IMAGE_IMAGE_PARAMS
      UpperCamelCase__							:  str				=TEXT_TO_IMAGE_IMAGE_PARAMS






      def 	lowerCamelCase	( self						:					int			) ->						Any:
             """simple docstring"""




             torch.manual_seed(0			)
             _lowerCamelCase	:		Dict          =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')      ,       attention_head_dim=(2, 4)      ,       use_linear_projection=lowercase_      ,       addition_embed_type='text_time'      ,       addition_time_embed_dim=8      ,       transformer_layers_per_block=(1, 2)      ,       projection_class_embeddings_input_dim=80      ,       cross_attention_dim=64      ,       )
             _lowerCamelCase	:		Dict          =EulerDiscreteScheduler(
                 beta_start=0.00085      ,       beta_end=0.012      ,       steps_offset=1      ,       beta_schedule='scaled_linear'      ,       timestep_spacing='leading'      ,       )
             torch.manual_seed(0			)
             _lowerCamelCase	:		List[Any]          =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			)
             _lowerCamelCase	:		Optional[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=32      ,       )
             _lowerCamelCase	:		Dict          =CLIPTextModel(lowercase_			)
             _lowerCamelCase	:		List[Any]          =CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      ,       local_files_only=lowercase_			)

             _lowerCamelCase	:		str          =CLIPTextModelWithProjection(lowercase_			)
             _lowerCamelCase	:		Tuple          =CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      ,       local_files_only=lowercase_			)

             _lowerCamelCase	:		Union[str, Any]          ={
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
                 'text_encoder_2': text_encoder_a,
                 'tokenizer_2': tokenizer_a,
                 # "safety_checker": None,
                 # "feature_extractor": None,
             }
             return components






      def 	lowerCamelCase	( self						:					Union[str, Any]      ,       lowercase_						:					Optional[Any]      ,       lowercase_						:					int=0			) ->						Optional[Any]:
             """simple docstring"""




             if str(lowercase_			).startswith('mps'			):
                    _lowerCamelCase	:		Union[str, Any]          =torch.manual_seed(lowercase_			)
             else:
                    _lowerCamelCase	:		List[Any]          =torch.Generator(device=lowercase_			).manual_seed(lowercase_			)
             _lowerCamelCase	:		Optional[int]          ={
                 'prompt': 'A painting of a squirrel eating a burger',
                 'generator': generator,
                 'num_inference_steps': 2,
                 'guidance_scale': 5.0,
                 'output_type': 'numpy',
             }
             return inputs






      def 	lowerCamelCase	( self						:					Tuple			) ->						List[Any]:
             """simple docstring"""




             _lowerCamelCase	:		Any          ='cpu'  # ensure determinism for the device-dependent torch.Generator
             _lowerCamelCase	:		Union[str, Any]          =self.get_dummy_components()
             _lowerCamelCase	:		Any          =StableDiffusionXLPipeline(**lowercase_			)
             _lowerCamelCase	:		str          =sd_pipe.to(lowercase_			)
             sd_pipe.set_progress_bar_config(disable=lowercase_			)

             _lowerCamelCase	:		str          =self.get_dummy_inputs(lowercase_			)
             _lowerCamelCase	:		Optional[Any]          =sd_pipe(**lowercase_			).images
             _lowerCamelCase	:		List[str]          =image[0, -3:, -3:, -1]

             assert image.shape == (1, 64, 64, 3)
             _lowerCamelCase	:		Any          =np.array([0.5873, 0.6128, 0.4797, 0.5122, 0.5674, 0.4639, 0.5227, 0.5149, 0.4747]			)

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






      def 	lowerCamelCase	( self						:					Tuple			) ->						Optional[Any]:
             """simple docstring"""




             _lowerCamelCase	:		List[str]          =self.get_dummy_components()
             _lowerCamelCase	:		Union[str, Any]          =StableDiffusionXLPipeline(**lowercase_			)
             _lowerCamelCase	:		Optional[Any]          =sd_pipe.to(lowercase_			)
             _lowerCamelCase	:		Optional[Any]          =sd_pipe.to(lowercase_			)
             sd_pipe.set_progress_bar_config(disable=lowercase_			)

             # forward without prompt embeds
             _lowerCamelCase	:		Union[str, Any]          =self.get_dummy_inputs(lowercase_			)
             _lowerCamelCase	:		Union[str, Any]          =3 * ['this is a negative prompt']
             _lowerCamelCase	:		Dict          =negative_prompt
             _lowerCamelCase	:		List[Any]          =3 * [inputs['prompt']]

             _lowerCamelCase	:		Dict          =sd_pipe(**lowercase_			)
             _lowerCamelCase	:		Optional[int]          =output.images[0, -3:, -3:, -1]

             # forward with prompt embeds
             _lowerCamelCase	:		Union[str, Any]          =self.get_dummy_inputs(lowercase_			)
             _lowerCamelCase	:		str          =3 * ['this is a negative prompt']
             _lowerCamelCase	:		Optional[int]          =3 * [inputs.pop('prompt'			)]

             (
                 (
                 _lowerCamelCase
             )     ,					(
                 _lowerCamelCase
             )     ,					(
                 _lowerCamelCase
             )     ,					(
                 _lowerCamelCase
             )     ,					
             )	:		Optional[Any]          =sd_pipe.encode_prompt(lowercase_      ,       negative_prompt=lowercase_			)

             _lowerCamelCase	:		Optional[int]          =sd_pipe(
                 **lowercase_      ,       prompt_embeds=lowercase_      ,       negative_prompt_embeds=lowercase_      ,       pooled_prompt_embeds=lowercase_      ,       negative_pooled_prompt_embeds=lowercase_      ,       )
             _lowerCamelCase	:		List[Any]          =output.images[0, -3:, -3:, -1]

             # make sure that it's equal
             assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()			).max() < 1E-4






      def 	lowerCamelCase	( self						:					int			) ->						Optional[Any]:
             """simple docstring"""




             super().test_attention_slicing_forward_pass(expected_max_diff=3E-3			)






      def 	lowerCamelCase	( self						:					Union[str, Any]			) ->						Optional[Any]:
             """simple docstring"""




             super().test_inference_batch_single_identical(expected_max_diff=3E-3			)




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






      def 	lowerCamelCase	( self						:					Any			) ->						int:
             """simple docstring"""




             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()






      def 	lowerCamelCase	( self						:					int      ,       lowercase_						:					Dict      ,       lowercase_						:					int="cpu"      ,       lowercase_						:					Any=torch.floataa      ,       lowercase_						:					List[str]=0			) ->						int:
             """simple docstring"""




             _lowerCamelCase	:		Any          =torch.Generator(device=lowercase_			).manual_seed(lowercase_			)
             _lowerCamelCase	:		Optional[int]          =np.random.RandomState(lowercase_			).standard_normal((1, 4, 64, 64)			)
             _lowerCamelCase	:		Any          =torch.from_numpy(lowercase_			).to(device=lowercase_      ,       dtype=lowercase_			)
             _lowerCamelCase	:		int          ={
                 'prompt': 'a photograph of an astronaut riding a horse',
                 'latents': latents,
                 'generator': generator,
                 'num_inference_steps': 3,
                 'guidance_scale': 7.5,
                 'output_type': 'numpy',
             }
             return inputs






      def 	lowerCamelCase	( self						:					Dict			) ->						Union[str, Any]:
             """simple docstring"""




             _lowerCamelCase	:		Optional[int]          =DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base'			)
             pipe.to(lowercase_			)
             pipe.set_progress_bar_config(disable=lowercase_			)

             _lowerCamelCase	:		str          =self.get_inputs(lowercase_			)
             _lowerCamelCase	:		Any          =pipe(**lowercase_			).images
             _lowerCamelCase	:		Any          =image[0, -3:, -3:, -1].flatten()

             assert image.shape == (1, 512, 512, 3)
             _lowerCamelCase	:		Optional[Any]          =np.array([0.49493, 0.47896, 0.40798, 0.54214, 0.53212, 0.48202, 0.47656, 0.46329, 0.48506]			)
             assert np.abs(image_slice - expected_slice			).max() < 7E-3






import bisect
import copy
from collections import defaultdict

import numpy as np
from torch.utils.data import BatchSampler, Sampler

from utils import logger




def      a_				(   SCREAMING_SNAKE_CASE__ :				Tuple		,      SCREAMING_SNAKE_CASE__ :				Optional[int]					):
       '''simple docstring'''





       _lowerCamelCase	:		Union[str, Any]          =copy.deepcopy(SCREAMING_SNAKE_CASE__					)
       _lowerCamelCase	:		Union[str, Any]          =sorted(SCREAMING_SNAKE_CASE__					)
       _lowerCamelCase	:		List[Any]          =[bisect.bisect_right(SCREAMING_SNAKE_CASE__		,      SCREAMING_SNAKE_CASE__					) for y in x]
       return quantized




def      a_				(   SCREAMING_SNAKE_CASE__ :				Any		,      SCREAMING_SNAKE_CASE__ :				Optional[int]=0					):
       '''simple docstring'''





       _lowerCamelCase	:		List[Any]          =np.arange(start=3		,      stop=SCREAMING_SNAKE_CASE__		,      step=4					).tolist() if k > 0 else [10]
       _lowerCamelCase	:		List[Any]          =_quantize(SCREAMING_SNAKE_CASE__		,      SCREAMING_SNAKE_CASE__					)
       # count number of elements per group
       _lowerCamelCase	:		Union[str, Any]          =np.unique(SCREAMING_SNAKE_CASE__		,      return_counts=SCREAMING_SNAKE_CASE__					)[1]
       _lowerCamelCase	:		int          =[0] + bins + [np.inf]
       logger.info('Using {} as bins for aspect lengths quantization'.format(SCREAMING_SNAKE_CASE__					)					)
       logger.info('Count of instances per bin: {}'.format(SCREAMING_SNAKE_CASE__					)					)
       return groups




class A  (       UpperCamelCase_					):






      def __init__( self						:					Optional[Any]      ,       lowercase_						:					int      ,       lowercase_						:					Optional[int]      ,       lowercase_						:					int			) ->						Any:
             """simple docstring"""




             if not isinstance(lowercase_      ,       lowercase_			):
                    raise ValueError(
                        'sampler should be an instance of torch.utils.data.Sampler, but got sampler={}'.format(lowercase_			)			)
             _lowerCamelCase	:		Any          =sampler
             _lowerCamelCase	:		Dict          =group_ids
             _lowerCamelCase	:		Dict          =batch_size






      def __iter__( self						:					Optional[int]			) ->						str:
             """simple docstring"""




             _lowerCamelCase	:		Union[str, Any]          =defaultdict(lowercase_			)
             _lowerCamelCase	:		Union[str, Any]          =defaultdict(lowercase_			)

             _lowerCamelCase	:		Any          =0
             for idx in self.sampler:
                    _lowerCamelCase	:		Any          =self.group_ids[idx]
                    buffer_per_group[group_id].append(lowercase_			)
                    samples_per_group[group_id].append(lowercase_			)
                    if len(buffer_per_group[group_id]			) == self.batch_size:
                           yield buffer_per_group[group_id]  # TODO
                           num_batches += 1
                           del buffer_per_group[group_id]
                    assert len(buffer_per_group[group_id]			) < self.batch_size

             # now we have run out of elements that satisfy
             # the group criteria, let's return the remaining
             # elements so that the size of the sampler is
             # deterministic
             _lowerCamelCase	:		Union[str, Any]          =len(self			)
             _lowerCamelCase	:		Dict          =expected_num_batches - num_batches
             if num_remaining > 0:
                    # for the remaining batches, group the batches by similar lengths
                    _lowerCamelCase	:		Union[str, Any]          =[]
                    for group_id, idxs in sorted(buffer_per_group.items()      ,       key=lambda lowercase_			: x[0]			):
                           batch_idx.extend(lowercase_			)
                           if len(lowercase_			) >= self.batch_size:
                                  yield batch_idx[: self.batch_size]
                                  _lowerCamelCase	:		str          =batch_idx[self.batch_size :]
                                  num_remaining -= 1
                    if len(lowercase_			) > 0:
                           yield batch_idx
                           num_remaining -= 1
             assert num_remaining == 0






      def __len__( self						:					Optional[int]			) ->						Union[str, Any]:
             """simple docstring"""




             return (len(self.sampler			) + self.batch_size - 1) // self.batch_size


