
"""simple docstring"""






from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)




@flax.struct.dataclass
class 			_SCREAMING_SNAKE_CASE					(			A__							):
			UpperCAmelCase_  :jnp.ndarray
			UpperCAmelCase_  :jnp.ndarray




class 			_SCREAMING_SNAKE_CASE					(			nn.Module							):
			UpperCAmelCase_  :int
			UpperCAmelCase_  :Tuple[int]				 =						(16, 32, 96, 256)
			UpperCAmelCase_  :jnp.dtype				 =						jnp.floataa






			def  __lowerCAmelCase (					self       )			->     Dict:
						lowerCAmelCase_    :Optional[int]			     =  nn.Conv(
						    self.block_out_channels[0]       ,   kernel_size=(3, 3)       ,   padding=((1, 1), (1, 1))       ,   dtype=self.dtype       ,   )

						lowerCAmelCase_    :int			     =  []
						for i in range(len(self.block_out_channels       ) - 1       ):
									lowerCAmelCase_    :Union[str, Any]			     =  self.block_out_channels[i]
									lowerCAmelCase_    :Optional[int]			     =  self.block_out_channels[i + 1]
									lowerCAmelCase_    :int			     =  nn.Conv(
									    __A       ,   kernel_size=(3, 3)       ,   padding=((1, 1), (1, 1))       ,   dtype=self.dtype       ,   )
									blocks.append(__A       )
									lowerCAmelCase_    :List[str]			     =  nn.Conv(
									    __A       ,   kernel_size=(3, 3)       ,   strides=(2, 2)       ,   padding=((1, 1), (1, 1))       ,   dtype=self.dtype       ,   )
									blocks.append(__A       )
						lowerCAmelCase_    :Optional[int]			     =  blocks

						lowerCAmelCase_    :int			     =  nn.Conv(
						    self.conditioning_embedding_channels       ,   kernel_size=(3, 3)       ,   padding=((1, 1), (1, 1))       ,   kernel_init=nn.initializers.zeros_init()       ,   bias_init=nn.initializers.zeros_init()       ,   dtype=self.dtype       ,   )






			def __call__(					self       ,   __A       )			->     Tuple:
						lowerCAmelCase_    :Dict			     =  self.conv_in(__A       )
						lowerCAmelCase_    :List[str]			     =  nn.silu(__A       )

						for block in self.blocks:
									lowerCAmelCase_    :Any			     =  block(__A       )
									lowerCAmelCase_    :Optional[int]			     =  nn.silu(__A       )

						lowerCAmelCase_    :List[Any]			     =  self.conv_out(__A       )

						return embedding




@flax_register_to_config
class 			_SCREAMING_SNAKE_CASE					(			nn.Module	, A__	, A__							):
			UpperCAmelCase_  :int				 =						32
			UpperCAmelCase_  :int				 =						4
			UpperCAmelCase_  :Tuple[str]				 =						(
			    "CrossAttnDownBlock2D",
			    "CrossAttnDownBlock2D",
			    "CrossAttnDownBlock2D",
			    "DownBlock2D",
			)
			UpperCAmelCase_  :Union[bool, Tuple[bool]]				 =						False
			UpperCAmelCase_  :Tuple[int]				 =						(320, 640, 1280, 1280)
			UpperCAmelCase_  :int				 =						2
			UpperCAmelCase_  :Union[int, Tuple[int]]				 =						8
			UpperCAmelCase_  :Optional[Union[int, Tuple[int]]]				 =						None
			UpperCAmelCase_  :int				 =						1280
			UpperCAmelCase_  :float				 =						0.0
			UpperCAmelCase_  :bool				 =						False
			UpperCAmelCase_  :jnp.dtype				 =						jnp.floataa
			UpperCAmelCase_  :bool				 =						True
			UpperCAmelCase_  :int				 =						0
			UpperCAmelCase_  :str				 =						"rgb"
			UpperCAmelCase_  :Tuple[int]				 =						(16, 32, 96, 256)






			def  __lowerCAmelCase (					self       ,   __A       )			->     FrozenDict:
						# init input tensors
						lowerCAmelCase_    :Optional[int]			     =  (1, self.in_channels, self.sample_size, self.sample_size)
						lowerCAmelCase_    :Dict			     =  jnp.zeros(__A       ,   dtype=jnp.floataa       )
						lowerCAmelCase_    :List[Any]			     =  jnp.ones((1,)       ,   dtype=jnp.intaa       )
						lowerCAmelCase_    :Optional[int]			     =  jnp.zeros((1, 1, self.cross_attention_dim)       ,   dtype=jnp.floataa       )
						lowerCAmelCase_    :Any			     =  (1, 3, self.sample_size * 8, self.sample_size * 8)
						lowerCAmelCase_    :Optional[int]			     =  jnp.zeros(__A       ,   dtype=jnp.floataa       )

						lowerCAmelCase_     ,       lowerCAmelCase_    :Optional[int]			     =  jax.random.split(__A       )
						lowerCAmelCase_    :Optional[int]			     =  {"""params""": params_rng, """dropout""": dropout_rng}

						return self.init(__A       ,   __A       ,   __A       ,   __A       ,   __A       )["params"]






			def  __lowerCAmelCase (					self       )			->     List[str]:
						lowerCAmelCase_    :Union[str, Any]			     =  self.block_out_channels
						lowerCAmelCase_    :int			     =  block_out_channels[0] * 4

						# If `num_attention_heads` is not defined (which is the case for most models)
						# it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
						# The reason for this behavior is to correct for incorrectly named variables that were introduced
						# when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
						# Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
						# which is why we correct for the naming here.
						lowerCAmelCase_    :Dict			     =  self.num_attention_heads or self.attention_head_dim

						# input
						lowerCAmelCase_    :int			     =  nn.Conv(
						    block_out_channels[0]       ,   kernel_size=(3, 3)       ,   strides=(1, 1)       ,   padding=((1, 1), (1, 1))       ,   dtype=self.dtype       ,   )

						# time
						lowerCAmelCase_    :Optional[Any]			     =  FlaxTimesteps(
						    block_out_channels[0]       ,   flip_sin_to_cos=self.flip_sin_to_cos       ,   freq_shift=self.config.freq_shift       )
						lowerCAmelCase_    :Optional[Any]			     =  FlaxTimestepEmbedding(__A       ,   dtype=self.dtype       )

						lowerCAmelCase_    :int			     =  FlaxControlNetConditioningEmbedding(
						    conditioning_embedding_channels=block_out_channels[0]       ,   block_out_channels=self.conditioning_embedding_out_channels       ,   )

						lowerCAmelCase_    :List[str]			     =  self.only_cross_attention
						if isinstance(__A       ,   __A       ):
									lowerCAmelCase_    :List[str]			     =  (only_cross_attention,) * len(self.down_block_types       )

						if isinstance(__A       ,   __A       ):
									lowerCAmelCase_    :Optional[Any]			     =  (num_attention_heads,) * len(self.down_block_types       )

						# down
						lowerCAmelCase_    :Dict			     =  []
						lowerCAmelCase_    :Optional[Any]			     =  []

						lowerCAmelCase_    :Dict			     =  block_out_channels[0]

						lowerCAmelCase_    :List[Any]			     =  nn.Conv(
						    __A       ,   kernel_size=(1, 1)       ,   padding="""VALID"""       ,   kernel_init=nn.initializers.zeros_init()       ,   bias_init=nn.initializers.zeros_init()       ,   dtype=self.dtype       ,   )
						controlnet_down_blocks.append(__A       )

						for i, down_block_type in enumerate(self.down_block_types       ):
									lowerCAmelCase_    :List[Any]			     =  output_channel
									lowerCAmelCase_    :List[str]			     =  block_out_channels[i]
									lowerCAmelCase_    :Tuple			     =  i == len(__A       ) - 1

									if down_block_type == "CrossAttnDownBlock2D":
												lowerCAmelCase_    :Tuple			     =  FlaxCrossAttnDownBlockaD(
												    in_channels=__A       ,   out_channels=__A       ,   dropout=self.dropout       ,   num_layers=self.layers_per_block       ,   num_attention_heads=num_attention_heads[i]       ,   add_downsample=not is_final_block       ,   use_linear_projection=self.use_linear_projection       ,   only_cross_attention=only_cross_attention[i]       ,   dtype=self.dtype       ,   )
									else:
												lowerCAmelCase_    :Optional[int]			     =  FlaxDownBlockaD(
												    in_channels=__A       ,   out_channels=__A       ,   dropout=self.dropout       ,   num_layers=self.layers_per_block       ,   add_downsample=not is_final_block       ,   dtype=self.dtype       ,   )

									down_blocks.append(__A       )

									for _ in range(self.layers_per_block       ):
												lowerCAmelCase_    :List[str]			     =  nn.Conv(
												    __A       ,   kernel_size=(1, 1)       ,   padding="""VALID"""       ,   kernel_init=nn.initializers.zeros_init()       ,   bias_init=nn.initializers.zeros_init()       ,   dtype=self.dtype       ,   )
												controlnet_down_blocks.append(__A       )

									if not is_final_block:
												lowerCAmelCase_    :str			     =  nn.Conv(
												    __A       ,   kernel_size=(1, 1)       ,   padding="""VALID"""       ,   kernel_init=nn.initializers.zeros_init()       ,   bias_init=nn.initializers.zeros_init()       ,   dtype=self.dtype       ,   )
												controlnet_down_blocks.append(__A       )

						lowerCAmelCase_    :List[Any]			     =  down_blocks
						lowerCAmelCase_    :Optional[Any]			     =  controlnet_down_blocks

						# mid
						lowerCAmelCase_    :int			     =  block_out_channels[-1]
						lowerCAmelCase_    :List[Any]			     =  FlaxUNetMidBlockaDCrossAttn(
						    in_channels=__A       ,   dropout=self.dropout       ,   num_attention_heads=num_attention_heads[-1]       ,   use_linear_projection=self.use_linear_projection       ,   dtype=self.dtype       ,   )

						lowerCAmelCase_    :Dict			     =  nn.Conv(
						    __A       ,   kernel_size=(1, 1)       ,   padding="""VALID"""       ,   kernel_init=nn.initializers.zeros_init()       ,   bias_init=nn.initializers.zeros_init()       ,   dtype=self.dtype       ,   )






			def __call__(					self       ,   __A       ,   __A       ,   __A       ,   __A       ,   __A = 1.0       ,   __A = True       ,   __A = False       ,   )			->     Union[FlaxControlNetOutput, Tuple]:
						lowerCAmelCase_    :Union[str, Any]			     =  self.controlnet_conditioning_channel_order
						if channel_order == "bgr":
									lowerCAmelCase_    :Optional[int]			     =  jnp.flip(__A       ,   axis=1       )

						# 1. time
						if not isinstance(__A       ,   jnp.ndarray       ):
									lowerCAmelCase_    :List[str]			     =  jnp.array([timesteps]       ,   dtype=jnp.intaa       )
						elif isinstance(__A       ,   jnp.ndarray       ) and len(timesteps.shape       ) == 0:
									lowerCAmelCase_    :str			     =  timesteps.astype(dtype=jnp.floataa       )
									lowerCAmelCase_    :Union[str, Any]			     =  jnp.expand_dims(__A       ,   0       )

						lowerCAmelCase_    :List[Any]			     =  self.time_proj(__A       )
						lowerCAmelCase_    :Optional[Any]			     =  self.time_embedding(__A       )

						# 2. pre-process
						lowerCAmelCase_    :int			     =  jnp.transpose(__A       ,   (0, 2, 3, 1)       )
						lowerCAmelCase_    :List[Any]			     =  self.conv_in(__A       )

						lowerCAmelCase_    :Union[str, Any]			     =  jnp.transpose(__A       ,   (0, 2, 3, 1)       )
						lowerCAmelCase_    :List[str]			     =  self.controlnet_cond_embedding(__A       )
						sample += controlnet_cond

						# 3. down
						lowerCAmelCase_    :Any			     =  (sample,)
						for down_block in self.down_blocks:
									if isinstance(__A       ,   __A       ):
												lowerCAmelCase_     ,       lowerCAmelCase_    :Any			     =  down_block(__A       ,   __A       ,   __A       ,   deterministic=not train       )
									else:
												lowerCAmelCase_     ,       lowerCAmelCase_    :Union[str, Any]			     =  down_block(__A       ,   __A       ,   deterministic=not train       )
									down_block_res_samples += res_samples

						# 4. mid
						lowerCAmelCase_    :int			     =  self.mid_block(__A       ,   __A       ,   __A       ,   deterministic=not train       )

						# 5. contronet blocks
						lowerCAmelCase_    :Dict			     =  ()
						for down_block_res_sample, controlnet_block in zip(__A       ,   self.controlnet_down_blocks       ):
									lowerCAmelCase_    :Union[str, Any]			     =  controlnet_block(__A       )
									controlnet_down_block_res_samples += (down_block_res_sample,)

						lowerCAmelCase_    :Optional[Any]			     =  controlnet_down_block_res_samples

						lowerCAmelCase_    :List[Any]			     =  self.controlnet_mid_block(__A       )

						# 6. scaling
						lowerCAmelCase_    :List[Any]			     =  [sample * conditioning_scale for sample in down_block_res_samples]
						mid_block_res_sample *= conditioning_scale

						if not return_dict:
									return (down_block_res_samples, mid_block_res_sample)

						return FlaxControlNetOutput(
						    down_block_res_samples=__A       ,   mid_block_res_sample=__A       )









"""simple docstring"""





from __future__ import annotations

import unittest

from transformers import XGLMConfig, XGLMTokenizer, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
   import tensorflow as tf

   from transformers.models.xglm.modeling_tf_xglm import (
       TF_XGLM_PRETRAINED_MODEL_ARCHIVE_LIST,
       TFXGLMForCausalLM,
       TFXGLMModel,
   )




@require_tf
class 					_lowerCAmelCase  :







 """simple docstring"""







 __UpperCAmelCase	:					Tuple   		=	XGLMConfig
 __UpperCAmelCase	:					Optional[Any]   		=	{}
 __UpperCAmelCase	:					Union[str, Any]   		=	"gelu"


 def __init__(			self							:			Optional[int],					UpperCAmelCase__							:			List[str],					UpperCAmelCase__							:			Optional[int]=1_4,					UpperCAmelCase__							:			str=7,					UpperCAmelCase__							:			Optional[Any]=True,					UpperCAmelCase__							:			List[Any]=True,					UpperCAmelCase__							:			int=True,					UpperCAmelCase__							:			List[str]=9_9,					UpperCAmelCase__							:			Union[str, Any]=3_2,					UpperCAmelCase__							:			Union[str, Any]=2,					UpperCAmelCase__							:			Union[str, Any]=4,					UpperCAmelCase__							:			Tuple=3_7,					UpperCAmelCase__							:			List[Any]="gelu",					UpperCAmelCase__							:			List[str]=0.1,					UpperCAmelCase__							:			Optional[int]=0.1,					UpperCAmelCase__							:			Tuple=5_1_2,					UpperCAmelCase__							:			Optional[Any]=0.02,					):
    __lowercase								=						parent
    __lowercase								=						batch_size
    __lowercase								=						seq_length
    __lowercase								=						is_training
    __lowercase								=						use_input_mask
    __lowercase								=						use_labels
    __lowercase								=						vocab_size
    __lowercase								=						d_model
    __lowercase								=						num_hidden_layers
    __lowercase								=						num_attention_heads
    __lowercase								=						ffn_dim
    __lowercase								=						activation_function
    __lowercase								=						activation_dropout
    __lowercase								=						attention_dropout
    __lowercase								=						max_position_embeddings
    __lowercase								=						initializer_range
    __lowercase								=						None
    __lowercase								=						0
    __lowercase								=						2
    __lowercase								=						1


 def 					_lowercase (			self							:			Union[str, Any]	):
    return XGLMConfig.from_pretrained("facebook/xglm-564M"	)


 def 					_lowercase (			self							:			Tuple	):
    __lowercase								=						tf.clip_by_value(
        ids_tensor([self.batch_size, self.seq_length],					self.vocab_size	),					clip_value_min=0,					clip_value_max=3	)

    __lowercase								=						None
    if self.use_input_mask:
       __lowercase								=						random_attention_mask([self.batch_size, self.seq_length]	)

    __lowercase								=						self.get_config()

    __lowercase								=						floats_tensor([self.num_hidden_layers, self.num_attention_heads],					2	)

    return (
        config,
        input_ids,
        input_mask,
        head_mask,
    )


 def 					_lowercase (			self							:			List[Any]	):
    return XGLMConfig(
        vocab_size=self.vocab_size,					d_model=self.hidden_size,					num_layers=self.num_hidden_layers,					attention_heads=self.num_attention_heads,					ffn_dim=self.ffn_dim,					activation_function=self.activation_function,					activation_dropout=self.activation_dropout,					attention_dropout=self.attention_dropout,					max_position_embeddings=self.max_position_embeddings,					initializer_range=self.initializer_range,					use_cache=UpperCAmelCase__,					bos_token_id=self.bos_token_id,					eos_token_id=self.eos_token_id,					pad_token_id=self.pad_token_id,					return_dict=UpperCAmelCase__,					)



 def 					_lowercase (			self							:			Dict	):
    __lowercase								=						self.prepare_config_and_inputs()

    (
        (
        __lowercase
    )		,(
        __lowercase
    )		,(
        __lowercase
    )		,(
        __lowercase
    )		,
    )								=						config_and_inputs

    __lowercase								=						{
        "input_ids": input_ids,
        "head_mask": head_mask,
    }

    return config, inputs_dict







@require_tf
class 					_lowerCAmelCase  (	lowercase     ,lowercase     ,unittest.TestCase   ):







 """simple docstring"""







 __UpperCAmelCase	:					Union[str, Any]   		=	(TFXGLMModel, TFXGLMForCausalLM) if is_tf_available() else ()
 __UpperCAmelCase	:					List[str]   		=	(TFXGLMForCausalLM,) if is_tf_available() else ()
 __UpperCAmelCase	:					Any   		=	(
     {"feature-extraction": TFXGLMModel, "text-generation": TFXGLMForCausalLM} if is_tf_available() else {}
 )
 __UpperCAmelCase	:					Optional[Any]   		=	False
 __UpperCAmelCase	:					List[str]   		=	False
 __UpperCAmelCase	:					int   		=	False


 def 					_lowercase (			self							:			Optional[Any]	):
    __lowercase								=						TFXGLMModelTester(self	)
    __lowercase								=						ConfigTester(self,					config_class=UpperCAmelCase__,					n_embd=3_7	)


 def 					_lowercase (			self							:			Any	):
    self.config_tester.run_common_tests()


 @slow
 def 					_lowercase (			self							:			List[str]	):
    for model_name in TF_XGLM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
       __lowercase								=						TFXGLMModel.from_pretrained(UpperCAmelCase__	)
       self.assertIsNotNone(UpperCAmelCase__	)



 @unittest.skip(reason="Currently, model embeddings are going to undergo a major refactor."	)
 def 					_lowercase (			self							:			int	):
    super().test_resize_token_embeddings()







@require_tf
class 					_lowerCAmelCase  (	unittest.TestCase   ):







 """simple docstring"""







 @slow
 def 					_lowercase (			self							:			Dict,					UpperCAmelCase__							:			Optional[int]=True	):
    __lowercase								=						TFXGLMForCausalLM.from_pretrained("facebook/xglm-564M"	)
    __lowercase								=						tf.convert_to_tensor([[2, 2_6_8, 9_8_6_5]],					dtype=tf.intaa	)  # The dog
    # </s> The dog is a very friendly dog. He is very affectionate and loves to play with other
    # fmt: off
    __lowercase								=						[2, 2_6_8, 9_8_6_5, 6_7, 1_1, 1_9_8_8, 5_7_2_5_2, 9_8_6_5, 5, 9_8_4, 6_7, 1_9_8_8, 2_1_3_8_3_8, 1_6_5_8, 5_3, 7_0_4_4_6, 3_3, 6_6_5_7, 2_7_8, 1_5_8_1]
    # fmt: on
    __lowercase								=						model.generate(UpperCAmelCase__,					do_sample=UpperCAmelCase__,					num_beams=1	)
    if verify_outputs:
       self.assertListEqual(output_ids[0].numpy().tolist(),					UpperCAmelCase__	)


 @slow
 def 					_lowercase (			self							:			List[Any]	):
    __lowercase								=						XGLMTokenizer.from_pretrained("facebook/xglm-564M"	)
    __lowercase								=						TFXGLMForCausalLM.from_pretrained("facebook/xglm-564M"	)

    tf.random.set_seed(0	)
    __lowercase								=						tokenizer("Today is a nice day and",					return_tensors="tf"	)
    __lowercase								=						tokenized.input_ids
    # forces the generation to happen on CPU, to avoid GPU-related quirks (and assure same output regardless of the available devices)
    with tf.device(":/CPU:0"	):
       __lowercase								=						model.generate(UpperCAmelCase__,					do_sample=UpperCAmelCase__,					seed=[7, 0]	)
    __lowercase								=						tokenizer.decode(output_ids[0],					skip_special_tokens=UpperCAmelCase__	)

    __lowercase								=						(
        "Today is a nice day and warm evening here over Southern Alberta!! Today when they closed schools due"
    )
    self.assertEqual(UpperCAmelCase__,					UpperCAmelCase__	)



 @slow
 def 					_lowercase (			self							:			Dict	):
    __lowercase								=						TFXGLMForCausalLM.from_pretrained("facebook/xglm-564M"	)
    __lowercase								=						XGLMTokenizer.from_pretrained("facebook/xglm-564M"	)

    __lowercase								=						"left"

    # use different length sentences to test batching
    __lowercase								=						[
        "This is an extremelly long sentence that only exists to test the ability of the model to cope with "
        "left-padding, such as in batched generation. The output for the sequence below should be the same "
        "regardless of whether left padding is applied or not. When",
        "Hello, my dog is a little",
    ]

    __lowercase								=						tokenizer(UpperCAmelCase__,					return_tensors="tf",					padding=UpperCAmelCase__	)
    __lowercase								=						inputs["input_ids"]

    __lowercase								=						model.generate(input_ids=UpperCAmelCase__,					attention_mask=inputs["attention_mask"],					max_new_tokens=1_2	)

    __lowercase								=						tokenizer(sentences[0],					return_tensors="tf"	).input_ids
    __lowercase								=						model.generate(input_ids=UpperCAmelCase__,					max_new_tokens=1_2	)

    __lowercase								=						tokenizer(sentences[1],					return_tensors="tf"	).input_ids
    __lowercase								=						model.generate(input_ids=UpperCAmelCase__,					max_new_tokens=1_2	)

    __lowercase								=						tokenizer.batch_decode(UpperCAmelCase__,					skip_special_tokens=UpperCAmelCase__	)
    __lowercase								=						tokenizer.decode(output_non_padded[0],					skip_special_tokens=UpperCAmelCase__	)
    __lowercase								=						tokenizer.decode(output_padded[0],					skip_special_tokens=UpperCAmelCase__	)

    __lowercase								=						[
        "This is an extremelly long sentence that only exists to test the ability of the model to cope with "
        "left-padding, such as in batched generation. The output for the sequence below should be the same "
        "regardless of whether left padding is applied or not. When left padding is applied, the sequence will be "
        "a single",
        "Hello, my dog is a little bit of a shy one, but he is very friendly",
    ]
    self.assertListEqual(UpperCAmelCase__,					UpperCAmelCase__	)
    self.assertListEqual(UpperCAmelCase__,					[non_padded_sentence, padded_sentence]	)




