


'''simple docstring'''
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class        _UpperCamelCase	(    SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			unittest.TestCase):


				'''simple docstring'''


				_snake_case						       =				StableDiffusionSAGPipeline
				_snake_case						       =				TEXT_TO_IMAGE_PARAMS
				_snake_case						       =				TEXT_TO_IMAGE_BATCH_PARAMS
				_snake_case						       =				TEXT_TO_IMAGE_IMAGE_PARAMS
				_snake_case						       =				TEXT_TO_IMAGE_IMAGE_PARAMS
				_snake_case						       =				False

				def 	a__		(     self							)    ->  str:
					torch.manual_seed(0							)
					lowercase						:		Tuple   					=		UNetaDConditionModel(
					    block_out_channels=(3_2, 6_4)							,      layers_per_block=2							,      sample_size=3_2							,      in_channels=4							,      out_channels=4							,      down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")							,      up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")							,      cross_attention_dim=3_2							,      )
					lowercase						:		Union[str, Any]   					=		DDIMScheduler(
					    beta_start=0.0_00_85							,      beta_end=0.0_12							,      beta_schedule="scaled_linear"							,      clip_sample=a_							,      set_alpha_to_one=a_							,      )
					torch.manual_seed(0							)
					lowercase						:		Any   					=		AutoencoderKL(
					    block_out_channels=[3_2, 6_4]							,      in_channels=3							,      out_channels=3							,      down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]							,      up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]							,      latent_channels=4							,      )
					torch.manual_seed(0							)
					lowercase						:		Optional[int]   					=		CLIPTextConfig(
					    bos_token_id=0							,      eos_token_id=2							,      hidden_size=3_2							,      intermediate_size=3_7							,      layer_norm_eps=1e-0_5							,      num_attention_heads=4							,      num_hidden_layers=5							,      pad_token_id=1							,      vocab_size=1_0_0_0							,      )
					lowercase						:		int   					=		CLIPTextModel(a_							)
					lowercase						:		str   					=		CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

					lowercase						:		Union[str, Any]   					=		{
					    "unet": unet,
					    "scheduler": scheduler,
					    "vae": vae,
					    "text_encoder": text_encoder,
					    "tokenizer": tokenizer,
					    "safety_checker": None,
					    "feature_extractor": None,
					}
					return components

				def 	a__		(     self							,      a_							,      a_=0							)    ->  Tuple:
					if str(a_							).startswith("mps"							):
						lowercase						:		Tuple   					=		torch.manual_seed(a_							)
					else:
						lowercase						:		List[Any]   					=		torch.Generator(device=a_							).manual_seed(a_							)
					lowercase						:		int   					=		{
					    "prompt": ".",
					    "generator": generator,
					    "num_inference_steps": 2,
					    "guidance_scale": 1.0,
					    "sag_scale": 1.0,
					    "output_type": "numpy",
					}
					return inputs


				def 	a__		(     self							)    ->  List[Any]:
					super().test_inference_batch_single_identical(expected_max_diff=3e-3							)




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


				'''simple docstring'''


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

				def 	a__		(     self							)    ->  Optional[Any]:
					lowercase						:		Optional[Any]   					=		StableDiffusionSAGPipeline.from_pretrained("CompVis/stable-diffusion-v1-4"							)
					lowercase						:		Any   					=		sag_pipe.to(a_							)
					sag_pipe.set_progress_bar_config(disable=a_							)

					lowercase						:		List[Any]   					=		"."
					lowercase						:		Optional[int]   					=		torch.manual_seed(0							)
					lowercase						:		int   					=		sag_pipe(
					    [prompt]							,      generator=a_							,      guidance_scale=7.5							,      sag_scale=1.0							,      num_inference_steps=2_0							,      output_type="np"							)

					lowercase						:		Optional[Any]   					=		output.images

					lowercase						:		Dict   					=		image[0, -3:, -3:, -1]

					assert image.shape == (1, 5_1_2, 5_1_2, 3)
					lowercase						:		Optional[int]   					=		np.array([0.15_68, 0.17_38, 0.16_95, 0.16_93, 0.15_07, 0.17_05, 0.15_47, 0.17_51, 0.19_49]							)

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

				def 	a__		(     self							)    ->  List[Any]:
					lowercase						:		Optional[Any]   					=		StableDiffusionSAGPipeline.from_pretrained("stabilityai/stable-diffusion-2-1-base"							)
					lowercase						:		List[Any]   					=		sag_pipe.to(a_							)
					sag_pipe.set_progress_bar_config(disable=a_							)

					lowercase						:		str   					=		"."
					lowercase						:		int   					=		torch.manual_seed(0							)
					lowercase						:		int   					=		sag_pipe(
					    [prompt]							,      generator=a_							,      guidance_scale=7.5							,      sag_scale=1.0							,      num_inference_steps=2_0							,      output_type="np"							)

					lowercase						:		str   					=		output.images

					lowercase						:		List[str]   					=		image[0, -3:, -3:, -1]

					assert image.shape == (1, 5_1_2, 5_1_2, 3)
					lowercase						:		str   					=		np.array([0.34_59, 0.28_76, 0.25_37, 0.30_02, 0.26_71, 0.21_60, 0.30_26, 0.22_62, 0.23_71]							)

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


				def 	a__		(     self							)    ->  int:
					lowercase						:		List[Any]   					=		StableDiffusionSAGPipeline.from_pretrained("stabilityai/stable-diffusion-2-1-base"							)
					lowercase						:		int   					=		sag_pipe.to(a_							)
					sag_pipe.set_progress_bar_config(disable=a_							)

					lowercase						:		List[Any]   					=		"."
					lowercase						:		List[Any]   					=		torch.manual_seed(0							)
					lowercase						:		Dict   					=		sag_pipe(
					    [prompt]							,      width=7_6_8							,      height=5_1_2							,      generator=a_							,      guidance_scale=7.5							,      sag_scale=1.0							,      num_inference_steps=2_0							,      output_type="np"							,      )

					lowercase						:		Union[str, Any]   					=		output.images

					assert image.shape == (1, 5_1_2, 7_6_8, 3)






'''simple docstring'''
from pathlib import Path
from typing import List

from transformers import is_torch_available, is_vision_available
from transformers.testing_utils import get_tests_dir, is_tool_test
from transformers.tools.agent_types import AGENT_TYPE_MAPPING, AgentAudio, AgentImage, AgentText


if is_torch_available():
					import torch

if is_vision_available():
					from PIL import Image


lowerCAmelCase   : List[Any]             =				["""text""", """image""", """audio"""]




def 	_A  (       A							)  ->				Dict:
	lowercase						:		str   					=		[]

	for input_type in input_types:
		if input_type == "text":
			inputs.append("Text input"							)
		elif input_type == "image":
			inputs.append(
			    Image.open(Path(get_tests_dir("fixtures/tests_samples/COCO"							)							) / "000000039769.png"							).resize((5_1_2, 5_1_2)							)							)
		elif input_type == "audio":
			inputs.append(torch.ones(3_0_0_0							)							)
		elif isinstance(A							,A							):
			inputs.append(create_inputs(A							)							)
		else:
			raise ValueError(F'''Invalid type requested: {input_type}'''							)

	return inputs




def 	_A  (       A							)  ->				str:
	lowercase						:		Tuple   					=		[]

	for output in outputs:
		if isinstance(A							,(str, AgentText)							):
			output_types.append("text"							)
		elif isinstance(A							,(Image.Image, AgentImage)							):
			output_types.append("image"							)
		elif isinstance(A							,(torch.Tensor, AgentAudio)							):
			output_types.append("audio"							)
		else:
			raise ValueError(F'''Invalid output: {output}'''							)

	return output_types






@is_tool_test
class        _UpperCamelCase	:


				'''simple docstring'''


				def 	a__		(     self							)    ->  Optional[Any]:
					self.assertTrue(hasattr(self.tool							,      "inputs"							)							)
					self.assertTrue(hasattr(self.tool							,      "outputs"							)							)

					lowercase						:		Optional[Any]   					=		self.tool.inputs
					for _input in inputs:
						if isinstance(_input							,      a_							):
							for __input in _input:
								self.assertTrue(__input in authorized_types							)
						else:
							self.assertTrue(_input in authorized_types							)

					lowercase						:		Any   					=		self.tool.outputs
					for _output in outputs:
						self.assertTrue(_output in authorized_types							)

				def 	a__		(     self							)    ->  Any:
					lowercase						:		Any   					=		create_inputs(self.tool.inputs							)
					lowercase						:		Tuple   					=		self.tool(*a_							)

					# There is a single output
					if len(self.tool.outputs							) == 1:
						lowercase						:		Any   					=		[outputs]

					self.assertListEqual(output_types(a_							)							,      self.tool.outputs							)

				def 	a__		(     self							)    ->  List[str]:
					self.assertTrue(hasattr(self.tool							,      "description"							)							)
					self.assertTrue(hasattr(self.tool							,      "default_checkpoint"							)							)
					self.assertTrue(self.tool.description.startswith("This is a tool that"							)							)

				def 	a__		(     self							)    ->  int:
					lowercase						:		str   					=		create_inputs(self.tool.inputs							)
					lowercase						:		str   					=		self.tool(*a_							)

					if not isinstance(a_							,      a_							):
						lowercase						:		Union[str, Any]   					=		[outputs]

					self.assertEqual(len(a_							)							,      len(self.tool.outputs							)							)

					for output, output_type in zip(a_							,      self.tool.outputs							):
						lowercase						:		List[str]   					=		AGENT_TYPE_MAPPING[output_type]
						self.assertTrue(isinstance(a_							,      a_							)							)


				def 	a__		(     self							)    ->  Optional[int]:
					lowercase						:		int   					=		create_inputs(self.tool.inputs							)

					lowercase						:		str   					=		[]

					for _input, input_type in zip(a_							,      self.tool.inputs							):
						if isinstance(a_							,      a_							):
							_inputs.append([AGENT_TYPE_MAPPING[_input_type](_input							) for _input_type in input_type]							)
						else:
							_inputs.append(AGENT_TYPE_MAPPING[input_type](_input							)							)

        # Should not raise an error
					lowercase						:		Optional[int]   					=		self.tool(*a_							)

					if not isinstance(a_							,      a_							):
						lowercase						:		str   					=		[outputs]

					self.assertEqual(len(a_							)							,      len(self.tool.outputs							)							)


