



'''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






import os
import zipfile

import pytest

from datasets.utils.extract import (
    BzipaExtractor,
    Extractor,
    GzipExtractor,
    LzaExtractor,
    SevenZipExtractor,
    TarExtractor,
    XzExtractor,
    ZipExtractor,
    ZstdExtractor,
)

from .utils import require_lza, require_pyazr, require_zstandard




@pytest.mark.parametrize(
    '''compression_format, is_archive'''						,   [
        ('''7z''', True),
        ('''bz2''', False),
        ('''gzip''', False),
        ('''lz4''', False),
        ('''tar''', True),
        ('''xz''', False),
        ('''zip''', True),
        ('''zstd''', False),
    ]						,   )
def 				a						(							snake_case__:					int						,   snake_case__:					Tuple						,   snake_case__:					Dict						,   snake_case__:					Dict						,   snake_case__:					List[Any]						,   snake_case__:					int						,   snake_case__:					List[Any]						,   snake_case__:					Optional[int]						,   snake_case__:					str						,   snake_case__:					Union[str, Any]						,   snake_case__:					List[str]						,   snake_case__:					int						,   ):


				'''simple docstring'''
				lowercase_       =					{
				    '''7z''': (seven_zip_file, SevenZipExtractor),
				    '''bz2''': (bza_file, BzipaExtractor),
				    '''gzip''': (gz_file, GzipExtractor),
				    '''lz4''': (lza_file, LzaExtractor),
				    '''tar''': (tar_file, TarExtractor),
				    '''xz''': (xz_file, XzExtractor),
				    '''zip''': (zip_file, ZipExtractor),
				    '''zstd''': (zstd_file, ZstdExtractor),
				}
				lowercase_  ,	lowercase_       =					input_paths_and_base_extractors[compression_format]
				if input_path is None:
								lowercase_       =					F'''for \'{compression_format}\' compression_format, '''
								if compression_format == "7z":
												reason += require_pyazr.kwargs["reason"]
								elif compression_format == "lz4":
												reason += require_lza.kwargs["reason"]
								elif compression_format == "zstd":
												reason += require_zstandard.kwargs["reason"]
								pytest.skip(snake_case__					)
				assert base_extractor.is_extractable(snake_case__					)
				lowercase_       =					tmp_path / ('''extracted''' if is_archive else '''extracted.txt''')
				base_extractor.extract(snake_case__						,   snake_case__					)
				if is_archive:
								assert output_path.is_dir()
								for file_path in output_path.iterdir():
												assert file_path.name == text_file.name
												lowercase_       =					file_path.read_text(encoding='''utf-8'''					)
				else:
								lowercase_       =					output_path.read_text(encoding='''utf-8'''					)
				lowercase_       =					text_file.read_text(encoding='''utf-8'''					)
				assert extracted_file_content == expected_file_content




@pytest.mark.parametrize(
    '''compression_format, is_archive'''						,   [
        ('''7z''', True),
        ('''bz2''', False),
        ('''gzip''', False),
        ('''lz4''', False),
        ('''tar''', True),
        ('''xz''', False),
        ('''zip''', True),
        ('''zstd''', False),
    ]						,   )
def 				a						(							snake_case__:					List[Any]						,   snake_case__:					int						,   snake_case__:					Optional[int]						,   snake_case__:					Union[str, Any]						,   snake_case__:					List[Any]						,   snake_case__:					Tuple						,   snake_case__:					Optional[int]						,   snake_case__:					List[str]						,   snake_case__:					Union[str, Any]						,   snake_case__:					Tuple						,   snake_case__:					int						,   snake_case__:					Optional[int]						,   ):


				'''simple docstring'''
				lowercase_       =					{
				    '''7z''': seven_zip_file,
				    '''bz2''': bza_file,
				    '''gzip''': gz_file,
				    '''lz4''': lza_file,
				    '''tar''': tar_file,
				    '''xz''': xz_file,
				    '''zip''': zip_file,
				    '''zstd''': zstd_file,
				}
				lowercase_       =					input_paths[compression_format]
				if input_path is None:
								lowercase_       =					F'''for \'{compression_format}\' compression_format, '''
								if compression_format == "7z":
												reason += require_pyazr.kwargs["reason"]
								elif compression_format == "lz4":
												reason += require_lza.kwargs["reason"]
								elif compression_format == "zstd":
												reason += require_zstandard.kwargs["reason"]
								pytest.skip(snake_case__					)
				lowercase_       =					Extractor.infer_extractor_format(snake_case__					)
				assert extractor_format is not None
				lowercase_       =					tmp_path / ('''extracted''' if is_archive else '''extracted.txt''')
				Extractor.extract(snake_case__						,   snake_case__						,   snake_case__					)
				if is_archive:
								assert output_path.is_dir()
								for file_path in output_path.iterdir():
												assert file_path.name == text_file.name
												lowercase_       =					file_path.read_text(encoding='''utf-8'''					)
				else:
								lowercase_       =					output_path.read_text(encoding='''utf-8'''					)
				lowercase_       =					text_file.read_text(encoding='''utf-8'''					)
				assert extracted_file_content == expected_file_content




@pytest.fixture
def 				a						(							snake_case__:					Union[str, Any]						,   snake_case__:					List[Any]					):


				'''simple docstring'''
				import tarfile

				lowercase_       =					tmp_path / '''data_dot_dot'''
				directory.mkdir()
				lowercase_       =					directory / '''tar_file_with_dot_dot.tar'''
				with tarfile.TarFile(snake_case__						,   '''w'''					) as f:
								f.add(snake_case__						,   arcname=os.path.join('''..'''						,   text_file.name					)					)
				return path




@pytest.fixture
def 				a						(							snake_case__:					int					):


				'''simple docstring'''
				import tarfile

				lowercase_       =					tmp_path / '''data_sym_link'''
				directory.mkdir()
				lowercase_       =					directory / '''tar_file_with_sym_link.tar'''
				os.symlink('''..'''						,   directory / '''subdir'''						,   target_is_directory=snake_case__					)
				with tarfile.TarFile(snake_case__						,   '''w'''					) as f:
								f.add(str(directory / '''subdir'''					)						,   arcname='''subdir'''					)  # str required by os.readlink on Windows and Python < 3.8
				return path




@pytest.mark.parametrize(
    '''insecure_tar_file, error_log'''						,   [('''tar_file_with_dot_dot''', '''illegal path'''), ('''tar_file_with_sym_link''', '''Symlink''')]						,   )
def 				a						(							snake_case__:					List[Any]						,   snake_case__:					Optional[int]						,   snake_case__:					List[str]						,   snake_case__:					List[str]						,   snake_case__:					int						,   snake_case__:					Optional[Any]					):


				'''simple docstring'''
				lowercase_       =					{
				    '''tar_file_with_dot_dot''': tar_file_with_dot_dot,
				    '''tar_file_with_sym_link''': tar_file_with_sym_link,
				}
				lowercase_       =					insecure_tar_files[insecure_tar_file]
				lowercase_       =					tmp_path / '''extracted'''
				TarExtractor.extract(snake_case__						,   snake_case__					)
				assert caplog.text
				for record in caplog.records:
								assert record.levelname == "ERROR"
								assert error_log in record.msg




def 				a						(							snake_case__:					Optional[int]					):


				'''simple docstring'''
				# We should have less false positives than zipfile.is_zipfile
				# We do that by checking only the magic number
				lowercase_       =					tmpdir / '''not_a_zip_file'''
				# From: https://github.com/python/cpython/pull/5053
				lowercase_       =					(
				    B'''\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00'''
				    B'''\x00\x02\x08\x06\x00\x00\x00\x99\x81\xb6\'\x00\x00\x00\x15I'''
				    B'''DATx\x01\x01\n\x00\xf5\xff\x00PK\x05\x06\x00PK\x06\x06\x07'''
				    B'''\xac\x01N\xc6|a\r\x00\x00\x00\x00IEND\xaeB`\x82'''
				)
				with not_a_zip_file.open('''wb'''					) as f:
								f.write(snake_case__					)
				assert zipfile.is_zipfile(str(snake_case__					)					)  # is a false positive for `zipfile`
				assert not ZipExtractor.is_extractable(snake_case__					)  # but we're right




