





'''simple docstring'''




import subprocess
import sys

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







class   A_	(  lowerCAmelCase_			):




 @require_torch
 def      lowercase	(					self       :    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
      _UpperCAmelCase     							=   "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      _UpperCAmelCase     							=   "\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        "

      _UpperCAmelCase     							=   "\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
      _UpperCAmelCase     							=   "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
      _UpperCAmelCase     							=   [sys.executable, "-c", "\n".join([load, run, mock]    )]

      # should succeed
      _UpperCAmelCase     							=   self.get_env()
      # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
      _UpperCAmelCase     							=   "1"
      _UpperCAmelCase     							=   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       :    List[Any]    ):
      # python one-liner segments
      # this must be loaded before socket.socket is monkey-patched
      _UpperCAmelCase     							=   "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      _UpperCAmelCase     							=   "\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        "

      _UpperCAmelCase     							=   "\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
      _UpperCAmelCase     							=   "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
      _UpperCAmelCase     							=   [sys.executable, "-c", "\n".join([load, run, mock]    )]

      # should succeed
      _UpperCAmelCase     							=   self.get_env()
      _UpperCAmelCase     							=   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       :    Tuple    ):
      # 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
      _UpperCAmelCase     							=   "\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        "

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

      _UpperCAmelCase     							=   "\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
      _UpperCAmelCase     							=   [sys.executable, "-c", "\n".join([load, run]    )]

      # should succeed
      _UpperCAmelCase     							=   self.get_env()
      _UpperCAmelCase     							=   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
      _UpperCAmelCase     							=   [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
      _UpperCAmelCase     							=   "1"
      _UpperCAmelCase     							=   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       :    Optional[Any]    ):
      _UpperCAmelCase     							=   "\nfrom transformers import pipeline\n        "
      _UpperCAmelCase     							=   "\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        "

      _UpperCAmelCase     							=   "\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        "
      _UpperCAmelCase     							=   self.get_env()
      _UpperCAmelCase     							=   "1"
      _UpperCAmelCase     							=   [sys.executable, "-c", "\n".join([load, mock, run]    )]
      _UpperCAmelCase     							=   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       :    Optional[int]    ):
      _UpperCAmelCase     							=   "\nfrom transformers import AutoModel\n        "
      _UpperCAmelCase     							=   "\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
      _UpperCAmelCase     							=   [sys.executable, "-c", "\n".join([load, run]    )]

      # should succeed
      _UpperCAmelCase     							=   self.get_env()
      _UpperCAmelCase     							=   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
      _UpperCAmelCase     							=   "1"
      _UpperCAmelCase     							=   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 gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, PNDMScheduler, StableDiffusionInpaintPipeline, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, slow

from ..pipeline_params import TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS, TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 						lowerCAmelCase__   (      a ,					a ,					a ,					unittest.TestCase					):



 """simple docstring"""



 lowerCAmelCase__  					=      StableDiffusionInpaintPipeline
 lowerCAmelCase__  					=      TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
 lowerCAmelCase__  					=      TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
 lowerCAmelCase__  					=      frozenset(
     []					)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 lowerCAmelCase__  					=      frozenset([]					)
 def       UpperCAmelCase__		(		self					:       str			)  -> List[Any]:


      """simple docstring"""




      torch.manual_seed(0			)
      __SCREAMING_SNAKE_CASE     =      UNetaDConditionModel(
          block_out_channels=(32, 64)					,							layers_per_block=2					,							sample_size=32					,							in_channels=9					,							out_channels=4					,							down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					,							up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")					,							cross_attention_dim=32					,							attention_head_dim=(2, 4)					,							use_linear_projection=__SCREAMING_SNAKE_CASE					,							)
      __SCREAMING_SNAKE_CASE     =      PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE			)
      torch.manual_seed(0			)
      __SCREAMING_SNAKE_CASE     =      AutoencoderKL(
          block_out_channels=[32, 64]					,							in_channels=3					,							out_channels=3					,							down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]					,							up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]					,							latent_channels=4					,							sample_size=128					,							)
      torch.manual_seed(0			)
      __SCREAMING_SNAKE_CASE     =      CLIPTextConfig(
          bos_token_id=0					,							eos_token_id=2					,							hidden_size=32					,							intermediate_size=37					,							layer_norm_eps=1E-05					,							num_attention_heads=4					,							num_hidden_layers=5					,							pad_token_id=1					,							vocab_size=1_000					,							hidden_act="""gelu"""					,							projection_dim=512					,							)
      __SCREAMING_SNAKE_CASE     =      CLIPTextModel(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

      __SCREAMING_SNAKE_CASE     =      {
          """unet""": unet,
          """scheduler""": scheduler,
          """vae""": vae,
          """text_encoder""": text_encoder,
          """tokenizer""": tokenizer,
          """safety_checker""": None,
          """feature_extractor""": None,
      }
      return components
 def       UpperCAmelCase__		(		self					:       Any					,							__SCREAMING_SNAKE_CASE					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       List[Any]=0			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      floats_tensor((1, 3, 32, 32)					,							rng=random.Random(__SCREAMING_SNAKE_CASE			)			).to(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      image.cpu().permute(0					,							2					,							3					,							1			)[0]
      __SCREAMING_SNAKE_CASE     =      Image.fromarray(np.uinta(__SCREAMING_SNAKE_CASE			)			).convert("""RGB"""			).resize((64, 64)			)
      __SCREAMING_SNAKE_CASE     =      Image.fromarray(np.uinta(image + 4			)			).convert("""RGB"""			).resize((64, 64)			)
      if str(__SCREAMING_SNAKE_CASE			).startswith("""mps"""			):
           __SCREAMING_SNAKE_CASE     =      torch.manual_seed(__SCREAMING_SNAKE_CASE			)
      else:
           __SCREAMING_SNAKE_CASE     =      torch.Generator(device=__SCREAMING_SNAKE_CASE			).manual_seed(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      {
          """prompt""": """A painting of a squirrel eating a burger""",
          """image""": init_image,
          """mask_image""": mask_image,
          """generator""": generator,
          """num_inference_steps""": 2,
          """guidance_scale""": 6.0,
          """output_type""": """numpy""",
      }
      return inputs
 def       UpperCAmelCase__		(		self					:       Union[str, Any]			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      __SCREAMING_SNAKE_CASE     =      self.get_dummy_components()
      __SCREAMING_SNAKE_CASE     =      StableDiffusionInpaintPipeline(**__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      sd_pipe.to(__SCREAMING_SNAKE_CASE			)
      sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      self.get_dummy_inputs(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      sd_pipe(**__SCREAMING_SNAKE_CASE			).images
      __SCREAMING_SNAKE_CASE     =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 64, 64, 3)
      __SCREAMING_SNAKE_CASE     =      np.array([0.4727, 0.5735, 0.3941, 0.5446, 0.5926, 0.4394, 0.5062, 0.4654, 0.4476]			)

      assert np.abs(image_slice.flatten() - expected_slice			).max() < 1E-2


 def       UpperCAmelCase__		(		self					:       Tuple			)  -> str:


      """simple docstring"""




      super().test_inference_batch_single_identical(expected_max_diff=3E-3			)







@slow
@require_torch_gpu
class 						lowerCAmelCase__   (      unittest.TestCase					):



 """simple docstring"""



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


      """simple docstring"""




      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()
 def       UpperCAmelCase__		(		self					:       List[str]			)  -> str:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
          """/sd2-inpaint/init_image.png"""			)
      __SCREAMING_SNAKE_CASE     =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"""			)
      __SCREAMING_SNAKE_CASE     =      load_numpy(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint"""
          """/yellow_cat_sitting_on_a_park_bench.npy"""			)

      __SCREAMING_SNAKE_CASE     =      """stabilityai/stable-diffusion-2-inpainting"""
      __SCREAMING_SNAKE_CASE     =      StableDiffusionInpaintPipeline.from_pretrained(__SCREAMING_SNAKE_CASE					,							safety_checker=__SCREAMING_SNAKE_CASE			)
      pipe.to(__SCREAMING_SNAKE_CASE			)
      pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE			)
      pipe.enable_attention_slicing()

      __SCREAMING_SNAKE_CASE     =      """Face of a yellow cat, high resolution, sitting on a park bench"""

      __SCREAMING_SNAKE_CASE     =      torch.manual_seed(0			)
      __SCREAMING_SNAKE_CASE     =      pipe(
          prompt=__SCREAMING_SNAKE_CASE					,							image=__SCREAMING_SNAKE_CASE					,							mask_image=__SCREAMING_SNAKE_CASE					,							generator=__SCREAMING_SNAKE_CASE					,							output_type="""np"""					,							)
      __SCREAMING_SNAKE_CASE     =      output.images[0]

      assert image.shape == (512, 512, 3)
      assert np.abs(expected_image - image			).max() < 9E-3
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> Optional[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
          """/sd2-inpaint/init_image.png"""			)
      __SCREAMING_SNAKE_CASE     =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"""			)
      __SCREAMING_SNAKE_CASE     =      load_numpy(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint"""
          """/yellow_cat_sitting_on_a_park_bench_fp16.npy"""			)

      __SCREAMING_SNAKE_CASE     =      """stabilityai/stable-diffusion-2-inpainting"""
      __SCREAMING_SNAKE_CASE     =      StableDiffusionInpaintPipeline.from_pretrained(
          __SCREAMING_SNAKE_CASE					,							torch_dtype=torch.floataa					,							safety_checker=__SCREAMING_SNAKE_CASE					,							)
      pipe.to(__SCREAMING_SNAKE_CASE			)
      pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE			)
      pipe.enable_attention_slicing()

      __SCREAMING_SNAKE_CASE     =      """Face of a yellow cat, high resolution, sitting on a park bench"""

      __SCREAMING_SNAKE_CASE     =      torch.manual_seed(0			)
      __SCREAMING_SNAKE_CASE     =      pipe(
          prompt=__SCREAMING_SNAKE_CASE					,							image=__SCREAMING_SNAKE_CASE					,							mask_image=__SCREAMING_SNAKE_CASE					,							generator=__SCREAMING_SNAKE_CASE					,							output_type="""np"""					,							)
      __SCREAMING_SNAKE_CASE     =      output.images[0]

      assert image.shape == (512, 512, 3)
      assert np.abs(expected_image - image			).max() < 5E-1


 def       UpperCAmelCase__		(		self					:       Tuple			)  -> Any:


      """simple docstring"""




      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

      __SCREAMING_SNAKE_CASE     =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
          """/sd2-inpaint/init_image.png"""			)
      __SCREAMING_SNAKE_CASE     =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"""			)

      __SCREAMING_SNAKE_CASE     =      """stabilityai/stable-diffusion-2-inpainting"""
      __SCREAMING_SNAKE_CASE     =      PNDMScheduler.from_pretrained(__SCREAMING_SNAKE_CASE					,							subfolder="""scheduler"""			)
      __SCREAMING_SNAKE_CASE     =      StableDiffusionInpaintPipeline.from_pretrained(
          __SCREAMING_SNAKE_CASE					,							safety_checker=__SCREAMING_SNAKE_CASE					,							scheduler=__SCREAMING_SNAKE_CASE					,							torch_dtype=torch.floataa					,							)
      pipe.to(__SCREAMING_SNAKE_CASE			)
      pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE			)
      pipe.enable_attention_slicing(1			)
      pipe.enable_sequential_cpu_offload()

      __SCREAMING_SNAKE_CASE     =      """Face of a yellow cat, high resolution, sitting on a park bench"""

      __SCREAMING_SNAKE_CASE     =      torch.manual_seed(0			)
      __SCREAMING_SNAKE_CASE     =      pipe(
          prompt=__SCREAMING_SNAKE_CASE					,							image=__SCREAMING_SNAKE_CASE					,							mask_image=__SCREAMING_SNAKE_CASE					,							generator=__SCREAMING_SNAKE_CASE					,							num_inference_steps=2					,							output_type="""np"""					,							)

      __SCREAMING_SNAKE_CASE     =      torch.cuda.max_memory_allocated()
      # make sure that less than 2.65 GB is allocated
      assert mem_bytes < 2.65 * 10**9






