


"""simple docstring"""







import subprocess
import sys

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


class 				SCREAMING_SNAKE_CASE__    (		lowercase    ):
       """simple docstring"""


       @require_torch
       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					List[Any]								=							"\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

         lowerCAmelCase					:					Optional[Any]								=							"\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        "

         lowerCAmelCase					:					str								=							"\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
         lowerCAmelCase					:					Union[str, Any]								=							"hf-internal-testing/tiny-random-bert"
         BertConfig.from_pretrained(snake_case__						)
         BertModel.from_pretrained(snake_case__						)
         BertTokenizer.from_pretrained(snake_case__						)
         pipeline(task="fill-mask"		,					model=snake_case__						)

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

         # should succeed
         lowerCAmelCase					:					List[Any]								=							self.get_env()
         # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
         lowerCAmelCase					:					List[Any]								=							"1"
         lowerCAmelCase					:					Optional[Any]								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         self.assertEqual(result.returncode		,					0		,					result.stderr						)
         self.assertIn("success"		,					result.stdout.decode()						)

       @require_torch
       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[Any]								=							"\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

         lowerCAmelCase					:					List[str]								=							"\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        "

         lowerCAmelCase					:					Any								=							"\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
         lowerCAmelCase					:					Any								=							"hf-internal-testing/tiny-random-bert"
         BertConfig.from_pretrained(snake_case__						)
         BertModel.from_pretrained(snake_case__						)
         BertTokenizer.from_pretrained(snake_case__						)
         pipeline(task="fill-mask"		,					model=snake_case__						)

         # baseline - just load from_pretrained with normal network
         lowerCAmelCase					:					Optional[Any]								=							[sys.executable, "-c", "\n".join([load, run, mock]						)]

         # should succeed
         lowerCAmelCase					:					List[Any]								=							self.get_env()
         lowerCAmelCase					:					Any								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         self.assertEqual(result.returncode		,					0		,					result.stderr						)
         self.assertIn("success"		,					result.stdout.decode()						)

       @require_torch
       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[int]								=							"\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        "

         lowerCAmelCase					:					Union[str, Any]								=							"\nmname = \"hf-internal-testing/tiny-random-bert-sharded\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint(\"success\")\n        "

         lowerCAmelCase					:					List[str]								=							"\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
         lowerCAmelCase					:					Optional[int]								=							[sys.executable, "-c", "\n".join([load, run]						)]

         # should succeed
         lowerCAmelCase					:					Optional[Any]								=							self.get_env()
         lowerCAmelCase					:					str								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         self.assertEqual(result.returncode		,					0		,					result.stderr						)
         self.assertIn("success"		,					result.stdout.decode()						)

         # next emulate no network
         lowerCAmelCase					:					int								=							[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
         lowerCAmelCase					:					List[str]								=							"1"
         lowerCAmelCase					:					List[Any]								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         self.assertEqual(result.returncode		,					0		,					result.stderr						)
         self.assertIn("success"		,					result.stdout.decode()						)

       @require_torch
       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Union[str, Any]								=							"\nfrom transformers import pipeline\n        "
         lowerCAmelCase					:					Any								=							"\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        "

         lowerCAmelCase					:					str								=							"\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        "
         lowerCAmelCase					:					Dict								=							self.get_env()
         lowerCAmelCase					:					int								=							"1"
         lowerCAmelCase					:					Dict								=							[sys.executable, "-c", "\n".join([load, mock, run]						)]
         lowerCAmelCase					:					Optional[Any]								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         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 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					str								=							"\nfrom transformers import AutoModel\n        "
         lowerCAmelCase					:					str								=							"\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
         lowerCAmelCase					:					Union[str, Any]								=							[sys.executable, "-c", "\n".join([load, run]						)]

         # should succeed
         lowerCAmelCase					:					Optional[int]								=							self.get_env()
         lowerCAmelCase					:					List[Any]								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         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
         lowerCAmelCase					:					List[Any]								=							"1"
         lowerCAmelCase					:					Any								=							subprocess.run(snake_case__		,					env=snake_case__		,					check=snake_case__		,					capture_output=snake_case__						)
         self.assertEqual(result.returncode		,					0		,					result.stderr						)
         self.assertIn("success"		,					result.stdout.decode()						)









"""simple docstring"""







import argparse

from diffusers.pipelines.stable_diffusion.convert_from_ckpt import download_controlnet_from_original_ckpt


if __name__ == "__main__":
     lowerCAmelCase__				    =       argparse.ArgumentParser()

     parser.add_argument(
         '''--checkpoint_path''', default=None, type=str, required=True, help='''Path to the checkpoint to convert.'''
     )
     parser.add_argument(
         '''--original_config_file''',
         type=str,
         required=True,
         help='''The YAML config file corresponding to the original architecture.''',
     )
     parser.add_argument(
         '''--num_in_channels''',
         default=None,
         type=int,
         help='''The number of input channels. If `None` number of input channels will be automatically inferred.''',
     )
     parser.add_argument(
         '''--image_size''',
         default=512,
         type=int,
         help=(
             '''The image size that the model was trained on. Use 512 for Stable Diffusion v1.X and Stable Siffusion v2'''
             ''' Base. Use 768 for Stable Diffusion v2.'''
         ),
     )
     parser.add_argument(
         '''--extract_ema''',
         action='''store_true''',
         help=(
             '''Only relevant for checkpoints that have both EMA and non-EMA weights. Whether to extract the EMA weights'''
             ''' or not. Defaults to `False`. Add `--extract_ema` to extract the EMA weights. EMA weights usually yield'''
             ''' higher quality images for inference. Non-EMA weights are usually better to continue fine-tuning.'''
         ),
     )
     parser.add_argument(
         '''--upcast_attention''',
         action='''store_true''',
         help=(
             '''Whether the attention computation should always be upcasted. This is necessary when running stable'''
             ''' diffusion 2.1.'''
         ),
     )
     parser.add_argument(
         '''--from_safetensors''',
         action='''store_true''',
         help='''If `--checkpoint_path` is in `safetensors` format, load checkpoint with safetensors instead of PyTorch.''',
     )
     parser.add_argument(
         '''--to_safetensors''',
         action='''store_true''',
         help='''Whether to store pipeline in safetensors format or not.''',
     )
     parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')
     parser.add_argument('''--device''', type=str, help='''Device to use (e.g. cpu, cuda:0, cuda:1, etc.)''')
     def  a__      (    SCREAMING_SNAKE_CASE   :   Tuple    ):


       '''simple docstring'''







       if string == "True":
         return True
       elif string == "False":
         return False
       else:
         raise ValueError(f"""could not parse string as bool {string}"""    )

     parser.add_argument(
         '''--use_linear_projection''', help='''Override for use linear projection''', required=False, type=parse_bool
     )

     parser.add_argument('''--cross_attention_dim''', help='''Override for cross attention_dim''', required=False, type=int)

     lowerCAmelCase__				    =       parser.parse_args()

     lowerCAmelCase__				    =       download_controlnet_from_original_ckpt(
         checkpoint_path=args.checkpoint_path,
         original_config_file=args.original_config_file,
         image_size=args.image_size,
         extract_ema=args.extract_ema,
         num_in_channels=args.num_in_channels,
         upcast_attention=args.upcast_attention,
         from_safetensors=args.from_safetensors,
         device=args.device,
         use_linear_projection=args.use_linear_projection,
         cross_attention_dim=args.cross_attention_dim,
     )





     controlnet.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)





