


'''simple docstring'''


import gc
import random
import unittest

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

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class        lowercase__    (		_snake_case	,	_snake_case	,	unittest.TestCase   ):





							'''simple docstring'''


							A_				: Optional[int]       =    CycleDiffusionPipeline
							A_				: Union[str, Any]       =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
							    """negative_prompt""",
							    """height""",
							    """width""",
							    """negative_prompt_embeds""",
							}
							A_				: Tuple       =    PipelineTesterMixin.required_optional_params - {"""latents"""}
							A_				: Optional[Any]       =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""}   )
							A_				: str       =    IMAGE_TO_IMAGE_IMAGE_PARAMS
							A_				: str       =    IMAGE_TO_IMAGE_IMAGE_PARAMS






							def UpperCAmelCase_      (       self    ):
											torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= UNetaDConditionModel(
											    block_out_channels=(32, 64)       ,   layers_per_block=2       ,   sample_size=32       ,   in_channels=4       ,   out_channels=4       ,   down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")       ,   up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")       ,   cross_attention_dim=32       ,   )
											_SCREAMING_SNAKE_CASE					:			Tuple							= DDIMScheduler(
											    beta_start=0.00085       ,   beta_end=0.012       ,   beta_schedule="""scaled_linear"""       ,   num_train_timesteps=1000       ,   clip_sample=__snake_case       ,   set_alpha_to_one=__snake_case       ,   )
											torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= 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       ,   )
											torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			int							= 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=1000       ,   )
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= CLIPTextModel(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			List[str]							= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

											_SCREAMING_SNAKE_CASE					:			Dict							= {
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											    """safety_checker""": None,
											    """feature_extractor""": None,
											}
											return components






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=0    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= floats_tensor((1, 3, 32, 32)       ,   rng=random.Random(__snake_case    )    ).to(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			str							= image / 2 + 0.5
											if str(__snake_case    ).startswith("""mps"""    ):
															_SCREAMING_SNAKE_CASE					:			Tuple							= torch.manual_seed(__snake_case    )
											else:
															_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= torch.Generator(device=__snake_case    ).manual_seed(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= {
											    """prompt""": """An astronaut riding an elephant""",
											    """source_prompt""": """An astronaut riding a horse""",
											    """image""": image,
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """eta""": 0.1,
											    """strength""": 0.8,
											    """guidance_scale""": 3,
											    """source_guidance_scale""": 1,
											    """output_type""": """numpy""",
											}
											return inputs






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """cpu"""  # ensure determinism for the device-dependent torch.Generator

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_dummy_components()
											_SCREAMING_SNAKE_CASE					:			Tuple							= CycleDiffusionPipeline(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= pipe.to(__snake_case    )
											pipe.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_dummy_inputs(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= pipe(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			List[Any]							= output.images

											_SCREAMING_SNAKE_CASE					:			Optional[int]							= images[0, -3:, -3:, -1]

											assert images.shape == (1, 32, 32, 3)
											_SCREAMING_SNAKE_CASE					:			List[Any]							= np.array([0.4459, 0.4943, 0.4544, 0.6643, 0.5474, 0.4327, 0.5701, 0.5959, 0.5179]    )

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






							@unittest.skipIf(torch_device != """cuda"""       ,   """This test requires a GPU"""    )
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_dummy_components()
											for name, module in components.items():
															if hasattr(__snake_case       ,   """half"""    ):
																			_SCREAMING_SNAKE_CASE					:			Tuple							= module.half()
											_SCREAMING_SNAKE_CASE					:			Tuple							= CycleDiffusionPipeline(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Tuple							= pipe.to(__snake_case    )
											pipe.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			List[Any]							= self.get_dummy_inputs(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= pipe(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			str							= output.images

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= images[0, -3:, -3:, -1]

											assert images.shape == (1, 32, 32, 3)
											_SCREAMING_SNAKE_CASE					:			List[str]							= np.array([0.3506, 0.4543, 0.446, 0.4575, 0.5195, 0.4155, 0.5273, 0.518, 0.4116]    )

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






							@skip_mps
							def UpperCAmelCase_      (       self    ):
											return super().test_save_load_local()






							@unittest.skip("""non-deterministic pipeline"""    )
							def UpperCAmelCase_      (       self    ):
											return super().test_inference_batch_single_identical()






							@skip_mps
							def UpperCAmelCase_      (       self    ):
											return super().test_dict_tuple_outputs_equivalent()






							@skip_mps
							def UpperCAmelCase_      (       self    ):
											return super().test_save_load_optional_components()






							@skip_mps
							def UpperCAmelCase_      (       self    ):
											return super().test_attention_slicing_forward_pass()





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





							'''simple docstring'''


							def UpperCAmelCase_      (       self    ):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			str							= load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											    """/cycle-diffusion/black_colored_car.png"""    )
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""    )
											_SCREAMING_SNAKE_CASE					:			Tuple							= init_image.resize((512, 512)    )

											_SCREAMING_SNAKE_CASE					:			Dict							= """CompVis/stable-diffusion-v1-4"""
											_SCREAMING_SNAKE_CASE					:			str							= DDIMScheduler.from_pretrained(__snake_case       ,   subfolder="""scheduler"""    )
											_SCREAMING_SNAKE_CASE					:			Dict							= CycleDiffusionPipeline.from_pretrained(
											    __snake_case       ,   scheduler=__snake_case       ,   safety_checker=__snake_case       ,   torch_dtype=torch.floataa       ,   revision="""fp16"""    )

											pipe.to(__snake_case    )
											pipe.set_progress_bar_config(disable=__snake_case    )
											pipe.enable_attention_slicing()

											_SCREAMING_SNAKE_CASE					:			Any							= """A black colored car"""
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """A blue colored car"""

											_SCREAMING_SNAKE_CASE					:			Tuple							= torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= pipe(
											    prompt=__snake_case       ,   source_prompt=__snake_case       ,   image=__snake_case       ,   num_inference_steps=100       ,   eta=0.1       ,   strength=0.85       ,   guidance_scale=3       ,   source_guidance_scale=1       ,   generator=__snake_case       ,   output_type="""np"""       ,   )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= output.images

											# the values aren't exactly equal, but the images look the same visually
											assert np.abs(image - expected_image    ).max() < 5e-1






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			str							= load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											    """/cycle-diffusion/black_colored_car.png"""    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""    )
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= init_image.resize((512, 512)    )

											_SCREAMING_SNAKE_CASE					:			Tuple							= """CompVis/stable-diffusion-v1-4"""
											_SCREAMING_SNAKE_CASE					:			Dict							= DDIMScheduler.from_pretrained(__snake_case       ,   subfolder="""scheduler"""    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= CycleDiffusionPipeline.from_pretrained(__snake_case       ,   scheduler=__snake_case       ,   safety_checker=__snake_case    )

											pipe.to(__snake_case    )
											pipe.set_progress_bar_config(disable=__snake_case    )
											pipe.enable_attention_slicing()

											_SCREAMING_SNAKE_CASE					:			Tuple							= """A black colored car"""
											_SCREAMING_SNAKE_CASE					:			List[str]							= """A blue colored car"""

											_SCREAMING_SNAKE_CASE					:			Any							= torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= pipe(
											    prompt=__snake_case       ,   source_prompt=__snake_case       ,   image=__snake_case       ,   num_inference_steps=100       ,   eta=0.1       ,   strength=0.85       ,   guidance_scale=3       ,   source_guidance_scale=1       ,   generator=__snake_case       ,   output_type="""np"""       ,   )
											_SCREAMING_SNAKE_CASE					:			List[Any]							= output.images

											assert np.abs(image - expected_image    ).max() < 2e-2








'''simple docstring'''


import subprocess
import sys

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





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							@require_torch
							def UpperCAmelCase_      (       self    ):
											# 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					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Dict							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Optional[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
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			int							= """1"""
											_SCREAMING_SNAKE_CASE					:			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()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# python one-liner segments
											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Tuple							= """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
											_SCREAMING_SNAKE_CASE					:			Tuple							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			str							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Tuple							= 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 UpperCAmelCase_      (       self    ):
											# 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					:			Tuple							= """
from transformers import BertConfig, BertModel, BertTokenizer
        """

											_SCREAMING_SNAKE_CASE					:			str							= """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

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

											# should succeed
											_SCREAMING_SNAKE_CASE					:			List[str]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			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()    )

											# next emulate no network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [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					:			Dict							= """1"""
											_SCREAMING_SNAKE_CASE					:			int							= 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 UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import pipeline
        """
											_SCREAMING_SNAKE_CASE					:			int							= """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Any							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]
											_SCREAMING_SNAKE_CASE					:			List[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 UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import AutoModel
        """
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

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

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			List[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()    )

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """1"""
											_SCREAMING_SNAKE_CASE					:			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()    )




