




'''simple docstring'''

import unittest
from typing import Dict, List, Optional, Union

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
							import torch

if is_vision_available():
							from PIL import Image

							from transformers import BridgeTowerImageProcessor






class 						a				(   unittest.TestCase     ):


				def __init__(		self    :					Optional[Any]       ,				lowercase_    :					Tuple       ,				lowercase_    :					bool = True       ,				lowercase_    :					Dict[str, int] = None       ,				lowercase_    :					int = 32       ,				lowercase_    :					bool = True       ,				lowercase_    :					Union[int, float] = 1 / 255       ,				lowercase_    :					bool = True       ,				lowercase_    :					bool = True       ,				lowercase_    :					Optional[Union[float, List[float]]] = [0.4814_5466, 0.457_8275, 0.4082_1073]       ,				lowercase_    :					Optional[Union[float, List[float]]] = [0.2686_2954, 0.2613_0258, 0.2757_7711]       ,				lowercase_    :					bool = True       ,				lowercase_    :					Dict=7       ,				lowercase_    :					Dict=30       ,				lowercase_    :					List[str]=400       ,				lowercase_    :					Optional[int]=3       ,				):
									snake_case_									=				parent
									snake_case_									=				do_resize
									snake_case_									=				size if size is not None else {'''shortest_edge''': 288}
									snake_case_									=				size_divisor
									snake_case_									=				do_rescale
									snake_case_									=				rescale_factor
									snake_case_									=				do_normalize
									snake_case_									=				do_center_crop
									snake_case_									=				image_mean
									snake_case_									=				image_std
									snake_case_									=				do_pad
									snake_case_									=				batch_size
									snake_case_									=				num_channels
									snake_case_									=				min_resolution
									snake_case_									=				max_resolution


				def   A_ (		self    :					str					):
									return {
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_normalize": self.do_normalize,
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "size_divisor": self.size_divisor,
									}


				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					Any       ,				lowercase_    :					List[Any]=False					):
									if not batched:
														snake_case_									=				self.size['''shortest_edge''']
														snake_case_									=				image_inputs[0]
														if isinstance(lowercase_       ,				Image.Image					):
																			snake_case_			,snake_case_									=				image.size
														else:
																			snake_case_			,snake_case_									=				image.shape[1], image.shape[2]
														snake_case_									=				size / min(lowercase_       ,				lowercase_					)
														if h < w:
																			snake_case_			,snake_case_									=				size, scale * w
														else:
																			snake_case_			,snake_case_									=				scale * h, size

														snake_case_									=				int((1333 / 800) * size					)
														if max(lowercase_       ,				lowercase_					) > max_size:
																			snake_case_									=				max_size / max(lowercase_       ,				lowercase_					)
																			snake_case_									=				newh * scale
																			snake_case_									=				neww * scale

														snake_case_			,snake_case_									=				int(newh + 0.5					), int(neww + 0.5					)
														snake_case_			,snake_case_									=				(
														    newh // self.size_divisor * self.size_divisor,
														    neww // self.size_divisor * self.size_divisor,
														)

									else:
														snake_case_									=				[]
														for image in image_inputs:
																			snake_case_			,snake_case_									=				self.get_expected_values([image]					)
																			expected_values.append((expected_height, expected_width)					)
														snake_case_									=				max(lowercase_       ,				key=lambda lowercase_					: item[0]					)[0]
														snake_case_									=				max(lowercase_       ,				key=lambda lowercase_					: item[1]					)[1]

									return expected_height, expected_width







@require_torch
@require_vision
class 						a				(   _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		BridgeTowerImageProcessor if is_vision_available() else None


				def   A_ (		self    :					str					):
									snake_case_									=				BridgeTowerImageProcessingTester(self					)


				@property
				def   A_ (		self    :					Any					):
									return self.image_processor_tester.prepare_image_processor_dict()


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				self.image_processing_class(**self.image_processor_dict					)
									self.assertTrue(hasattr(lowercase_       ,				'''image_mean'''					)					)
									self.assertTrue(hasattr(lowercase_       ,				'''image_std'''					)					)
									self.assertTrue(hasattr(lowercase_       ,				'''do_normalize'''					)					)
									self.assertTrue(hasattr(lowercase_       ,				'''do_resize'''					)					)
									self.assertTrue(hasattr(lowercase_       ,				'''size'''					)					)
									self.assertTrue(hasattr(lowercase_       ,				'''size_divisor'''					)					)


				def   A_ (		self    :					Union[str, Any]					):
									pass


				def   A_ (		self    :					List[Any]					):
									# Initialize image processor
									snake_case_									=				self.image_processing_class(**self.image_processor_dict					)
									# create random PIL images
									snake_case_									=				prepare_image_inputs(self.image_processor_tester       ,				equal_resolution=lowercase_					)
									for image in image_inputs:
														self.assertIsInstance(lowercase_       ,				Image.Image					)

									# Test not batched input
									snake_case_									=				image_processing(image_inputs[0]       ,				return_tensors='''pt'''					).pixel_values

									snake_case_			,snake_case_									=				self.image_processor_tester.get_expected_values(lowercase_					)
									self.assertEqual(
									    encoded_images.shape       ,				(1, self.image_processor_tester.num_channels, expected_height, expected_width)       ,				)

									# Test batched
									snake_case_									=				image_processing(lowercase_       ,				return_tensors='''pt'''					).pixel_values

									snake_case_			,snake_case_									=				self.image_processor_tester.get_expected_values(lowercase_       ,				batched=lowercase_					)
									self.assertEqual(
									    encoded_images.shape       ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        expected_height,
									        expected_width,
									    )       ,				)


				def   A_ (		self    :					Dict					):
									# Initialize image processor
									snake_case_									=				self.image_processing_class(**self.image_processor_dict					)
									# create random numpy tensors
									snake_case_									=				prepare_image_inputs(self.image_processor_tester       ,				equal_resolution=lowercase_       ,				numpify=lowercase_					)
									for image in image_inputs:
														self.assertIsInstance(lowercase_       ,				np.ndarray					)

									# Test not batched input
									snake_case_									=				image_processing(image_inputs[0]       ,				return_tensors='''pt'''					).pixel_values

									snake_case_			,snake_case_									=				self.image_processor_tester.get_expected_values(lowercase_					)
									self.assertEqual(
									    encoded_images.shape       ,				(1, self.image_processor_tester.num_channels, expected_height, expected_width)       ,				)

									# Test batched
									snake_case_									=				image_processing(lowercase_       ,				return_tensors='''pt'''					).pixel_values

									snake_case_			,snake_case_									=				self.image_processor_tester.get_expected_values(lowercase_       ,				batched=lowercase_					)
									self.assertEqual(
									    encoded_images.shape       ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        expected_height,
									        expected_width,
									    )       ,				)


				def   A_ (		self    :					Any					):
									# Initialize image processor
									snake_case_									=				self.image_processing_class(**self.image_processor_dict					)
									# create random PyTorch tensors
									snake_case_									=				prepare_image_inputs(self.image_processor_tester       ,				equal_resolution=lowercase_       ,				torchify=lowercase_					)
									for image in image_inputs:
														self.assertIsInstance(lowercase_       ,				torch.Tensor					)

									# Test not batched input
									snake_case_									=				image_processing(image_inputs[0]       ,				return_tensors='''pt'''					).pixel_values

									snake_case_			,snake_case_									=				self.image_processor_tester.get_expected_values(lowercase_					)
									self.assertEqual(
									    encoded_images.shape       ,				(1, self.image_processor_tester.num_channels, expected_height, expected_width)       ,				)

									# Test batched
									snake_case_									=				image_processing(lowercase_       ,				return_tensors='''pt'''					).pixel_values

									snake_case_			,snake_case_									=				self.image_processor_tester.get_expected_values(lowercase_       ,				batched=lowercase_					)
									self.assertEqual(
									    encoded_images.shape       ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        expected_height,
									        expected_width,
									    )       ,				)







'''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 						a				(   _lowerCamelCase   ,     _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		CycleDiffusionPipeline
				snake_case_      =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
				    "negative_prompt",
				    "height",
				    "width",
				    "negative_prompt_embeds",
				}
				snake_case_      =		PipelineTesterMixin.required_optional_params - {"latents"}
				snake_case_      =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"source_prompt"}     )
				snake_case_      =		IMAGE_TO_IMAGE_IMAGE_PARAMS
				snake_case_      =		IMAGE_TO_IMAGE_IMAGE_PARAMS


				def   A_ (		self    :					Tuple					):
									torch.manual_seed(0					)
									snake_case_									=				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       ,				)
									snake_case_									=				DDIMScheduler(
									    beta_start=0.0_0085       ,				beta_end=0.012       ,				beta_schedule='''scaled_linear'''       ,				num_train_timesteps=1000       ,				clip_sample=lowercase_       ,				set_alpha_to_one=lowercase_       ,				)
									torch.manual_seed(0					)
									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       ,				)
									torch.manual_seed(0					)
									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=1000       ,				)
									snake_case_									=				CLIPTextModel(lowercase_					)
									snake_case_									=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

									snake_case_									=				{
									    '''unet''': unet,
									    '''scheduler''': scheduler,
									    '''vae''': vae,
									    '''text_encoder''': text_encoder,
									    '''tokenizer''': tokenizer,
									    '''safety_checker''': None,
									    '''feature_extractor''': None,
									}
									return components


				def   A_ (		self    :					Any       ,				lowercase_    :					int       ,				lowercase_    :					Optional[Any]=0					):
									snake_case_									=				floats_tensor((1, 3, 32, 32)       ,				rng=random.Random(lowercase_					)					).to(lowercase_					)
									snake_case_									=				image / 2 + 0.5
									if str(lowercase_					).startswith('''mps'''					):
														snake_case_									=				torch.manual_seed(lowercase_					)
									else:
														snake_case_									=				torch.Generator(device=lowercase_					).manual_seed(lowercase_					)
									snake_case_									=				{
									    '''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   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				'''cpu'''  # ensure determinism for the device-dependent torch.Generator

									snake_case_									=				self.get_dummy_components()
									snake_case_									=				CycleDiffusionPipeline(**lowercase_					)
									snake_case_									=				pipe.to(lowercase_					)
									pipe.set_progress_bar_config(disable=lowercase_					)

									snake_case_									=				self.get_dummy_inputs(lowercase_					)
									snake_case_									=				pipe(**lowercase_					)
									snake_case_									=				output.images

									snake_case_									=				images[0, -3:, -3:, -1]

									assert images.shape == (1, 32, 32, 3)
									snake_case_									=				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   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				self.get_dummy_components()
									for name, module in components.items():
														if hasattr(lowercase_       ,				'''half'''					):
																			snake_case_									=				module.half()
									snake_case_									=				CycleDiffusionPipeline(**lowercase_					)
									snake_case_									=				pipe.to(lowercase_					)
									pipe.set_progress_bar_config(disable=lowercase_					)

									snake_case_									=				self.get_dummy_inputs(lowercase_					)
									snake_case_									=				pipe(**lowercase_					)
									snake_case_									=				output.images

									snake_case_									=				images[0, -3:, -3:, -1]

									assert images.shape == (1, 32, 32, 3)
									snake_case_									=				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   A_ (		self    :					Optional[int]					):
									return super().test_save_load_local()


				@unittest.skip('''non-deterministic pipeline'''					)
				def   A_ (		self    :					List[Any]					):
									return super().test_inference_batch_single_identical()


				@skip_mps
				def   A_ (		self    :					Union[str, Any]					):
									return super().test_dict_tuple_outputs_equivalent()


				@skip_mps
				def   A_ (		self    :					Union[str, Any]					):
									return super().test_save_load_optional_components()


				@skip_mps
				def   A_ (		self    :					Union[str, Any]					):
									return super().test_attention_slicing_forward_pass()







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


				def   A_ (		self    :					List[Any]					):
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
									    '''/cycle-diffusion/black_colored_car.png'''					)
									snake_case_									=				load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'''					)
									snake_case_									=				init_image.resize((512, 512)					)

									snake_case_									=				'''CompVis/stable-diffusion-v1-4'''
									snake_case_									=				DDIMScheduler.from_pretrained(lowercase_       ,				subfolder='''scheduler'''					)
									snake_case_									=				CycleDiffusionPipeline.from_pretrained(
									    lowercase_       ,				scheduler=lowercase_       ,				safety_checker=lowercase_       ,				torch_dtype=torch.floataa       ,				revision='''fp16'''					)

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

									snake_case_									=				'''A black colored car'''
									snake_case_									=				'''A blue colored car'''

									snake_case_									=				torch.manual_seed(0					)
									snake_case_									=				pipe(
									    prompt=lowercase_       ,				source_prompt=lowercase_       ,				image=lowercase_       ,				num_inference_steps=100       ,				eta=0.1       ,				strength=0.85       ,				guidance_scale=3       ,				source_guidance_scale=1       ,				generator=lowercase_       ,				output_type='''np'''       ,				)
									snake_case_									=				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   A_ (		self    :					List[str]					):
									snake_case_									=				load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
									    '''/cycle-diffusion/black_colored_car.png'''					)
									snake_case_									=				load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'''					)
									snake_case_									=				init_image.resize((512, 512)					)

									snake_case_									=				'''CompVis/stable-diffusion-v1-4'''
									snake_case_									=				DDIMScheduler.from_pretrained(lowercase_       ,				subfolder='''scheduler'''					)
									snake_case_									=				CycleDiffusionPipeline.from_pretrained(lowercase_       ,				scheduler=lowercase_       ,				safety_checker=lowercase_					)

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

									snake_case_									=				'''A black colored car'''
									snake_case_									=				'''A blue colored car'''

									snake_case_									=				torch.manual_seed(0					)
									snake_case_									=				pipe(
									    prompt=lowercase_       ,				source_prompt=lowercase_       ,				image=lowercase_       ,				num_inference_steps=100       ,				eta=0.1       ,				strength=0.85       ,				guidance_scale=3       ,				source_guidance_scale=1       ,				generator=lowercase_       ,				output_type='''np'''       ,				)
									snake_case_									=				output.images

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

