


import gc
import random
import tempfile
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion_safe import StableDiffusionPipelineSafe as StableDiffusionPipeline
from diffusers.utils import floats_tensor, nightly, torch_device
from diffusers.utils.testing_utils import require_torch_gpu
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Optional[Any]:
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Tuple:
             a_					:		Optional[int]							=  1
             a_					:		Union[str, Any]							=  3
             a_					:		Union[str, Any]							=  (3_2, 3_2)

             a_					:		str							=  floats_tensor((batch_size, num_channels) + sizes     ,							rng=random.Random(0     )     ).to(SCREAMING_SNAKE_CASE__     )
             return image
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Tuple     )		->			str:
             torch.manual_seed(0     )
             a_					:		List[Any]							=  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     ,							)
             return model
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			List[Any]:
             torch.manual_seed(0     )
             a_					:		int							=  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     ,							)
             return model
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     )		->			Tuple:
             torch.manual_seed(0     )
             a_					:		List[str]							=  CLIPTextConfig(
                 bos_token_id=0     ,							eos_token_id=2     ,							hidden_size=3_2     ,							intermediate_size=3_7     ,							layer_norm_eps=1E-05     ,							num_attention_heads=4     ,							num_hidden_layers=5     ,							pad_token_id=1     ,							vocab_size=1_0_0_0     ,							)
             return CLIPTextModel(SCREAMING_SNAKE_CASE__     )
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[Any]:
             def extract(*SCREAMING_SNAKE_CASE__				:       Dict     ,							**SCREAMING_SNAKE_CASE__				:       Optional[int]     ):
                    class  SCREAMING_SNAKE_CASE__  :
                          def __init__(       self				:       Union[str, Any]     )		->			List[Any]:
                                 a_					:		str							=  torch.ones([0]     )

                          def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       Tuple     )		->			str:
                                 self.pixel_values.to(SCREAMING_SNAKE_CASE__     )
                                 return self
                    return Out()

             return extract
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			Optional[int]:
             a_					:		Optional[Any]							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Optional[Any]							=  self.dummy_cond_unet
             a_					:		List[str]							=  DDIMScheduler(
                 beta_start=0.00085     ,							beta_end=0.012     ,							beta_schedule='scaled_linear'     ,							clip_sample=SCREAMING_SNAKE_CASE__     ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__     ,							)

             a_					:		Optional[Any]							=  self.dummy_vae
             a_					:		int							=  self.dummy_text_encoder
             a_					:		Any							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             # make sure here that pndm scheduler skips prk
             a_					:		List[Any]							=  StableDiffusionPipeline(
                 unet=SCREAMING_SNAKE_CASE__     ,							scheduler=SCREAMING_SNAKE_CASE__     ,							vae=SCREAMING_SNAKE_CASE__     ,							text_encoder=SCREAMING_SNAKE_CASE__     ,							tokenizer=SCREAMING_SNAKE_CASE__     ,							safety_checker=SCREAMING_SNAKE_CASE__     ,							feature_extractor=self.dummy_extractor     ,							)
             a_					:		List[str]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Optional[int]							=  'A painting of a squirrel eating a burger'

             a_					:		str							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(0     )
             a_					:		List[Any]							=  sd_pipe([prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=6.0     ,							num_inference_steps=2     ,							output_type='np'     )
             a_					:		str							=  output.images

             a_					:		str							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(0     )
             a_					:		Optional[Any]							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=6.0     ,							num_inference_steps=2     ,							output_type='np'     ,							return_dict=SCREAMING_SNAKE_CASE__     ,							)[0]

             a_					:		str							=  image[0, -3:, -3:, -1]
             a_					:		int							=  image_from_tuple[0, -3:, -3:, -1]

             assert image.shape == (1, 6_4, 6_4, 3)
             a_					:		Dict							=  np.array([0.5756, 0.6118, 0.5005, 0.5041, 0.5471, 0.4726, 0.4976, 0.4865, 0.4864]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
             assert np.abs(image_from_tuple_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     )		->			Tuple:
             a_					:		str							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Dict							=  self.dummy_cond_unet
             a_					:		List[Any]							=  PNDMScheduler(skip_prk_steps=SCREAMING_SNAKE_CASE__     )
             a_					:		Any							=  self.dummy_vae
             a_					:		Optional[Any]							=  self.dummy_text_encoder
             a_					:		List[str]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             # make sure here that pndm scheduler skips prk
             a_					:		Any							=  StableDiffusionPipeline(
                 unet=SCREAMING_SNAKE_CASE__     ,							scheduler=SCREAMING_SNAKE_CASE__     ,							vae=SCREAMING_SNAKE_CASE__     ,							text_encoder=SCREAMING_SNAKE_CASE__     ,							tokenizer=SCREAMING_SNAKE_CASE__     ,							safety_checker=SCREAMING_SNAKE_CASE__     ,							feature_extractor=self.dummy_extractor     ,							)
             a_					:		Dict							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		int							=  'A painting of a squirrel eating a burger'
             a_					:		Any							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(0     )
             a_					:		Any							=  sd_pipe([prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=6.0     ,							num_inference_steps=2     ,							output_type='np'     )

             a_					:		Any							=  output.images

             a_					:		Optional[int]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(0     )
             a_					:		Dict							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=6.0     ,							num_inference_steps=2     ,							output_type='np'     ,							return_dict=SCREAMING_SNAKE_CASE__     ,							)[0]

             a_					:		Tuple							=  image[0, -3:, -3:, -1]
             a_					:		List[str]							=  image_from_tuple[0, -3:, -3:, -1]

             assert image.shape == (1, 6_4, 6_4, 3)
             a_					:		Any							=  np.array([0.5125, 0.5716, 0.4828, 0.5060, 0.5650, 0.4768, 0.5185, 0.4895, 0.4993]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
             assert np.abs(image_from_tuple_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     )		->			Optional[Any]:
             a_					:		Optional[Any]							=  StableDiffusionPipeline.from_pretrained(
                 'hf-internal-testing/tiny-stable-diffusion-lms-pipe'     ,							safety_checker=SCREAMING_SNAKE_CASE__     )
             assert isinstance(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     )
             assert isinstance(pipe.scheduler     ,							SCREAMING_SNAKE_CASE__     )
             assert pipe.safety_checker is None

             a_					:		Union[str, Any]							=  pipe('example prompt'     ,							num_inference_steps=2     ).images[0]
             assert image is not None

             # check that there's no error when saving a pipeline with one of the models being None
             with tempfile.TemporaryDirectory() as tmpdirname:
                    pipe.save_pretrained(SCREAMING_SNAKE_CASE__     )
                    a_					:		List[str]							=  StableDiffusionPipeline.from_pretrained(SCREAMING_SNAKE_CASE__     )

             # sanity check that the pipeline still works
             assert pipe.safety_checker is None
             a_					:		str							=  pipe('example prompt'     ,							num_inference_steps=2     ).images[0]
             assert image is not None

      @unittest.skipIf(torch_device != 'cuda'     ,							'This test requires a GPU'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			Optional[int]:
             a_					:		Union[str, Any]							=  self.dummy_cond_unet
             a_					:		Dict							=  PNDMScheduler(skip_prk_steps=SCREAMING_SNAKE_CASE__     )
             a_					:		Any							=  self.dummy_vae
             a_					:		Tuple							=  self.dummy_text_encoder
             a_					:		List[Any]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             # put models in fp16
             a_					:		Union[str, Any]							=  unet.half()
             a_					:		Dict							=  vae.half()
             a_					:		Tuple							=  bert.half()

             # make sure here that pndm scheduler skips prk
             a_					:		Tuple							=  StableDiffusionPipeline(
                 unet=SCREAMING_SNAKE_CASE__     ,							scheduler=SCREAMING_SNAKE_CASE__     ,							vae=SCREAMING_SNAKE_CASE__     ,							text_encoder=SCREAMING_SNAKE_CASE__     ,							tokenizer=SCREAMING_SNAKE_CASE__     ,							safety_checker=SCREAMING_SNAKE_CASE__     ,							feature_extractor=self.dummy_extractor     ,							)
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Dict							=  'A painting of a squirrel eating a burger'
             a_					:		Tuple							=  sd_pipe([prompt]     ,							num_inference_steps=2     ,							output_type='np'     ).images

             assert image.shape == (1, 6_4, 6_4, 3)
@nightly
@require_torch_gpu
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     )		->			str:
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     )		->			Optional[Any]:
             a_					:		Dict							=  StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'     ,							safety_checker=SCREAMING_SNAKE_CASE__     )
             a_					:		Optional[int]							=  LMSDiscreteScheduler.from_config(sd_pipe.scheduler.config     )
             a_					:		Optional[Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Union[str, Any]							=  (
                 'portrait of girl with smokey eyes makeup in abandoned hotel, grange clothes, redshift, wide high angle'
                 ' coloured polaroid photograph with flash, kodak film, hyper real, stunning moody cinematography, with'
                 ' anamorphic lenses, by maripol, fallen angels by wong kar - wai, style of suspiria and neon demon and'
                 ' children from bahnhof zoo, detailed '
             )
             a_					:		Dict							=  4_0_0_3_6_6_0_3_4_6
             a_					:		Union[str, Any]							=  7

             # without safety guidance (sld_guidance_scale = 0)
             a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		str							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=5_0     ,							output_type='np'     ,							width=5_1_2     ,							height=5_1_2     ,							sld_guidance_scale=0     ,							)

             a_					:		List[str]							=  output.images
             a_					:		Union[str, Any]							=  image[0, -3:, -3:, -1]
             a_					:		Any							=  [0.2278, 0.2231, 0.2249, 0.2333, 0.2303, 0.1885, 0.2273, 0.2144, 0.2176]

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

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

             # without safety guidance (strong configuration)
             a_					:		Optional[Any]							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		str							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=5_0     ,							output_type='np'     ,							width=5_1_2     ,							height=5_1_2     ,							sld_guidance_scale=2_0_0_0     ,							sld_warmup_steps=7     ,							sld_threshold=0.025     ,							sld_momentum_scale=0.5     ,							sld_mom_beta=0.7     ,							)

             a_					:		str							=  output.images
             a_					:		Any							=  image[0, -3:, -3:, -1]
             a_					:		Optional[Any]							=  [0.2383, 0.2276, 0.236, 0.2192, 0.2186, 0.2053, 0.1971, 0.1901, 0.1719]

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

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     )		->			Any:
             a_					:		int							=  StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'     ,							safety_checker=SCREAMING_SNAKE_CASE__     )
             a_					:		Tuple							=  LMSDiscreteScheduler.from_config(sd_pipe.scheduler.config     )
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		int							=  'padme amidala taking a bath artwork, safe for work, no nudity'
             a_					:		Any							=  2_7_3_4_9_7_1_7_5_5
             a_					:		str							=  7

             a_					:		Tuple							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		Dict							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=5_0     ,							output_type='np'     ,							width=5_1_2     ,							height=5_1_2     ,							sld_guidance_scale=0     ,							)

             a_					:		Optional[int]							=  output.images
             a_					:		Any							=  image[0, -3:, -3:, -1]
             a_					:		List[str]							=  [0.3502, 0.3622, 0.3396, 0.3642, 0.3478, 0.3318, 0.35, 0.3348, 0.3297]

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

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

             a_					:		Optional[Any]							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		Any							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=5_0     ,							output_type='np'     ,							width=5_1_2     ,							height=5_1_2     ,							sld_guidance_scale=2_0_0_0     ,							sld_warmup_steps=7     ,							sld_threshold=0.025     ,							sld_momentum_scale=0.5     ,							sld_mom_beta=0.7     ,							)

             a_					:		Optional[Any]							=  output.images
             a_					:		Any							=  image[0, -3:, -3:, -1]
             a_					:		Any							=  [0.5531, 0.5206, 0.4895, 0.5156, 0.5182, 0.4751, 0.4802, 0.4803, 0.4443]

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

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

      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			int:
             a_					:		Dict							=  StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'     )
             a_					:		Optional[int]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Union[str, Any]							=  (
                 'the four horsewomen of the apocalypse, painting by tom of finland, gaston bussiere, craig mullins, j. c.'
                 ' leyendecker'
             )
             a_					:		List[Any]							=  1_0_4_4_3_5_5_2_3_4
             a_					:		int							=  1_2

             a_					:		str							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		List[Any]							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=5_0     ,							output_type='np'     ,							width=5_1_2     ,							height=5_1_2     ,							sld_guidance_scale=0     ,							)

             a_					:		Dict							=  output.images
             a_					:		Optional[int]							=  image[0, -3:, -3:, -1]
             a_					:		int							=  np.array([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]     )

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

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

             a_					:		Union[str, Any]							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  sd_pipe(
                 [prompt]     ,							generator=SCREAMING_SNAKE_CASE__     ,							guidance_scale=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=5_0     ,							output_type='np'     ,							width=5_1_2     ,							height=5_1_2     ,							sld_guidance_scale=2_0_0_0     ,							sld_warmup_steps=7     ,							sld_threshold=0.025     ,							sld_momentum_scale=0.5     ,							sld_mom_beta=0.7     ,							)

             a_					:		Tuple							=  output.images
             a_					:		str							=  image[0, -3:, -3:, -1]
             a_					:		str							=  np.array([0.5818, 0.6285, 0.6835, 0.6019, 0.625, 0.6754, 0.6096, 0.6334, 0.6561]     )
             assert image.shape == (1, 5_1_2, 5_1_2, 3)

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






import jax.numpy as jnp

from ...utils import logging
from ..ta.modeling_flax_ta import FlaxTaEncoderModel, FlaxTaForConditionalGeneration, FlaxTaModel
from .configuration_mta import MTaConfig


UpperCAmelCase_      : Optional[Any]  					=       logging.get_logger(__name__)

UpperCAmelCase_      : str  					=       'T5Config'







def        SCREAMING_SNAKE_CASE_						(			__A :			jnp.array  ,    __A :			int  ,    __A :			int				)	->		jnp.ndarray:

       """simple docstring"""

       a_					:		Dict							=  jnp.zeros_like(__A				)
       a_					:		Dict							=  shifted_input_ids.at[:, 1:].set(input_ids[:, :-1]				)
       a_					:		str							=  shifted_input_ids.at[:, 0].set(__A				)

       a_					:		int							=  jnp.where(shifted_input_ids == -1_00  ,    __A  ,    __A				)
       return shifted_input_ids
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     str        =						'''mt5'''
      snake_case__				:     List[Any]        =						MTaConfig
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     str        =						'''mt5'''
      snake_case__				:     List[str]        =						MTaConfig
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     Any        =						'''mt5'''
      snake_case__				:     Union[str, Any]        =						MTaConfig


