



"""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,
)


lowerCAmelCase__              =							'''hf-internal-testing/tiny-random-bert'''
lowerCAmelCase__              =							os.path.join(TRANSFORMERS_CACHE, '''models--hf-internal-testing--tiny-random-bert''')
lowerCAmelCase__              =							'''9b8c223d42b2188cb49d29af482996f9d0f3e5a6'''






class 			__snake_case			(      unittest.TestCase):




  def 		SCREAMING_SNAKE_CASE							(					self	:		Any						):






         """simple docstring"""






         _lowerCamelCase		:							Optional[Any]            =					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:
                _lowerCamelCase		:							Optional[int]            =					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.
         _lowerCamelCase		:							Tuple            =					cached_file(__lowerCAmelCase		,   __lowerCAmelCase						)
         self.assertEqual(__lowerCAmelCase		,   __lowerCAmelCase						)

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




  def 		SCREAMING_SNAKE_CASE							(					self	:		Dict						):






         """simple docstring"""






         with self.assertRaisesRegex(__lowerCAmelCase		,   '''is not a valid model identifier'''						):
                _lowerCamelCase		:							Optional[int]            =					cached_file('''tiny-random-bert'''		,   __lowerCAmelCase						)

         with self.assertRaisesRegex(__lowerCAmelCase		,   '''is not a valid git identifier'''						):
                _lowerCamelCase		:							str            =					cached_file(__lowerCAmelCase		,   __lowerCAmelCase		,   revision='''aaaa'''						)

         with self.assertRaisesRegex(__lowerCAmelCase		,   '''does not appear to have a file named'''						):
                _lowerCamelCase		:							int            =					cached_file(__lowerCAmelCase		,   '''conf'''						)




  def 		SCREAMING_SNAKE_CASE							(					self	:		Optional[int]						):






         """simple docstring"""






         with self.assertRaisesRegex(__lowerCAmelCase		,   '''does not appear to have a file named'''						):
                _lowerCamelCase		:							Dict            =					cached_file(__lowerCAmelCase		,   '''conf'''						)

         with open(os.path.join(__lowerCAmelCase		,   '''refs'''		,   '''main'''						)						) as f:
                _lowerCamelCase		:							List[Any]            =					f.read()
         self.assertTrue(os.path.isfile(os.path.join(__lowerCAmelCase		,   '''.no_exist'''		,   __lowerCAmelCase		,   '''conf'''						)						)						)

         _lowerCamelCase		:							str            =					cached_file(__lowerCAmelCase		,   '''conf'''		,   _raise_exceptions_for_missing_entries=__lowerCAmelCase						)
         self.assertIsNone(__lowerCAmelCase						)

         _lowerCamelCase		:							Optional[int]            =					cached_file(__lowerCAmelCase		,   '''conf'''		,   local_files_only=__lowerCAmelCase		,   _raise_exceptions_for_missing_entries=__lowerCAmelCase						)
         self.assertIsNone(__lowerCAmelCase						)

         _lowerCamelCase		:							Any            =					mock.Mock()
         _lowerCamelCase		:							Optional[Any]            =					5_0_0
         _lowerCamelCase		:							Dict            =					{}
         _lowerCamelCase		:							List[Any]            =					HTTPError
         _lowerCamelCase		:							int            =					{}

         # 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:
                _lowerCamelCase		:							Union[str, Any]            =					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 		SCREAMING_SNAKE_CASE							(					self	:		Union[str, Any]						):






         """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 		SCREAMING_SNAKE_CASE							(					self	:		Union[str, Any]						):






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

         _lowerCamelCase		:							Dict            =					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.
         _lowerCamelCase		:							Dict            =					json.loads(open(__lowerCAmelCase		,   '''r'''						).read()						)
         self.assertEqual(config['''hidden_size''']		,   7_6_8						)




  def 		SCREAMING_SNAKE_CASE							(					self	:		Union[str, Any]						):






         """simple docstring"""






         with tempfile.TemporaryDirectory() as tmp_dir:
                _lowerCamelCase		:							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'''						)						)






import unittest

import torch

from diffusers import DDIMScheduler, DDPMScheduler, UNetaDModel
from diffusers.training_utils import set_seed
from diffusers.utils.testing_utils import slow


SCREAMING_SNAKE_CASE__     					=			False



class 						_UpperCAmelCase      (     unittest.TestCase							):






		def _snake_case		(						self				:     str   ,  UpperCAmelCase				:     Dict=32):
				set_seed(0)
				SCREAMING_SNAKE_CASE_				:Dict				     =						UNetaDModel(sample_size=UpperCAmelCase   ,  in_channels=3   ,  out_channels=3)
				SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						torch.optim.SGD(model.parameters()   ,  lr=0.0001)
				return model, optimizer







		@slow
		def _snake_case		(						self				:     str):
				SCREAMING_SNAKE_CASE_				:List[str]				     =						"cpu"  # ensure full determinism without setting the CUBLAS_WORKSPACE_CONFIG env variable
				SCREAMING_SNAKE_CASE_				:Any				     =						DDPMScheduler(
				    num_train_timesteps=10_00   ,  beta_start=0.0001   ,  beta_end=0.02   ,  beta_schedule="linear"   ,  clip_sample=UpperCAmelCase   ,  )
				SCREAMING_SNAKE_CASE_				:str				     =						DDIMScheduler(
				    num_train_timesteps=10_00   ,  beta_start=0.0001   ,  beta_end=0.02   ,  beta_schedule="linear"   ,  clip_sample=UpperCAmelCase   ,  )

				assert ddpm_scheduler.config.num_train_timesteps == ddim_scheduler.config.num_train_timesteps

				# shared batches for DDPM and DDIM
				set_seed(0)
				SCREAMING_SNAKE_CASE_				:Any				     =						[torch.randn((4, 3, 32, 32)).clip(-1   ,  1).to(UpperCAmelCase) for _ in range(4)]
				SCREAMING_SNAKE_CASE_				:str				     =						[torch.randn((4, 3, 32, 32)).to(UpperCAmelCase) for _ in range(4)]
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						[torch.randint(0   ,  10_00   ,  (4,)).long().to(UpperCAmelCase) for _ in range(4)]

				# train with a DDPM scheduler
				SCREAMING_SNAKE_CASE_     ,				SCREAMING_SNAKE_CASE_				:Tuple				     =						self.get_model_optimizer(resolution=32)
				model.train().to(UpperCAmelCase)
				for i in range(4):
						optimizer.zero_grad()
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						ddpm_scheduler.add_noise(clean_images[i]   ,  noise[i]   ,  timesteps[i])
						SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						model(UpperCAmelCase   ,  timesteps[i]).sample
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						torch.nn.functional.mse_loss(UpperCAmelCase   ,  noise[i])
						loss.backward()
						optimizer.step()
				del model, optimizer

				# recreate the model and optimizer, and retry with DDIM
				SCREAMING_SNAKE_CASE_     ,				SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						self.get_model_optimizer(resolution=32)
				model.train().to(UpperCAmelCase)
				for i in range(4):
						optimizer.zero_grad()
						SCREAMING_SNAKE_CASE_				:List[str]				     =						ddim_scheduler.add_noise(clean_images[i]   ,  noise[i]   ,  timesteps[i])
						SCREAMING_SNAKE_CASE_				:Tuple				     =						model(UpperCAmelCase   ,  timesteps[i]).sample
						SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						torch.nn.functional.mse_loss(UpperCAmelCase   ,  noise[i])
						loss.backward()
						optimizer.step()
				del model, optimizer

				self.assertTrue(torch.allclose(UpperCAmelCase   ,  UpperCAmelCase   ,  atol=1E-5))
				self.assertTrue(torch.allclose(UpperCAmelCase   ,  UpperCAmelCase   ,  atol=1E-5))






