






'''simple docstring'''




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    _UpperCAmelCase			(     snake_case_							,    snake_case_							,    unittest.TestCase						):




   """simple docstring"""

   snake_case    				=						StableDiffusionPanoramaPipeline
   snake_case    				=						TEXT_TO_IMAGE_PARAMS
   snake_case    				=						TEXT_TO_IMAGE_BATCH_PARAMS
   snake_case    				=						TEXT_TO_IMAGE_IMAGE_PARAMS
   snake_case    				=						TEXT_TO_IMAGE_IMAGE_PARAMS

   def 						lowerCAmelCase (		self :	str		):





         '''simple docstring'''





         torch.manual_seed(0		)
         _A			     =	UNetaDConditionModel(
             block_out_channels=(32, 64)      ,      layers_per_block=1      ,      sample_size=32      ,      in_channels=4      ,      out_channels=4      ,      down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")      ,      up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")      ,      cross_attention_dim=32      ,      )
         _A			     =	DDIMScheduler()
         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			     =	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      ,      )
         _A			     =	CLIPTextModel(__UpperCAmelCase		)
         _A			     =	CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

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

   def 						lowerCAmelCase (		self :	List[Any]      ,      __UpperCAmelCase :	Any      ,      __UpperCAmelCase :	List[Any]=0		):





         '''simple docstring'''





         _A			     =	torch.manual_seed(__UpperCAmelCase		)
         _A			     =	{
             "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 						lowerCAmelCase (		self :	List[Any]		):





         '''simple docstring'''





         _A			     =	"cpu"  # ensure determinism for the device-dependent torch.Generator
         _A			     =	self.get_dummy_components()
         _A			     =	StableDiffusionPanoramaPipeline(**__UpperCAmelCase		)
         _A			     =	sd_pipe.to(__UpperCAmelCase		)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

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

         _A			     =	np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757]		)

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

   def 						lowerCAmelCase (		self :	List[Any]		):





         '''simple docstring'''





         super().test_inference_batch_consistent(batch_sizes=[1, 2]		)

   def 						lowerCAmelCase (		self :	List[Any]		):





         '''simple docstring'''





         super().test_inference_batch_single_identical(batch_size=2      ,      expected_max_diff=3.25E-3		)

   def 						lowerCAmelCase (		self :	Optional[int]		):





         '''simple docstring'''





         _A			     =	"cpu"  # ensure determinism for the device-dependent torch.Generator
         _A			     =	self.get_dummy_components()
         _A			     =	StableDiffusionPanoramaPipeline(**__UpperCAmelCase		)
         _A			     =	sd_pipe.to(__UpperCAmelCase		)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

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

         assert image.shape == (1, 64, 64, 3)

         _A			     =	np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]		)

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

   def 						lowerCAmelCase (		self :	Optional[Any]		):





         '''simple docstring'''





         _A			     =	"cpu"  # ensure determinism for the device-dependent torch.Generator
         _A			     =	self.get_dummy_components()
         _A			     =	StableDiffusionPanoramaPipeline(**__UpperCAmelCase		)
         _A			     =	sd_pipe.to(__UpperCAmelCase		)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

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

         assert image.shape == (1, 64, 64, 3)

         _A			     =	np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]		)

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

   def 						lowerCAmelCase (		self :	List[str]		):





         '''simple docstring'''





         _A			     =	"cpu"  # ensure determinism for the device-dependent torch.Generator
         _A			     =	self.get_dummy_components()
         _A			     =	EulerAncestralDiscreteScheduler(
             beta_start=0.00085      ,      beta_end=0.012      ,      beta_schedule="scaled_linear"		)
         _A			     =	StableDiffusionPanoramaPipeline(**__UpperCAmelCase		)
         _A			     =	sd_pipe.to(__UpperCAmelCase		)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

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

         assert image.shape == (1, 64, 64, 3)

         _A			     =	np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952]		)

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

   def 						lowerCAmelCase (		self :	Dict		):





         '''simple docstring'''





         _A			     =	"cpu"  # ensure determinism for the device-dependent torch.Generator
         _A			     =	self.get_dummy_components()
         _A			     =	PNDMScheduler(
             beta_start=0.00085      ,      beta_end=0.012      ,      beta_schedule="scaled_linear"      ,      skip_prk_steps=__UpperCAmelCase		)
         _A			     =	StableDiffusionPanoramaPipeline(**__UpperCAmelCase		)
         _A			     =	sd_pipe.to(__UpperCAmelCase		)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

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

         assert image.shape == (1, 64, 64, 3)

         _A			     =	np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539]		)

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





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




   """simple docstring"""

   def 						lowerCAmelCase (		self :	Tuple		):





         '''simple docstring'''





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

   def 						lowerCAmelCase (		self :	int      ,      __UpperCAmelCase :	Union[str, Any]=0		):





         '''simple docstring'''





         _A			     =	torch.manual_seed(__UpperCAmelCase		)
         _A			     =	{
             "prompt": "a photo of the dolomites",
             "generator": generator,
             "num_inference_steps": 3,
             "guidance_scale": 7.5,
             "output_type": "numpy",
         }
         return inputs

   def 						lowerCAmelCase (		self :	List[Any]		):





         '''simple docstring'''





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

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

         assert image.shape == (1, 512, 2048, 3)

         _A			     =	np.array(
             [
                 0.36968392,
                 0.27025372,
                 0.32446766,
                 0.28379387,
                 0.36363274,
                 0.30733347,
                 0.27100027,
                 0.27054125,
                 0.25536096,
             ]		)

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

   def 						lowerCAmelCase (		self :	Optional[Any]		):





         '''simple docstring'''





         _A			     =	StableDiffusionPanoramaPipeline.from_pretrained(
             "stabilityai/stable-diffusion-2-base"      ,      safety_checker=__UpperCAmelCase		)
         _A			     =	LMSDiscreteScheduler.from_config(pipe.scheduler.config		)
         pipe.to(__UpperCAmelCase		)
         pipe.set_progress_bar_config(disable=__UpperCAmelCase		)
         pipe.enable_attention_slicing()

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

         assert image.shape == (1, 512, 2048, 3)

         _A			     =	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 						lowerCAmelCase (		self :	Tuple		):





         '''simple docstring'''





         _A			     =	0

         def callback_fn(__UpperCAmelCase :	int      ,      __UpperCAmelCase :	int      ,      __UpperCAmelCase :	torch.FloatTensor		) -> 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, 256)
                     _A			     =	latents[0, -3:, -3:, -1]

                     _A			     =	np.array(
                         [
                             0.18681869,
                             0.33907816,
                             0.5361276,
                             0.14432865,
                             -0.02856611,
                             -0.73941123,
                             0.23397987,
                             0.47322682,
                             -0.37823164,
                         ]		)
                     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, 256)
                     _A			     =	latents[0, -3:, -3:, -1]

                     _A			     =	np.array(
                         [
                             0.18539645,
                             0.33987248,
                             0.5378559,
                             0.14437142,
                             -0.02455261,
                             -0.7338317,
                             0.23990755,
                             0.47356272,
                             -0.3786505,
                         ]		)

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

         _A			     =	False

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

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

   def 						lowerCAmelCase (		self :	Dict		):





         '''simple docstring'''





         torch.cuda.empty_cache()
         torch.cuda.reset_max_memory_allocated()
         torch.cuda.reset_peak_memory_stats()

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

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

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










import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


__snake_case						:List[str]      		=     '''\
    Text data.
    Second line of data.'''

__snake_case						:Optional[Any]      		=     '''file'''





@pytest.fixture(scope='''session'''				)
def      __snake_case	(			_UpperCAmelCase				):
    __a						       =				tmp_path_factory.mktemp('''data'''				) / (FILE_PATH + '''.zstd''')
    __a						       =				bytes(_UpperCAmelCase   , '''utf-8'''				)
    with zstd.open(_UpperCAmelCase   , '''wb'''				) as f:
        f.write(_UpperCAmelCase				)
    return path





@pytest.fixture
def      __snake_case	(			_UpperCAmelCase				):
    with open(os.path.join(tmpfs.local_root_dir   , _UpperCAmelCase				)   , '''w'''				) as f:
        f.write(_UpperCAmelCase				)
    return FILE_PATH





@pytest.mark.parametrize('''compression_format'''   , ['''gzip''', '''xz''', '''zstd''']				)
def      __snake_case	(			_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase				):
    __a						       =				{'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
    __a						       =				input_paths[compression_format]
    __a						       =				tmp_path / '''cache'''
    __a						       =				DownloadConfig(cache_dir=_UpperCAmelCase   , extract_compressed_file=_UpperCAmelCase				)
    __a						       =				cached_path(_UpperCAmelCase   , download_config=_UpperCAmelCase				)
    with open(_UpperCAmelCase				) as f:
        __a						       =				f.read()
    with open(_UpperCAmelCase				) as f:
        __a						       =				f.read()
    assert extracted_file_content == expected_file_content





@pytest.mark.parametrize('''default_extracted'''   , [True, False]				)
@pytest.mark.parametrize('''default_cache_dir'''   , [True, False]				)
def      __snake_case	(			_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase				):
    __a						       =				'''custom_cache'''
    __a						       =				'''custom_extracted_dir'''
    __a						       =				tmp_path / '''custom_extracted_path'''
    if default_extracted:
        __a						       =				('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
    else:
        monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''   , _UpperCAmelCase				)
        monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''   , str(_UpperCAmelCase				)				)
        __a						       =				custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

    __a						       =				xz_file
    __a						       =				(
        DownloadConfig(extract_compressed_file=_UpperCAmelCase				)
        if default_cache_dir
        else DownloadConfig(cache_dir=tmp_path / custom_cache_dir   , extract_compressed_file=_UpperCAmelCase				)
    )
    __a						       =				cached_path(_UpperCAmelCase   , download_config=_UpperCAmelCase				)
    assert Path(_UpperCAmelCase				).parent.parts[-2:] == expected





def      __snake_case	(			_UpperCAmelCase				):
    # absolute path
    __a						       =				str(Path(_UpperCAmelCase				).resolve()				)
    assert cached_path(_UpperCAmelCase				) == text_file
    # relative path
    __a						       =				str(Path(_UpperCAmelCase				).resolve().relative_to(Path(os.getcwd()				)				)				)
    assert cached_path(_UpperCAmelCase				) == text_file





def      __snake_case	(			_UpperCAmelCase				):
    # absolute path
    __a						       =				str(tmp_path.resolve() / '''__missing_file__.txt'''				)
    with pytest.raises(_UpperCAmelCase				):
        cached_path(_UpperCAmelCase				)
    # relative path
    __a						       =				'''./__missing_file__.txt'''
    with pytest.raises(_UpperCAmelCase				):
        cached_path(_UpperCAmelCase				)





def      __snake_case	(			_UpperCAmelCase				):
    __a						       =				get_from_cache(f'tmp://{tmpfs_file}'				)
    with open(_UpperCAmelCase				) as f:
        __a						       =				f.read()
    assert output_file_content == FILE_CONTENT





@patch('''datasets.config.HF_DATASETS_OFFLINE'''   , _UpperCAmelCase				)
def      __snake_case	(			):
    with pytest.raises(_UpperCAmelCase				):
        cached_path('''https://huggingface.co'''				)





@patch('''datasets.config.HF_DATASETS_OFFLINE'''   , _UpperCAmelCase				)
def      __snake_case	(			_UpperCAmelCase				):
    __a						       =				tmp_path_factory.mktemp('''data'''				) / '''file.html'''
    with pytest.raises(_UpperCAmelCase				):
        http_get('''https://huggingface.co'''   , temp_file=_UpperCAmelCase				)
    with pytest.raises(_UpperCAmelCase				):
        http_head('''https://huggingface.co'''				)





@patch('''datasets.config.HF_DATASETS_OFFLINE'''   , _UpperCAmelCase				)
def      __snake_case	(			_UpperCAmelCase				):
    __a						       =				tmp_path_factory.mktemp('''data'''				) / '''file.html'''
    with pytest.raises(_UpperCAmelCase				):
        ftp_get('''ftp://huggingface.co'''   , temp_file=_UpperCAmelCase				)
    with pytest.raises(_UpperCAmelCase				):
        ftp_head('''ftp://huggingface.co'''				)





@patch('''datasets.config.HF_DATASETS_OFFLINE'''   , _UpperCAmelCase				)
def      __snake_case	(			_UpperCAmelCase				):
    __a						       =				tmp_path_factory.mktemp('''data'''				) / '''file.html'''
    with pytest.raises(_UpperCAmelCase				):
        fsspec_get('''s3://huggingface.co'''   , temp_file=_UpperCAmelCase				)
    with pytest.raises(_UpperCAmelCase				):
        fsspec_head('''s3://huggingface.co'''				)





