




import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoImageProcessor, ViTImageProcessor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_image_processing import CustomImageProcessor  # noqa E402


_UpperCAmelCase		:  Tuple	=   get_tests_dir("fixtures")




class       __lowerCAmelCase				(							unittest.TestCase):






       def     SCREAMING_SNAKE_CASE			(      self:    int ):
         # A mock response for an HTTP head request to emulate server down
         lowercase			:Dict							=  mock.Mock()
         lowercase			:Any							=  5_00
         lowercase			:List[str]							=  {}
         lowercase			:int							=  HTTPError
         lowercase			:Union[str, Any]							=  {}

         # Download this model to make sure it's in the cache.
         lowercase			:List[str]							=  ViTImageProcessor.from_pretrained("hf-internal-testing/tiny-random-vit" )
         # Under the mock environment we get a 500 error when trying to reach the model.
         with mock.patch("requests.Session.request"				,    return_value=_lowerCAmelCase ) as mock_head:
           lowercase			:Tuple							=  ViTImageProcessor.from_pretrained("hf-internal-testing/tiny-random-vit" )
           # This check we did call the fake head request
           mock_head.assert_called()






       def     SCREAMING_SNAKE_CASE			(      self:    Any ):
         # This test is for deprecated behavior and can be removed in v5
         lowercase			:int							=  ViTImageProcessor.from_pretrained(
             "https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json" )






       def     SCREAMING_SNAKE_CASE			(      self:    str ):
         with self.assertRaises(_lowerCAmelCase ):
           # config is in subfolder, the following should not work without specifying the subfolder
           lowercase			:str							=  AutoImageProcessor.from_pretrained("hf-internal-testing/stable-diffusion-all-variants" )

         lowercase			:Dict							=  AutoImageProcessor.from_pretrained(
             "hf-internal-testing/stable-diffusion-all-variants"				,    subfolder="feature_extractor" )

         self.assertIsNotNone(_lowerCAmelCase )







@is_staging_test
class       __lowerCAmelCase				(							unittest.TestCase):






       @classmethod
       def     SCREAMING_SNAKE_CASE			(      cls:    Any ):
         lowercase			:Optional[int]							=  TOKEN
         HfFolder.save_token(_lowerCAmelCase )






       @classmethod
       def     SCREAMING_SNAKE_CASE			(      cls:    int ):
         try:
           delete_repo(token=cls._token				,    repo_id="test-image-processor" )
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token				,    repo_id="valid_org/test-image-processor-org" )
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token				,    repo_id="test-dynamic-image-processor" )
         except HTTPError:
           pass






       def     SCREAMING_SNAKE_CASE			(      self:    List[Any] ):
         lowercase			:Optional[int]							=  ViTImageProcessor.from_pretrained(_lowerCAmelCase )
         image_processor.push_to_hub("test-image-processor"				,    use_auth_token=self._token )

         lowercase			:str							=  ViTImageProcessor.from_pretrained(F"{USER}/test-image-processor" )
         for k, v in image_processor.__dict__.items():
           self.assertEqual(_lowerCAmelCase				,    getattr(_lowerCAmelCase				,    _lowerCAmelCase ) )

         # Reset repo
         delete_repo(token=self._token				,    repo_id="test-image-processor" )

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           image_processor.save_pretrained(
               _lowerCAmelCase				,    repo_id="test-image-processor"				,    push_to_hub=_lowerCAmelCase				,    use_auth_token=self._token )

         lowercase			:List[Any]							=  ViTImageProcessor.from_pretrained(F"{USER}/test-image-processor" )
         for k, v in image_processor.__dict__.items():
           self.assertEqual(_lowerCAmelCase				,    getattr(_lowerCAmelCase				,    _lowerCAmelCase ) )






       def     SCREAMING_SNAKE_CASE			(      self:    Dict ):
         lowercase			:Tuple							=  ViTImageProcessor.from_pretrained(_lowerCAmelCase )
         image_processor.push_to_hub("valid_org/test-image-processor"				,    use_auth_token=self._token )

         lowercase			:int							=  ViTImageProcessor.from_pretrained("valid_org/test-image-processor" )
         for k, v in image_processor.__dict__.items():
           self.assertEqual(_lowerCAmelCase				,    getattr(_lowerCAmelCase				,    _lowerCAmelCase ) )

         # Reset repo
         delete_repo(token=self._token				,    repo_id="valid_org/test-image-processor" )

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           image_processor.save_pretrained(
               _lowerCAmelCase				,    repo_id="valid_org/test-image-processor-org"				,    push_to_hub=_lowerCAmelCase				,    use_auth_token=self._token )

         lowercase			:Optional[int]							=  ViTImageProcessor.from_pretrained("valid_org/test-image-processor-org" )
         for k, v in image_processor.__dict__.items():
           self.assertEqual(_lowerCAmelCase				,    getattr(_lowerCAmelCase				,    _lowerCAmelCase ) )






       def     SCREAMING_SNAKE_CASE			(      self:    Dict ):
         CustomImageProcessor.register_for_auto_class()
         lowercase			:Optional[int]							=  CustomImageProcessor.from_pretrained(_lowerCAmelCase )

         image_processor.push_to_hub("test-dynamic-image-processor"				,    use_auth_token=self._token )

         # This has added the proper auto_map field to the config
         self.assertDictEqual(
             image_processor.auto_map				,    {"AutoImageProcessor": "custom_image_processing.CustomImageProcessor"}				,    )

         lowercase			:List[str]							=  AutoImageProcessor.from_pretrained(
             F"{USER}/test-dynamic-image-processor"				,    trust_remote_code=_lowerCAmelCase )
         # Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
         self.assertEqual(new_image_processor.__class__.__name__				,    "CustomImageProcessor" )









from typing import Union

import fire
import torch
from tqdm import tqdm



def 					UpperCAmelCase__     (   lowerCamelCase,	lowerCamelCase = "cpu",	lowerCamelCase = None      ):
  lowercase			:Optional[Any]							=  torch.load(lowerCamelCase,	map_location=lowerCamelCase      )
  for k, v in tqdm(state_dict.items()      ):
    if not isinstance(lowerCamelCase,	torch.Tensor      ):
      raise TypeError("FP16 conversion only works on paths that are saved state dicts, like pytorch_model.bin"      )
    lowercase			:List[Any]							=  v.half()
  if save_path is None:  # overwrite src_path
    lowercase			:Optional[Any]							=  src_path
  torch.save(lowerCamelCase,	lowerCamelCase      )


if __name__ == "__main__":
     fire.Fire(convert)



