"""simple docstring"""




import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()
class 						__A							( SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_    ,	unittest.TestCase		):
     _UpperCamelCase							:			Optional[int]							=		StableUnCLIPPipeline
     _UpperCamelCase							:			Optional[Any]							=		TEXT_TO_IMAGE_PARAMS
     _UpperCamelCase							:			Optional[int]							=		TEXT_TO_IMAGE_BATCH_PARAMS
     _UpperCamelCase							:			str							=		TEXT_TO_IMAGE_IMAGE_PARAMS
     _UpperCamelCase							:			Tuple							=		TEXT_TO_IMAGE_IMAGE_PARAMS

     # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
     _UpperCamelCase							:			Optional[Any]							=		False






     def     __A  (		self  ):
       _lowerCAmelCase						:       Any  			=     32
       _lowerCAmelCase						:       str  			=     embedder_hidden_size

       # prior components

       torch.manual_seed(0  )
       _lowerCAmelCase						:       List[Any]  			=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       str  			=     CLIPTextModelWithProjection(
           CLIPTextConfig(
               bos_token_id=0				,     eos_token_id=2				,     hidden_size=a__				,     projection_dim=a__				,     intermediate_size=37				,     layer_norm_eps=1e-05				,     num_attention_heads=4				,     num_hidden_layers=5				,     pad_token_id=1				,     vocab_size=1000				,     )  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       List[Any]  			=     PriorTransformer(
           num_attention_heads=2				,     attention_head_dim=12				,     embedding_dim=a__				,     num_layers=1				,     )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Optional[Any]  			=     DDPMScheduler(
           variance_type="""fixed_small_log"""				,     prediction_type="""sample"""				,     num_train_timesteps=1000				,     clip_sample=a__				,     clip_sample_range=5.0				,     beta_schedule="""squaredcos_cap_v2"""				,     )

       # regular denoising components

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Any  			=     StableUnCLIPImageNormalizer(embedding_dim=a__  )
       _lowerCAmelCase						:       int  			=     DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Union[str, Any]  			=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       List[Any]  			=     CLIPTextModel(
           CLIPTextConfig(
               bos_token_id=0				,     eos_token_id=2				,     hidden_size=a__				,     projection_dim=32				,     intermediate_size=37				,     layer_norm_eps=1e-05				,     num_attention_heads=4				,     num_hidden_layers=5				,     pad_token_id=1				,     vocab_size=1000				,     )  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Tuple  			=     UNetaDConditionModel(
           sample_size=32				,     in_channels=4				,     out_channels=4				,     down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")				,     up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")				,     block_out_channels=(32, 64)				,     attention_head_dim=(2, 4)				,     class_embed_type="""projection"""				,     projection_class_embeddings_input_dim=embedder_projection_dim * 2				,     cross_attention_dim=a__				,     layers_per_block=1				,     upcast_attention=a__				,     use_linear_projection=a__				,     )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       List[Any]  			=     DDIMScheduler(
           beta_schedule="""scaled_linear"""				,     beta_start=0.0_0_0_8_5				,     beta_end=0.0_1_2				,     prediction_type="""v_prediction"""				,     set_alpha_to_one=a__				,     steps_offset=1				,     )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       str  			=     AutoencoderKL()

       _lowerCAmelCase						:       Any  			=     {
           # prior components
           """prior_tokenizer""": prior_tokenizer,
           """prior_text_encoder""": prior_text_encoder,
           """prior""": prior,
           """prior_scheduler""": prior_scheduler,
           # image noising components
           """image_normalizer""": image_normalizer,
           """image_noising_scheduler""": image_noising_scheduler,
           # regular denoising components
           """tokenizer""": tokenizer,
           """text_encoder""": text_encoder,
           """unet""": unet,
           """scheduler""": scheduler,
           """vae""": vae,
       }

       return components






     def     __A  (		self				,     a__				,     a__=0  ):
       if str(a__  ).startswith("""mps"""  ):
         _lowerCAmelCase						:       Optional[int]  			=     torch.manual_seed(a__  )
       else:
         _lowerCAmelCase						:       Tuple  			=     torch.Generator(device=a__  ).manual_seed(a__  )
       _lowerCAmelCase						:       List[Any]  			=     {
           """prompt""": """A painting of a squirrel eating a burger""",
           """generator""": generator,
           """num_inference_steps""": 2,
           """prior_num_inference_steps""": 2,
           """output_type""": """numpy""",
       }
       return inputs






     def     __A  (		self  ):
       _lowerCAmelCase						:       Optional[Any]  			=     torch_device == """cpu"""

       self._test_attention_slicing_forward_pass(test_max_difference=a__  )






     def     __A  (		self  ):
       _lowerCAmelCase						:       Union[str, Any]  			=     torch_device in ["""cpu""", """mps"""]

       self._test_inference_batch_single_identical(test_max_difference=a__  )







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






     def     __A  (		self  ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()






     def     __A  (		self  ):
       _lowerCAmelCase						:       Optional[int]  			=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""  )

       _lowerCAmelCase						:       Optional[Any]  			=     StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""				,     torch_dtype=torch.floataa  )
       pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       # stable unclip will oom when integration tests are run on a V100,
       # so turn on memory savings
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _lowerCAmelCase						:       str  			=     torch.Generator(device="""cpu"""  ).manual_seed(0  )
       _lowerCAmelCase						:       Tuple  			=     pipe("""anime turle"""				,     generator=a__				,     output_type="""np"""  )

       _lowerCAmelCase						:       Optional[int]  			=     output.images[0]

       assert image.shape == (768, 768, 3)

       assert_mean_pixel_difference(a__				,     a__  )






     def     __A  (		self  ):
       torch.cuda.empty_cache()
       torch.cuda.reset_max_memory_allocated()
       torch.cuda.reset_peak_memory_stats()

       _lowerCAmelCase						:       Optional[int]  			=     StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""				,     torch_dtype=torch.floataa  )
       _lowerCAmelCase						:       int  			=     pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _lowerCAmelCase						:       Any  			=     pipe(
           """anime turtle"""				,     prior_num_inference_steps=2				,     num_inference_steps=2				,     output_type="""np"""				,     )

       _lowerCAmelCase						:       int  			=     torch.cuda.max_memory_allocated()
       # make sure that less than 7 GB is allocated
       assert mem_bytes < 7 * 10**9





"""simple docstring"""




from pathlib import PurePosixPath
from typing import Optional

import fsspec
from fsspec import AbstractFileSystem
from huggingface_hub.hf_api import DatasetInfo

from ..utils.file_utils import get_authentication_headers_for_url
from ..utils.hub import hf_hub_url
class 						__A							( SCREAMING_SNAKE_CASE_		):
     _UpperCamelCase							:			Union[str, Any]							=		""
     _UpperCamelCase							:			str							=		"hf-legacy"  # "hf://"" is reserved for hffs






     def __init__(		self				,     a__ = None				,     a__ = None				,     **a__				,     ):
       super().__init__(self				,     **a__  )
       _lowerCAmelCase						:       Any  			=     repo_info
       _lowerCAmelCase						:       Optional[Any]  			=     token
       _lowerCAmelCase						:       Optional[int]  			=     None






     def     __A  (		self  ):
       if self.dir_cache is None:
         _lowerCAmelCase						:       Optional[Any]  			=     {}
         for hf_file in self.repo_info.siblings:
           # TODO(QL): add sizes
           _lowerCAmelCase						:       Any  			=     {
               """name""": hf_file.rfilename,
               """size""": None,
               """type""": """file""",
           }
           self.dir_cache.update(
               {
                   str(a__  ): {"""name""": str(a__  ), """size""": None, """type""": """directory"""}
                   for d in list(PurePosixPath(hf_file.rfilename  ).parents  )[:-1]
               }  )






     def     __A  (		self				,     a__				,     a__ = "rb"				,     **a__				,     ):
       if not isinstance(self.repo_info				,     a__  ):
         raise NotImplementedError(F"Open is only implemented for dataset repositories, but got {self.repo_info}"  )
       _lowerCAmelCase						:       Tuple  			=     hf_hub_url(self.repo_info.id				,     a__				,     revision=self.repo_info.sha  )
       return fsspec.open(
           a__				,     mode=a__				,     headers=get_authentication_headers_for_url(a__				,     use_auth_token=self.token  )				,     client_kwargs={"""trust_env""": True}				,     ).open()






     def     __A  (		self				,     a__				,     **a__  ):
       self._get_dirs()
       _lowerCAmelCase						:       Union[str, Any]  			=     self._strip_protocol(a__  )
       if path in self.dir_cache:
         return self.dir_cache[path]
       else:
         raise FileNotFoundError(a__  )






     def     __A  (		self				,     a__				,     a__=False				,     **a__  ):
       self._get_dirs()
       _lowerCAmelCase						:       Any  			=     PurePosixPath(path.strip("""/"""  )  )
       _lowerCAmelCase						:       List[str]  			=     {}
       for p, f in self.dir_cache.items():
         _lowerCAmelCase						:       Any  			=     PurePosixPath(p.strip("""/"""  )  )
         _lowerCAmelCase						:       Optional[int]  			=     p.parent
         if root == path:
           _lowerCAmelCase						:       Dict  			=     f
       _lowerCAmelCase						:       Union[str, Any]  			=     list(paths.values()  )
       if detail:
         return out
       else:
         return sorted(f["""name"""] for f in out  )




