



'''simple docstring'''

import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()



def      __lowerCamelCase     (      lowerCAmelCase_  )			->      Union[str, Any]:
		_a     :      str        =							[tensor.shape for tensor in tensor_list]
		return all(shape == shapes[0] for shape in shapes[1:]  )

class 	__magic_name__						(    _UpperCamelCase			,						_UpperCamelCase			,						_UpperCamelCase			,						unittest.TestCase   ):
							lowerCAmelCase       :    int   				=					StableDiffusionLatentUpscalePipeline
							lowerCAmelCase       :    Optional[Any]   				=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
							    'height',
							    'width',
							    'cross_attention_kwargs',
							    'negative_prompt_embeds',
							    'prompt_embeds',
							}
							lowerCAmelCase       :    Union[str, Any]   				=					PipelineTesterMixin.required_optional_params - {'num_images_per_prompt'}
							lowerCAmelCase       :    Union[str, Any]   				=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
							lowerCAmelCase       :    Dict   				=					frozenset(
							    []   )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
							lowerCAmelCase       :    List[Any]   				=					frozenset([]   )

							lowerCAmelCase       :    List[str]   				=					True







							@property
							def 							__lowercase			(							self					:       List[str]		):
									_a     :      int        =							1
									_a     :      str        =							4
									_a     :      str        =							(16, 16)

									_a     :      Dict        =							floats_tensor((batch_size, num_channels) + sizes	,rng=random.Random(0		)		).to(_UpperCAmelCase		)
									return image







							def 							__lowercase			(							self					:       List[str]		):
									torch.manual_seed(0		)
									_a     :      Tuple        =							UNetaDConditionModel(
									    act_fn='gelu'	,attention_head_dim=8	,norm_num_groups=_UpperCAmelCase	,block_out_channels=[32, 32, 64, 64]	,time_cond_proj_dim=160	,conv_in_kernel=1	,conv_out_kernel=1	,cross_attention_dim=32	,down_block_types=(
									        'KDownBlock2D',
									        'KCrossAttnDownBlock2D',
									        'KCrossAttnDownBlock2D',
									        'KCrossAttnDownBlock2D',
									    )	,in_channels=8	,mid_block_type=_UpperCAmelCase	,only_cross_attention=_UpperCAmelCase	,out_channels=5	,resnet_time_scale_shift='scale_shift'	,time_embedding_type='fourier'	,timestep_post_act='gelu'	,up_block_types=('KCrossAttnUpBlock2D', 'KCrossAttnUpBlock2D', 'KCrossAttnUpBlock2D', 'KUpBlock2D')	,)
									_a     :      Dict        =							AutoencoderKL(
									    block_out_channels=[32, 32, 64, 64]	,in_channels=3	,out_channels=3	,down_block_types=[
									        'DownEncoderBlock2D',
									        'DownEncoderBlock2D',
									        'DownEncoderBlock2D',
									        'DownEncoderBlock2D',
									    ]	,up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D']	,latent_channels=4	,)
									_a     :      Optional[int]        =							EulerDiscreteScheduler(prediction_type='sample'		)
									_a     :      List[str]        =							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='quick_gelu'	,projection_dim=512	,)
									_a     :      str        =							CLIPTextModel(_UpperCAmelCase		)
									_a     :      str        =							CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

									_a     :      Dict        =							{
									    'unet': model.eval(),
									    'vae': vae.eval(),
									    'scheduler': scheduler,
									    'text_encoder': text_encoder,
									    'tokenizer': tokenizer,
									}

									return components







							def 							__lowercase			(							self					:       str	,_UpperCAmelCase					:       Optional[int]	,_UpperCAmelCase					:       int=0		):
									if str(_UpperCAmelCase		).startswith('mps'		):
											_a     :      List[str]        =							torch.manual_seed(_UpperCAmelCase		)
									else:
											_a     :      List[Any]        =							torch.Generator(device=_UpperCAmelCase		).manual_seed(_UpperCAmelCase		)
									_a     :      int        =							{
									    'prompt': 'A painting of a squirrel eating a burger',
									    'image': self.dummy_image.cpu(),
									    'generator': generator,
									    'num_inference_steps': 2,
									    'output_type': 'numpy',
									}
									return inputs







							def 							__lowercase			(							self					:       int		):
									_a     :      Tuple        =							'cpu'

									_a     :      Tuple        =							self.get_dummy_components()
									_a     :      Any        =							self.pipeline_class(**_UpperCAmelCase		)
									pipe.to(_UpperCAmelCase		)
									pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_a     :      Dict        =							self.get_dummy_inputs(_UpperCAmelCase		)
									_a     :      List[Any]        =							pipe(**_UpperCAmelCase		).images
									_a     :      int        =							image[0, -3:, -3:, -1]

									self.assertEqual(image.shape	,(1, 256, 256, 3)		)
									_a     :      Union[str, Any]        =							np.array(
									    [0.47_22_24_12, 0.41_92_16_33, 0.44_71_74_34, 0.46_87_41_92, 0.42_58_82_58, 0.46_15_07_26, 0.4_67_75_34, 0.45_58_38_32, 0.48_57_90_55]		)
									_a     :      Tuple        =							np.abs(image_slice.flatten() - expected_slice		).max()
									self.assertLessEqual(_UpperCAmelCase	,1E-3		)







							def 							__lowercase			(							self					:       Any		):
									super().test_attention_slicing_forward_pass(expected_max_diff=7E-3		)







							def 							__lowercase			(							self					:       Any		):
									super().test_cpu_offload_forward_pass(expected_max_diff=3E-3		)







							def 							__lowercase			(							self					:       str		):
									super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3		)







							def 							__lowercase			(							self					:       str		):
									super().test_inference_batch_single_identical(expected_max_diff=7E-3		)







							def 							__lowercase			(							self					:       List[str]		):
									super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3E-3		)







							def 							__lowercase			(							self					:       Dict		):
									super().test_save_load_local(expected_max_difference=3E-3		)







							def 							__lowercase			(							self					:       Union[str, Any]		):
									super().test_save_load_optional_components(expected_max_difference=3E-3		)







							def 							__lowercase			(							self					:       Tuple		):
									_a     :      int        =							[
									    'DDIMScheduler',
									    'DDPMScheduler',
									    'PNDMScheduler',
									    'HeunDiscreteScheduler',
									    'EulerAncestralDiscreteScheduler',
									    'KDPM2DiscreteScheduler',
									    'KDPM2AncestralDiscreteScheduler',
									    'DPMSolverSDEScheduler',
									]
									_a     :      Optional[int]        =							self.get_dummy_components()
									_a     :      Optional[int]        =							self.pipeline_class(**_UpperCAmelCase		)

									# make sure that PNDM does not need warm-up
									pipe.scheduler.register_to_config(skip_prk_steps=_UpperCAmelCase		)

									pipe.to(_UpperCAmelCase		)
									pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
									_a     :      Union[str, Any]        =							self.get_dummy_inputs(_UpperCAmelCase		)
									_a     :      Tuple        =							2

									_a     :      int        =							[]
									for scheduler_enum in KarrasDiffusionSchedulers:
											if scheduler_enum.name in skip_schedulers:
													# no sigma schedulers are not supported
													# no schedulers
													continue

											_a     :      int        =							getattr(_UpperCAmelCase	,scheduler_enum.name		)
											_a     :      int        =							scheduler_cls.from_config(pipe.scheduler.config		)
											_a     :      List[str]        =							pipe(**_UpperCAmelCase		)[0]
											outputs.append(_UpperCAmelCase		)

									assert check_same_shape(_UpperCAmelCase		)

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







							def 							__lowercase			(							self					:       Dict		):
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()







							def 							__lowercase			(							self					:       List[str]		):
									_a     :      Dict        =							torch.manual_seed(33		)

									_a     :      str        =							StableDiffusionPipeline.from_pretrained('CompVis/stable-diffusion-v1-4'	,torch_dtype=torch.floataa		)
									pipe.to('cuda'		)

									_a     :      List[str]        =							StableDiffusionLatentUpscalePipeline.from_pretrained(
									    'stabilityai/sd-x2-latent-upscaler'	,torch_dtype=torch.floataa		)
									upscaler.to('cuda'		)

									_a     :      Optional[Any]        =							'a photo of an astronaut high resolution, unreal engine, ultra realistic'

									_a     :      Any        =							pipe(_UpperCAmelCase	,generator=_UpperCAmelCase	,output_type='latent'		).images

									_a     :      Optional[int]        =							upscaler(
									    prompt=_UpperCAmelCase	,image=_UpperCAmelCase	,num_inference_steps=20	,guidance_scale=0	,generator=_UpperCAmelCase	,output_type='np'	,).images[0]

									_a     :      Tuple        =							load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy'		)
									assert np.abs((expected_image - image).mean()		) < 5E-2







							def 							__lowercase			(							self					:       Optional[int]		):
									_a     :      Any        =							torch.manual_seed(33		)

									_a     :      Optional[int]        =							StableDiffusionLatentUpscalePipeline.from_pretrained(
									    'stabilityai/sd-x2-latent-upscaler'	,torch_dtype=torch.floataa		)
									upscaler.to('cuda'		)

									_a     :      Tuple        =							'the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas'

									_a     :      Dict        =							load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png'		)

									_a     :      Optional[Any]        =							upscaler(
									    prompt=_UpperCAmelCase	,image=_UpperCAmelCase	,num_inference_steps=20	,guidance_scale=0	,generator=_UpperCAmelCase	,output_type='np'	,).images[0]

									_a     :      Dict        =							load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy'		)
									assert np.abs((expected_image - image).max()		) < 5E-2









'''simple docstring'''

# Copyright 2022 The HuggingFace Team and The OpenBMB Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


__lowerCAmelCase									=   {
    '''configuration_cpmant''': ['''CPMANT_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''CpmAntConfig'''],
    '''tokenization_cpmant''': ['''CpmAntTokenizer'''],
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__lowerCAmelCase									=   [
							    '''CPMANT_PRETRAINED_MODEL_ARCHIVE_LIST''',
							    '''CpmAntForCausalLM''',
							    '''CpmAntModel''',
							    '''CpmAntPreTrainedModel''',
							]


if TYPE_CHECKING:
							from .configuration_cpmant import CPMANT_PRETRAINED_CONFIG_ARCHIVE_MAP, CpmAntConfig
							from .tokenization_cpmant import CpmAntTokenizer

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_cpmant import (
														    CPMANT_PRETRAINED_MODEL_ARCHIVE_LIST,
														    CpmAntForCausalLM,
														    CpmAntModel,
														    CpmAntPreTrainedModel,
														)


else:
							import sys

							__lowerCAmelCase									=   _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




