

'''simple docstring'''



import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


__snake_case			:     Optional[Any]										=      'hf-internal-testing/tiny-random-bert'
__snake_case			:     str										=      os.path.join(TRANSFORMERS_CACHE, 'models--hf-internal-testing--tiny-random-bert')
__snake_case			:     int										=      '9b8c223d42b2188cb49d29af482996f9d0f3e5a6'



class lowerCamelCase			(  unittest.TestCase						):



			'''simple docstring'''






			def lowercase__						(				self	:						Any			)      ->		Optional[Any]:





									'''simple docstring'''
									A__					:       Dict     =cached_file(lowerCAmelCase_		,      lowerCAmelCase_			)
									# Should have downloaded the file in here
									self.assertTrue(os.path.isdir(lowerCAmelCase_			)			)
									# Cache should contain at least those three subfolders:
									for subfolder in ["blobs", "refs", "snapshots"]:
															self.assertTrue(os.path.isdir(os.path.join(lowerCAmelCase_		,      lowerCAmelCase_			)			)			)
									with open(os.path.join(lowerCAmelCase_		,      """refs"""		,      """main"""			)			) as f:
															A__					:       str     =f.read()
									self.assertEqual(lowerCAmelCase_		,      os.path.join(lowerCAmelCase_		,      """snapshots"""		,      lowerCAmelCase_		,      lowerCAmelCase_			)			)
									self.assertTrue(os.path.isfile(lowerCAmelCase_			)			)

									# File is cached at the same place the second time.
									A__					:       Dict     =cached_file(lowerCAmelCase_		,      lowerCAmelCase_			)
									self.assertEqual(lowerCAmelCase_		,      lowerCAmelCase_			)

									# Using a specific revision to test the full commit hash.
									A__					:       Dict     =cached_file(lowerCAmelCase_		,      lowerCAmelCase_		,      revision="""9b8c223"""			)
									self.assertEqual(lowerCAmelCase_		,      os.path.join(lowerCAmelCase_		,      """snapshots"""		,      lowerCAmelCase_		,      lowerCAmelCase_			)			)


			def lowercase__						(				self	:						Optional[int]			)      ->		Optional[int]:





									'''simple docstring'''
									with self.assertRaisesRegex(lowerCAmelCase_		,      """is not a valid model identifier"""			):
															A__					:       Union[str, Any]     =cached_file("""tiny-random-bert"""		,      lowerCAmelCase_			)

									with self.assertRaisesRegex(lowerCAmelCase_		,      """is not a valid git identifier"""			):
															A__					:       List[Any]     =cached_file(lowerCAmelCase_		,      lowerCAmelCase_		,      revision="""aaaa"""			)

									with self.assertRaisesRegex(lowerCAmelCase_		,      """does not appear to have a file named"""			):
															A__					:       Any     =cached_file(lowerCAmelCase_		,      """conf"""			)


			def lowercase__						(				self	:						str			)      ->		Dict:





									'''simple docstring'''
									with self.assertRaisesRegex(lowerCAmelCase_		,      """does not appear to have a file named"""			):
															A__					:       List[Any]     =cached_file(lowerCAmelCase_		,      """conf"""			)

									with open(os.path.join(lowerCAmelCase_		,      """refs"""		,      """main"""			)			) as f:
															A__					:       Tuple     =f.read()
									self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase_		,      """.no_exist"""		,      lowerCAmelCase_		,      """conf"""			)			)			)

									A__					:       str     =cached_file(lowerCAmelCase_		,      """conf"""		,      _raise_exceptions_for_missing_entries=lowerCAmelCase_			)
									self.assertIsNone(lowerCAmelCase_			)

									A__					:       Any     =cached_file(lowerCAmelCase_		,      """conf"""		,      local_files_only=lowerCAmelCase_		,      _raise_exceptions_for_missing_entries=lowerCAmelCase_			)
									self.assertIsNone(lowerCAmelCase_			)

									A__					:       List[Any]     =mock.Mock()
									A__					:       str     =5_00
									A__					:       str     ={}
									A__					:       Union[str, Any]     =HTTPError
									A__					:       List[Any]     ={}

									# Under the mock environment we get a 500 error when trying to reach the tokenizer.
									with mock.patch("""requests.Session.request"""		,      return_value=lowerCAmelCase_			) as mock_head:
															A__					:       Optional[int]     =cached_file(lowerCAmelCase_		,      """conf"""		,      _raise_exceptions_for_connection_errors=lowerCAmelCase_			)
															self.assertIsNone(lowerCAmelCase_			)
															# This check we did call the fake head request
															mock_head.assert_called()


			def lowercase__						(				self	:						Union[str, Any]			)      ->		Optional[int]:





									'''simple docstring'''
									self.assertTrue(has_file("""hf-internal-testing/tiny-bert-pt-only"""		,      lowerCAmelCase_			)			)
									self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""		,      lowerCAmelCase_			)			)
									self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""		,      lowerCAmelCase_			)			)


			def lowercase__						(				self	:						List[Any]			)      ->		int:





									'''simple docstring'''
									self.assertIsNone(get_file_from_repo("""bert-base-cased"""		,      """ahah.txt"""			)			)

									# The function raises if the repository does not exist.
									with self.assertRaisesRegex(lowerCAmelCase_		,      """is not a valid model identifier"""			):
															get_file_from_repo("""bert-base-case"""		,      lowerCAmelCase_			)

									# The function raises if the revision does not exist.
									with self.assertRaisesRegex(lowerCAmelCase_		,      """is not a valid git identifier"""			):
															get_file_from_repo("""bert-base-cased"""		,      lowerCAmelCase_		,      revision="""ahaha"""			)

									A__					:       Union[str, Any]     =get_file_from_repo("""bert-base-cased"""		,      lowerCAmelCase_			)
									# The name is the cached name which is not very easy to test, so instead we load the content.
									A__					:       List[str]     =json.loads(open(lowerCAmelCase_		,      """r"""			).read()			)
									self.assertEqual(config["""hidden_size"""]		,      7_68			)



			def lowercase__						(				self	:						List[str]			)      ->		List[Any]:





									'''simple docstring'''
									with tempfile.TemporaryDirectory() as tmp_dir:
															A__					:       Optional[Any]     =Path(lowerCAmelCase_			) / """a.txt"""
															filename.touch()
															self.assertEqual(get_file_from_repo(lowerCAmelCase_		,      """a.txt"""			)		,      str(lowerCAmelCase_			)			)

															self.assertIsNone(get_file_from_repo(lowerCAmelCase_		,      """b.txt"""			)			)




'''simple docstring'''



import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class lowerCamelCase			:



			'''simple docstring'''






			def lowercase__						(				self	:						Any			)      ->		Tuple:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       str     =TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       int     =AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       Union[str, Any]     =UNetaDConditionModel(
									    sample_size=32		,      layers_per_block=1		,      block_out_channels=[32, 64]		,      down_block_types=[
									        """ResnetDownsampleBlock2D""",
									        """SimpleCrossAttnDownBlock2D""",
									    ]		,      mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""		,      up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]		,      in_channels=3		,      out_channels=6		,      cross_attention_dim=32		,      encoder_hid_dim=32		,      attention_head_dim=8		,      addition_embed_type="""text"""		,      addition_embed_type_num_heads=2		,      cross_attention_norm="""group_norm"""		,      resnet_time_scale_shift="""scale_shift"""		,      act_fn="""gelu"""		,      )
									unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									torch.manual_seed(0			)
									A__					:       Dict     =DDPMScheduler(
									    num_train_timesteps=10_00		,      beta_schedule="""squaredcos_cap_v2"""		,      beta_start=0.0001		,      beta_end=0.02		,      thresholding=lowerCAmelCase_		,      dynamic_thresholding_ratio=0.95		,      sample_max_value=1.0		,      prediction_type="""epsilon"""		,      variance_type="""learned_range"""		,      )

									torch.manual_seed(0			)
									A__					:       Union[str, Any]     =IFWatermarker()

									return {
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "unet": unet,
									    "scheduler": scheduler,
									    "watermarker": watermarker,
									    "safety_checker": None,
									    "feature_extractor": None,
									}


			def lowercase__						(				self	:						Union[str, Any]			)      ->		int:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       List[Any]     =TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       str     =AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       Dict     =UNetaDConditionModel(
									    sample_size=32		,      layers_per_block=[1, 2]		,      block_out_channels=[32, 64]		,      down_block_types=[
									        """ResnetDownsampleBlock2D""",
									        """SimpleCrossAttnDownBlock2D""",
									    ]		,      mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""		,      up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]		,      in_channels=6		,      out_channels=6		,      cross_attention_dim=32		,      encoder_hid_dim=32		,      attention_head_dim=8		,      addition_embed_type="""text"""		,      addition_embed_type_num_heads=2		,      cross_attention_norm="""group_norm"""		,      resnet_time_scale_shift="""scale_shift"""		,      act_fn="""gelu"""		,      class_embed_type="""timestep"""		,      mid_block_scale_factor=1.414		,      time_embedding_act_fn="""gelu"""		,      time_embedding_dim=32		,      )
									unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									torch.manual_seed(0			)
									A__					:       Optional[int]     =DDPMScheduler(
									    num_train_timesteps=10_00		,      beta_schedule="""squaredcos_cap_v2"""		,      beta_start=0.0001		,      beta_end=0.02		,      thresholding=lowerCAmelCase_		,      dynamic_thresholding_ratio=0.95		,      sample_max_value=1.0		,      prediction_type="""epsilon"""		,      variance_type="""learned_range"""		,      )

									torch.manual_seed(0			)
									A__					:       int     =DDPMScheduler(
									    num_train_timesteps=10_00		,      beta_schedule="""squaredcos_cap_v2"""		,      beta_start=0.0001		,      beta_end=0.02		,      )

									torch.manual_seed(0			)
									A__					:       List[str]     =IFWatermarker()

									return {
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "unet": unet,
									    "scheduler": scheduler,
									    "image_noising_scheduler": image_noising_scheduler,
									    "watermarker": watermarker,
									    "safety_checker": None,
									    "feature_extractor": None,
									}


			def lowercase__						(				self	:						str			)      ->		Tuple:





									'''simple docstring'''
									A__					:       Tuple     =self.get_dummy_components()
									A__					:       str     =self.pipeline_class(**lowerCAmelCase_			)
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       Optional[int]     =self.get_dummy_inputs(lowerCAmelCase_			)

									A__					:       str     =inputs["""prompt"""]
									A__					:       Optional[int]     =inputs["""generator"""]
									A__					:       Optional[Any]     =inputs["""num_inference_steps"""]
									A__					:       Union[str, Any]     =inputs["""output_type"""]

									if "image" in inputs:
															A__					:       Union[str, Any]     =inputs["""image"""]
									else:
															A__					:       List[Any]     =None

									if "mask_image" in inputs:
															A__					:       Union[str, Any]     =inputs["""mask_image"""]
									else:
															A__					:       Tuple     =None

									if "original_image" in inputs:
															A__					:       Optional[Any]     =inputs["""original_image"""]
									else:
															A__					:       Tuple     =None

									A__   ,						A__					:       Optional[Any]     =pipe.encode_prompt(lowerCAmelCase_			)

									# inputs with prompt converted to embeddings
									A__					:       Optional[int]     ={
									    """prompt_embeds""": prompt_embeds,
									    """negative_prompt_embeds""": negative_prompt_embeds,
									    """generator""": generator,
									    """num_inference_steps""": num_inference_steps,
									    """output_type""": output_type,
									}

									if image is not None:
															A__					:       int     =image

									if mask_image is not None:
															A__					:       Tuple     =mask_image

									if original_image is not None:
															A__					:       Optional[int]     =original_image

									# set all optional components to None
									for optional_component in pipe._optional_components:
															setattr(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_			)

									A__					:       Any     =pipe(**lowerCAmelCase_			)[0]

									with tempfile.TemporaryDirectory() as tmpdir:
															pipe.save_pretrained(lowerCAmelCase_			)
															A__					:       int     =self.pipeline_class.from_pretrained(lowerCAmelCase_			)
															pipe_loaded.to(lowerCAmelCase_			)
															pipe_loaded.set_progress_bar_config(disable=lowerCAmelCase_			)

									pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									for optional_component in pipe._optional_components:
															self.assertTrue(
															    getattr(lowerCAmelCase_		,      lowerCAmelCase_			) is None		,      f"`{optional_component}` did not stay set to None after loading."		,      )

									A__					:       Dict     =self.get_dummy_inputs(lowerCAmelCase_			)

									A__					:       int     =inputs["""generator"""]
									A__					:       str     =inputs["""num_inference_steps"""]
									A__					:       Optional[Any]     =inputs["""output_type"""]

									# inputs with prompt converted to embeddings
									A__					:       List[Any]     ={
									    """prompt_embeds""": prompt_embeds,
									    """negative_prompt_embeds""": negative_prompt_embeds,
									    """generator""": generator,
									    """num_inference_steps""": num_inference_steps,
									    """output_type""": output_type,
									}

									if image is not None:
															A__					:       int     =image

									if mask_image is not None:
															A__					:       int     =mask_image

									if original_image is not None:
															A__					:       Optional[int]     =original_image

									A__					:       List[str]     =pipe_loaded(**lowerCAmelCase_			)[0]

									A__					:       Union[str, Any]     =np.abs(to_np(lowerCAmelCase_			) - to_np(lowerCAmelCase_			)			).max()
									self.assertLess(lowerCAmelCase_		,      1e-4			)



			def lowercase__						(				self	:						List[str]			)      ->		Dict:





									'''simple docstring'''
									A__					:       Union[str, Any]     =self.get_dummy_components()
									A__					:       int     =self.pipeline_class(**lowerCAmelCase_			)
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       int     =self.get_dummy_inputs(lowerCAmelCase_			)
									A__					:       List[Any]     =pipe(**lowerCAmelCase_			)[0]

									with tempfile.TemporaryDirectory() as tmpdir:
															pipe.save_pretrained(lowerCAmelCase_			)
															A__					:       List[Any]     =self.pipeline_class.from_pretrained(lowerCAmelCase_			)
															pipe_loaded.to(lowerCAmelCase_			)
															pipe_loaded.set_progress_bar_config(disable=lowerCAmelCase_			)

									pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									A__					:       int     =self.get_dummy_inputs(lowerCAmelCase_			)
									A__					:       Tuple     =pipe_loaded(**lowerCAmelCase_			)[0]

									A__					:       Tuple     =np.abs(to_np(lowerCAmelCase_			) - to_np(lowerCAmelCase_			)			).max()
									self.assertLess(lowerCAmelCase_		,      1e-4			)

