




import argparse
import json
import os

import torch
from transformers.file_utils import has_file

from diffusers import UNetaDConditionModel, UNetaDModel


lowerCamelCase__					:  Union[str, Any]      							=				False
lowerCamelCase__					:  List[str]      							=				True
lowerCamelCase__					:  Optional[Any]      							=				False


if __name__ == "__main__":
   lowerCamelCase__					:  Optional[Any]      							=				argparse.ArgumentParser()

   parser.add_argument(
       '--repo_path',
       default=None,
       type=str,
       required=True,
       help='The config json file corresponding to the architecture.',
   )

   parser.add_argument('--dump_path', default=None, type=str, required=True, help='Path to the output model.')

   lowerCamelCase__					:  Optional[Any]      							=				parser.parse_args()

   lowerCamelCase__					:  List[Any]      							=				{
       'image_size': 'sample_size',
       'num_res_blocks': 'layers_per_block',
       'block_channels': 'block_out_channels',
       'down_blocks': 'down_block_types',
       'up_blocks': 'up_block_types',
       'downscale_freq_shift': 'freq_shift',
       'resnet_num_groups': 'norm_num_groups',
       'resnet_act_fn': 'act_fn',
       'resnet_eps': 'norm_eps',
       'num_head_channels': 'attention_head_dim',
   }

   lowerCamelCase__					:  str      							=				{
       'time_steps': 'time_proj',
       'mid': 'mid_block',
       'downsample_blocks': 'down_blocks',
       'upsample_blocks': 'up_blocks',
   }

   lowerCamelCase__					:  str      							=				'' if has_file(args.repo_path, 'config.json') else 'unet'

   with open(os.path.join(args.repo_path, subfolder, 'config.json'), 'r', encoding='utf-8') as reader:
      lowerCamelCase__					:  Optional[int]      							=				reader.read()
      lowerCamelCase__					:  Optional[int]      							=				json.loads(text)

   if do_only_config:
      for key in config_parameters_to_change.keys():
         config.pop(key, None)

   if has_file(args.repo_path, 'config.json'):
      lowerCamelCase__					:  Any      							=				UNetaDModel(**config)
   else:
      lowerCamelCase__					:  Optional[Any]      							=				UNetaDConditionModel if 'ldm-text2im-large-256' in args.repo_path else UNetaDModel
      lowerCamelCase__					:  Optional[Any]      							=				class_name(**config)

   if do_only_config:
      model.save_config(os.path.join(args.repo_path, subfolder))

   lowerCamelCase__					:  Optional[Any]      							=				dict(model.config)

   if do_only_renaming:
      for key, value in config_parameters_to_change.items():
         if key in config:
            lowerCamelCase__					:  List[str]      							=				config[key]
            del config[key]

      lowerCamelCase__					:  str      							=				[k.replace('UNetRes', '') for k in config['down_block_types']]
      lowerCamelCase__					:  List[str]      							=				[k.replace('UNetRes', '') for k in config['up_block_types']]

   if do_only_weights:
      lowerCamelCase__					:  Optional[Any]      							=				torch.load(os.path.join(args.repo_path, subfolder, 'diffusion_pytorch_model.bin'))

      lowerCamelCase__					:  Dict      							=				{}
      for param_key, param_value in state_dict.items():
         if param_key.endswith('.op.bias') or param_key.endswith('.op.weight'):
            continue
         lowerCamelCase__					:  int      							=				False
         for key, new_key in key_parameters_to_change.items():
            if not has_changed and param_key.split('.')[0] == key:
               lowerCamelCase__					:  Dict      							=				param_value
               lowerCamelCase__					:  str      							=				True
         if not has_changed:
            lowerCamelCase__					:  Dict      							=				param_value

      model.load_state_dict(new_state_dict)
      model.save_pretrained(os.path.join(args.repo_path, subfolder))





import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch





class 			lowerCamelCase_				( _SCREAMING_SNAKE_CASE   ):


 '''simple docstring'''







 @require_torch
 def  lowerCAmelCase_    (				self    :						int		):
       # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
       # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
       # while running an external program

       # python one-liner segments

       # this must be loaded before socket.socket is monkey-patched
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

       # Force fetching the files so that we can use the cache
       SCREAMING_SNAKE_CASE_      		=     'hf-internal-testing/tiny-random-bert'
       BertConfig.from_pretrained(_lowerCAmelCase		)
       BertModel.from_pretrained(_lowerCAmelCase		)
       BertTokenizer.from_pretrained(_lowerCAmelCase		)
       pipeline(task='fill-mask'							,							model=_lowerCAmelCase		)

       # baseline - just load from_pretrained with normal network
       SCREAMING_SNAKE_CASE_      		=     [sys.executable, '-c', '\n'.join([load, run, mock]		)]

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							0							,							result.stderr		)
       self.assertIn('success'							,							result.stdout.decode()		)



 @require_torch
 def  lowerCAmelCase_    (				self    :						Tuple		):
       # python one-liner segments
       # this must be loaded before socket.socket is monkey-patched
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

       # Force fetching the files so that we can use the cache
       SCREAMING_SNAKE_CASE_      		=     'hf-internal-testing/tiny-random-bert'
       BertConfig.from_pretrained(_lowerCAmelCase		)
       BertModel.from_pretrained(_lowerCAmelCase		)
       BertTokenizer.from_pretrained(_lowerCAmelCase		)
       pipeline(task='fill-mask'							,							model=_lowerCAmelCase		)

       # baseline - just load from_pretrained with normal network
       SCREAMING_SNAKE_CASE_      		=     [sys.executable, '-c', '\n'.join([load, run, mock]		)]

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							0							,							result.stderr		)
       self.assertIn('success'							,							result.stdout.decode()		)



 @require_torch
 def  lowerCAmelCase_    (				self    :						List[str]		):
       # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
       # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
       # while running an external program

       # python one-liner segments

       # this must be loaded before socket.socket is monkey-patched
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

       # baseline - just load from_pretrained with normal network
       SCREAMING_SNAKE_CASE_      		=     [sys.executable, '-c', '\n'.join([load, run]		)]

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							0							,							result.stderr		)
       self.assertIn('success'							,							result.stdout.decode()		)

       # next emulate no network
       SCREAMING_SNAKE_CASE_      		=     [sys.executable, '-c', '\n'.join([load, mock, run]		)]

       # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
       # env["TRANSFORMERS_OFFLINE"] = "0"
       # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
       # self.assertEqual(result.returncode, 1, result.stderr)

       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							0							,							result.stderr		)
       self.assertIn('success'							,							result.stdout.decode()		)



 @require_torch
 def  lowerCAmelCase_    (				self    :						Any		):
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import pipeline\n        '
       SCREAMING_SNAKE_CASE_      		=     '\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

       SCREAMING_SNAKE_CASE_      		=     '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_SNAKE_CASE_      		=     [sys.executable, '-c', '\n'.join([load, mock, run]		)]
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							1							,							result.stderr		)
       self.assertIn(
           'You cannot infer task automatically within `pipeline` when using offline mode'							,							result.stderr.decode().replace('\n'							,							''		)							,							)




 @require_torch
 def  lowerCAmelCase_    (				self    :						Optional[int]		):
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import AutoModel\n        '
       SCREAMING_SNAKE_CASE_      		=     '\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

       # baseline - just load from_pretrained with normal network
       SCREAMING_SNAKE_CASE_      		=     [sys.executable, '-c', '\n'.join([load, run]		)]

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							0							,							result.stderr		)
       self.assertIn('success'							,							result.stdout.decode()		)

       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_SNAKE_CASE_      		=     subprocess.run(_lowerCAmelCase							,							env=_lowerCAmelCase							,							check=_lowerCAmelCase							,							capture_output=_lowerCAmelCase		)
       self.assertEqual(result.returncode							,							0							,							result.stderr		)
       self.assertIn('success'							,							result.stdout.decode()		)