



'''simple docstring'''





import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np







class        UpperCAmelCase_			:
    """simple docstring"""







    def lowerCamelCase			(       self   :	Any		):
          torch.manual_seed(0		)
          snake_case__    :       Tuple       		= TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

          torch.manual_seed(0		)
          snake_case__    :       int       		= AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

          torch.manual_seed(0		)
          snake_case__    :       int       		= UNetaDConditionModel(
              sample_size=32					,	layers_per_block=1					,	block_out_channels=[32, 64]					,	down_block_types=[
                  """ResnetDownsampleBlock2D""",
                  """SimpleCrossAttnDownBlock2D""",
              ]					,	mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""					,	up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]					,	in_channels=3					,	out_channels=6					,	cross_attention_dim=32					,	encoder_hid_dim=32					,	attention_head_dim=8					,	addition_embed_type="""text"""					,	addition_embed_type_num_heads=2					,	cross_attention_norm="""group_norm"""					,	resnet_time_scale_shift="""scale_shift"""					,	act_fn="""gelu"""					,	)
          unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          torch.manual_seed(0		)
          snake_case__    :       Optional[int]       		= DDPMScheduler(
              num_train_timesteps=1_000					,	beta_schedule="""squaredcos_cap_v2"""					,	beta_start=0.0001					,	beta_end=0.02					,	thresholding=snake_case_					,	dynamic_thresholding_ratio=0.95					,	sample_max_value=1.0					,	prediction_type="""epsilon"""					,	variance_type="""learned_range"""					,	)

          torch.manual_seed(0		)
          snake_case__    :       Union[str, Any]       		= IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }




    def lowerCamelCase			(       self   :	Optional[Any]		):
          torch.manual_seed(0		)
          snake_case__    :       Tuple       		= TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

          torch.manual_seed(0		)
          snake_case__    :       Optional[int]       		= AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

          torch.manual_seed(0		)
          snake_case__    :       List[str]       		= UNetaDConditionModel(
              sample_size=32					,	layers_per_block=[1, 2]					,	block_out_channels=[32, 64]					,	down_block_types=[
                  """ResnetDownsampleBlock2D""",
                  """SimpleCrossAttnDownBlock2D""",
              ]					,	mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""					,	up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]					,	in_channels=6					,	out_channels=6					,	cross_attention_dim=32					,	encoder_hid_dim=32					,	attention_head_dim=8					,	addition_embed_type="""text"""					,	addition_embed_type_num_heads=2					,	cross_attention_norm="""group_norm"""					,	resnet_time_scale_shift="""scale_shift"""					,	act_fn="""gelu"""					,	class_embed_type="""timestep"""					,	mid_block_scale_factor=1.414					,	time_embedding_act_fn="""gelu"""					,	time_embedding_dim=32					,	)
          unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          torch.manual_seed(0		)
          snake_case__    :       List[str]       		= DDPMScheduler(
              num_train_timesteps=1_000					,	beta_schedule="""squaredcos_cap_v2"""					,	beta_start=0.0001					,	beta_end=0.02					,	thresholding=snake_case_					,	dynamic_thresholding_ratio=0.95					,	sample_max_value=1.0					,	prediction_type="""epsilon"""					,	variance_type="""learned_range"""					,	)

          torch.manual_seed(0		)
          snake_case__    :       int       		= DDPMScheduler(
              num_train_timesteps=1_000					,	beta_schedule="""squaredcos_cap_v2"""					,	beta_start=0.0001					,	beta_end=0.02					,	)

          torch.manual_seed(0		)
          snake_case__    :       List[str]       		= IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "image_noising_scheduler": image_noising_scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }




    def lowerCamelCase			(       self   :	Optional[Any]		):
          snake_case__    :       List[Any]       		= self.get_dummy_components()
          snake_case__    :       Tuple       		= self.pipeline_class(**snake_case_		)
          pipe.to(snake_case_		)
          pipe.set_progress_bar_config(disable=snake_case_		)

          snake_case__    :       Tuple       		= self.get_dummy_inputs(snake_case_		)

          snake_case__    :       int       		= inputs["""prompt"""]
          snake_case__    :       Optional[int]       		= inputs["""generator"""]
          snake_case__    :       Optional[int]       		= inputs["""num_inference_steps"""]
          snake_case__    :       List[str]       		= inputs["""output_type"""]

          if "image" in inputs:
                snake_case__    :       Optional[int]       		= inputs["""image"""]
          else:
                snake_case__    :       str       		= None

          if "mask_image" in inputs:
                snake_case__    :       List[Any]       		= inputs["""mask_image"""]
          else:
                snake_case__    :       Optional[Any]       		= None

          if "original_image" in inputs:
                snake_case__    :       Union[str, Any]       		= inputs["""original_image"""]
          else:
                snake_case__    :       List[Any]       		= None

          snake_case__		,							snake_case__    :       Union[str, Any]       		= pipe.encode_prompt(snake_case_		)

          # inputs with prompt converted to embeddings
          snake_case__    :       List[str]       		= {
              """prompt_embeds""": prompt_embeds,
              """negative_prompt_embeds""": negative_prompt_embeds,
              """generator""": generator,
              """num_inference_steps""": num_inference_steps,
              """output_type""": output_type,
          }

          if image is not None:
                snake_case__    :       List[str]       		= image

          if mask_image is not None:
                snake_case__    :       Dict       		= mask_image

          if original_image is not None:
                snake_case__    :       Tuple       		= original_image

          # set all optional components to None
          for optional_component in pipe._optional_components:
                setattr(snake_case_					,	snake_case_					,	snake_case_		)

          snake_case__    :       Optional[int]       		= pipe(**snake_case_		)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
                pipe.save_pretrained(snake_case_		)
                snake_case__    :       Optional[Any]       		= self.pipeline_class.from_pretrained(snake_case_		)
                pipe_loaded.to(snake_case_		)
                pipe_loaded.set_progress_bar_config(disable=snake_case_		)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          for optional_component in pipe._optional_components:
                self.assertTrue(
                    getattr(snake_case_					,	snake_case_		) is None					,	f"`{optional_component}` did not stay set to None after loading."					,	)

          snake_case__    :       Optional[Any]       		= self.get_dummy_inputs(snake_case_		)

          snake_case__    :       int       		= inputs["""generator"""]
          snake_case__    :       Optional[int]       		= inputs["""num_inference_steps"""]
          snake_case__    :       Any       		= inputs["""output_type"""]

          # inputs with prompt converted to embeddings
          snake_case__    :       Union[str, Any]       		= {
              """prompt_embeds""": prompt_embeds,
              """negative_prompt_embeds""": negative_prompt_embeds,
              """generator""": generator,
              """num_inference_steps""": num_inference_steps,
              """output_type""": output_type,
          }

          if image is not None:
                snake_case__    :       int       		= image

          if mask_image is not None:
                snake_case__    :       Dict       		= mask_image

          if original_image is not None:
                snake_case__    :       List[Any]       		= original_image

          snake_case__    :       Optional[int]       		= pipe_loaded(**snake_case_		)[0]

          snake_case__    :       Tuple       		= np.abs(to_np(snake_case_		) - to_np(snake_case_		)		).max()
          self.assertLess(snake_case_					,	1E-4		)




    def lowerCamelCase			(       self   :	List[str]		):
          snake_case__    :       Union[str, Any]       		= self.get_dummy_components()
          snake_case__    :       List[str]       		= self.pipeline_class(**snake_case_		)
          pipe.to(snake_case_		)
          pipe.set_progress_bar_config(disable=snake_case_		)

          snake_case__    :       List[Any]       		= self.get_dummy_inputs(snake_case_		)
          snake_case__    :       List[Any]       		= pipe(**snake_case_		)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
                pipe.save_pretrained(snake_case_		)
                snake_case__    :       str       		= self.pipeline_class.from_pretrained(snake_case_		)
                pipe_loaded.to(snake_case_		)
                pipe_loaded.set_progress_bar_config(disable=snake_case_		)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          snake_case__    :       Dict       		= self.get_dummy_inputs(snake_case_		)
          snake_case__    :       Optional[Any]       		= pipe_loaded(**snake_case_		)[0]

          snake_case__    :       List[str]       		= np.abs(to_np(snake_case_		) - to_np(snake_case_		)		).max()
          self.assertLess(snake_case_					,	1E-4		)





'''simple docstring'''





import unittest

import torch
from torch import nn

from diffusers.models.activations import get_activation







class        UpperCAmelCase_			(			unittest.TestCase      ):
    """simple docstring"""







    def lowerCamelCase			(       self   :	Union[str, Any]		):
          snake_case__    :       int       		= get_activation("""swish"""		)

          self.assertIsInstance(snake_case_					,	nn.SiLU		)

          self.assertEqual(act(torch.tensor(-100					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertNotEqual(act(torch.tensor(-1					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(0					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(20					,	dtype=torch.floataa		)		).item()					,	20		)




    def lowerCamelCase			(       self   :	Optional[Any]		):
          snake_case__    :       int       		= get_activation("""silu"""		)

          self.assertIsInstance(snake_case_					,	nn.SiLU		)

          self.assertEqual(act(torch.tensor(-100					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertNotEqual(act(torch.tensor(-1					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(0					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(20					,	dtype=torch.floataa		)		).item()					,	20		)




    def lowerCamelCase			(       self   :	Dict		):
          snake_case__    :       Union[str, Any]       		= get_activation("""mish"""		)

          self.assertIsInstance(snake_case_					,	nn.Mish		)

          self.assertEqual(act(torch.tensor(-200					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertNotEqual(act(torch.tensor(-1					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(0					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(20					,	dtype=torch.floataa		)		).item()					,	20		)




    def lowerCamelCase			(       self   :	Union[str, Any]		):
          snake_case__    :       int       		= get_activation("""gelu"""		)

          self.assertIsInstance(snake_case_					,	nn.GELU		)

          self.assertEqual(act(torch.tensor(-100					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertNotEqual(act(torch.tensor(-1					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(0					,	dtype=torch.floataa		)		).item()					,	0		)
          self.assertEqual(act(torch.tensor(20					,	dtype=torch.floataa		)		).item()					,	20		)
