

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			:						List[Any]  					=     'hf-internal-testing/tiny-random-bert'
_lowerCAmelCase			:						Dict  					=     os.path.join(TRANSFORMERS_CACHE, 'models--hf-internal-testing--tiny-random-bert')
_lowerCAmelCase			:						Optional[Any]  					=     '9b8c223d42b2188cb49d29af482996f9d0f3e5a6'


class      lowerCAmelCase					(     unittest.TestCase					):


       '''simple docstring'''







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







        '''simple docstring'''

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

        # File is cached at the same place the second time.
        lowerCamelCase       =     cached_file(__UpperCamelCase    ,						__UpperCamelCase  )
        self.assertEqual(__UpperCamelCase    ,						__UpperCamelCase  )

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

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







        '''simple docstring'''

        with self.assertRaisesRegex(__UpperCamelCase    ,						'is not a valid model identifier'  ):
         lowerCamelCase       =     cached_file('tiny-random-bert'    ,						__UpperCamelCase  )

        with self.assertRaisesRegex(__UpperCamelCase    ,						'is not a valid git identifier'  ):
         lowerCamelCase       =     cached_file(__UpperCamelCase    ,						__UpperCamelCase    ,						revision='aaaa'  )

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

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







        '''simple docstring'''

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

        with open(os.path.join(__UpperCamelCase    ,						'refs'    ,						'main'  )  ) as f:
         lowerCamelCase       =     f.read()
        self.assertTrue(os.path.isfile(os.path.join(__UpperCamelCase    ,						'.no_exist'    ,						__UpperCamelCase    ,						'conf'  )  )  )

        lowerCamelCase       =     cached_file(__UpperCamelCase    ,						'conf'    ,						_raise_exceptions_for_missing_entries=__UpperCamelCase  )
        self.assertIsNone(__UpperCamelCase  )

        lowerCamelCase       =     cached_file(__UpperCamelCase    ,						'conf'    ,						local_files_only=__UpperCamelCase    ,						_raise_exceptions_for_missing_entries=__UpperCamelCase  )
        self.assertIsNone(__UpperCamelCase  )

        lowerCamelCase       =     mock.Mock()
        lowerCamelCase       =     500
        lowerCamelCase       =     {}
        lowerCamelCase       =     HTTPError
        lowerCamelCase       =     {}

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

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







        '''simple docstring'''

        self.assertTrue(has_file('hf-internal-testing/tiny-bert-pt-only'    ,						__UpperCamelCase  )  )
        self.assertFalse(has_file('hf-internal-testing/tiny-bert-pt-only'    ,						__UpperCamelCase  )  )
        self.assertFalse(has_file('hf-internal-testing/tiny-bert-pt-only'    ,						__UpperCamelCase  )  )

       def 	lowerCamelCase__    (						self       :	Any  ) ->			str:







        '''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(__UpperCamelCase    ,						'is not a valid model identifier'  ):
         get_file_from_repo('bert-base-case'    ,						__UpperCamelCase  )

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

        lowerCamelCase       =     get_file_from_repo('bert-base-cased'    ,						__UpperCamelCase  )
        # The name is the cached name which is not very easy to test, so instead we load the content.
        lowerCamelCase       =     json.loads(open(__UpperCamelCase    ,						'r'  ).read()  )
        self.assertEqual(config['hidden_size']    ,						768  )





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







        '''simple docstring'''

        with tempfile.TemporaryDirectory() as tmp_dir:
         lowerCamelCase       =     Path(__UpperCamelCase  ) / 'a.txt'
         filename.touch()
         self.assertEqual(get_file_from_repo(__UpperCamelCase    ,						'a.txt'  )    ,						str(__UpperCamelCase  )  )

         self.assertIsNone(get_file_from_repo(__UpperCamelCase    ,						'b.txt'  )  )






import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest





class    __A(					UpperCAmelCase							):
					SCREAMING_SNAKE_CASE           =     (KDPMaDiscreteScheduler,)
					SCREAMING_SNAKE_CASE           =     1_0

					def  lowercase__   (						self       :  str ,       **__UpperCamelCase       :  Optional[int] ):
												lowerCamelCase_					=       {
												    """num_train_timesteps""": 1_1_0_0,
												    """beta_start""": 0.0001,
												    """beta_end""": 0.02,
												    """beta_schedule""": """linear""",
												}

												config.update(**__UpperCamelCase )
												return config

					def  lowercase__   (						self       :  Optional[Any] ):
												for timesteps in [1_0, 5_0, 1_0_0, 1_0_0_0]:
																			self.check_over_configs(num_train_timesteps=__UpperCamelCase )

					def  lowercase__   (						self       :  Dict ):
												for beta_start, beta_end in zip([0.00001, 0.0001, 0.001] ,       [0.0002, 0.002, 0.02] ):
																			self.check_over_configs(beta_start=__UpperCamelCase ,       beta_end=__UpperCamelCase )

					def  lowercase__   (						self       :  Union[str, Any] ):
												for schedule in ["linear", "scaled_linear"]:
																			self.check_over_configs(beta_schedule=__UpperCamelCase )

					def  lowercase__   (						self       :  List[str] ):
												for prediction_type in ["epsilon", "v_prediction"]:
																			self.check_over_configs(prediction_type=__UpperCamelCase )

					def  lowercase__   (						self       :  Optional[int] ):
												lowerCamelCase_					=       self.scheduler_classes[0]
												lowerCamelCase_					=       self.get_scheduler_config(prediction_type="""v_prediction""" )
												lowerCamelCase_					=       scheduler_class(**__UpperCamelCase )

												scheduler.set_timesteps(self.num_inference_steps )

												lowerCamelCase_					=       self.dummy_model()
												lowerCamelCase_					=       self.dummy_sample_deter * scheduler.init_noise_sigma
												lowerCamelCase_					=       sample.to(__UpperCamelCase )

												for i, t in enumerate(scheduler.timesteps ):
																			lowerCamelCase_					=       scheduler.scale_model_input(__UpperCamelCase ,       __UpperCamelCase )

																			lowerCamelCase_					=       model(__UpperCamelCase ,       __UpperCamelCase )

																			lowerCamelCase_					=       scheduler.step(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase )
																			lowerCamelCase_					=       output.prev_sample

												lowerCamelCase_					=       torch.sum(torch.abs(__UpperCamelCase ) )
												lowerCamelCase_					=       torch.mean(torch.abs(__UpperCamelCase ) )

												if torch_device in ["cpu", "mps"]:
																			assert abs(result_sum.item() - 4.6934E-07 ) < 1E-2
																			assert abs(result_mean.item() - 6.1112E-10 ) < 1E-3
												else:
																			# CUDA
																			assert abs(result_sum.item() - 4.693_4286_5017_0972E-07 ) < 1E-2
																			assert abs(result_mean.item() - 0.0002 ) < 1E-3

					def  lowercase__   (						self       :  List[Any] ):
												if torch_device == "mps":
																			return
												lowerCamelCase_					=       self.scheduler_classes[0]
												lowerCamelCase_					=       self.get_scheduler_config()
												lowerCamelCase_					=       scheduler_class(**__UpperCamelCase )

												scheduler.set_timesteps(self.num_inference_steps )

												lowerCamelCase_					=       self.dummy_model()
												lowerCamelCase_					=       self.dummy_sample_deter * scheduler.init_noise_sigma
												lowerCamelCase_					=       sample.to(__UpperCamelCase )

												for i, t in enumerate(scheduler.timesteps ):
																			lowerCamelCase_					=       scheduler.scale_model_input(__UpperCamelCase ,       __UpperCamelCase )

																			lowerCamelCase_					=       model(__UpperCamelCase ,       __UpperCamelCase )

																			lowerCamelCase_					=       scheduler.step(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase )
																			lowerCamelCase_					=       output.prev_sample

												lowerCamelCase_					=       torch.sum(torch.abs(__UpperCamelCase ) )
												lowerCamelCase_					=       torch.mean(torch.abs(__UpperCamelCase ) )

												if torch_device in ["cpu", "mps"]:
																			assert abs(result_sum.item() - 20.4125 ) < 1E-2
																			assert abs(result_mean.item() - 0.0266 ) < 1E-3
												else:
																			# CUDA
																			assert abs(result_sum.item() - 20.4125 ) < 1E-2
																			assert abs(result_mean.item() - 0.0266 ) < 1E-3



					def  lowercase__   (						self       :  Optional[int] ):
												if torch_device == "mps":
																			return
												lowerCamelCase_					=       self.scheduler_classes[0]
												lowerCamelCase_					=       self.get_scheduler_config()
												lowerCamelCase_					=       scheduler_class(**__UpperCamelCase )

												scheduler.set_timesteps(self.num_inference_steps ,       device=__UpperCamelCase )

												lowerCamelCase_					=       self.dummy_model()
												lowerCamelCase_					=       self.dummy_sample_deter.to(__UpperCamelCase ) * scheduler.init_noise_sigma

												for t in scheduler.timesteps:
																			lowerCamelCase_					=       scheduler.scale_model_input(__UpperCamelCase ,       __UpperCamelCase )

																			lowerCamelCase_					=       model(__UpperCamelCase ,       __UpperCamelCase )

																			lowerCamelCase_					=       scheduler.step(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase )
																			lowerCamelCase_					=       output.prev_sample

												lowerCamelCase_					=       torch.sum(torch.abs(__UpperCamelCase ) )
												lowerCamelCase_					=       torch.mean(torch.abs(__UpperCamelCase ) )

												if str(__UpperCamelCase ).startswith("""cpu""" ):
																			# The following sum varies between 148 and 156 on mps. Why?
																			assert abs(result_sum.item() - 20.4125 ) < 1E-2
																			assert abs(result_mean.item() - 0.0266 ) < 1E-3
												else:
																			# CUDA
																			assert abs(result_sum.item() - 20.4125 ) < 1E-2
																			assert abs(result_mean.item() - 0.0266 ) < 1E-3





