


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        snake_case_   (UpperCamelCase__      ):
					UpperCAmelCase__					:    jnp.ndarray
					UpperCAmelCase__					:    jnp.ndarray







class        snake_case_   (nn.Module      ):
					UpperCAmelCase__					:    int
					UpperCAmelCase__					:    Tuple[int]          =			(1_6, 3_2, 9_6, 2_5_6)
					UpperCAmelCase__					:    jnp.dtype          =			jnp.floataa
					def  lowerCamelCase__(      self   :List[Any]      )					->    Optional[int]:
												a__             =     nn.Conv(
												    self.block_out_channels[0]			,kernel_size=(3, 3)			,padding=((1, 1), (1, 1))			,dtype=self.dtype			,)

												a__             =     []
												for i in range(len(self.block_out_channels      ) - 1      ):
																			a__             =     self.block_out_channels[i]
																			a__             =     self.block_out_channels[i + 1]
																			a__             =     nn.Conv(
																			    UpperCamelCase_			,kernel_size=(3, 3)			,padding=((1, 1), (1, 1))			,dtype=self.dtype			,)
																			blocks.append(UpperCamelCase_      )
																			a__             =     nn.Conv(
																			    UpperCamelCase_			,kernel_size=(3, 3)			,strides=(2, 2)			,padding=((1, 1), (1, 1))			,dtype=self.dtype			,)
																			blocks.append(UpperCamelCase_      )
												a__             =     blocks

												a__             =     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   :Optional[int]			,__snake_case   :Optional[int]      )					->    List[Any]:
												a__             =     self.conv_in(UpperCamelCase_      )
												a__             =     nn.silu(UpperCamelCase_      )

												for block in self.blocks:
																			a__             =     block(UpperCamelCase_      )
																			a__             =     nn.silu(UpperCamelCase_      )

												a__             =     self.conv_out(UpperCamelCase_      )

												return embedding







@flax_register_to_config
class        snake_case_   (nn.Module	,						UpperCamelCase__	,						UpperCamelCase__      ):
					UpperCAmelCase__					:    int          =			3_2
					UpperCAmelCase__					:    int          =			4
					UpperCAmelCase__					:    Tuple[str]          =			(
					    "CrossAttnDownBlock2D",
					    "CrossAttnDownBlock2D",
					    "CrossAttnDownBlock2D",
					    "DownBlock2D",
					)
					UpperCAmelCase__					:    Union[bool, Tuple[bool]]          =			False
					UpperCAmelCase__					:    Tuple[int]          =			(3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
					UpperCAmelCase__					:    int          =			2
					UpperCAmelCase__					:    Union[int, Tuple[int]]          =			8
					UpperCAmelCase__					:    Optional[Union[int, Tuple[int]]]          =			None
					UpperCAmelCase__					:    int          =			1_2_8_0
					UpperCAmelCase__					:    float          =			0.0
					UpperCAmelCase__					:    bool          =			False
					UpperCAmelCase__					:    jnp.dtype          =			jnp.floataa
					UpperCAmelCase__					:    bool          =			True
					UpperCAmelCase__					:    int          =			0
					UpperCAmelCase__					:    str          =			"rgb"
					UpperCAmelCase__					:    Tuple[int]          =			(1_6, 3_2, 9_6, 2_5_6)
					def  lowerCamelCase__(      self   :List[Any]			,__snake_case   :jax.random.KeyArray      )					->    FrozenDict:
												a__             =     (1, self.in_channels, self.sample_size, self.sample_size)
												a__             =     jnp.zeros(UpperCamelCase_			,dtype=jnp.floataa      )
												a__             =     jnp.ones((1,)			,dtype=jnp.intaa      )
												a__             =     jnp.zeros((1, 1, self.cross_attention_dim)			,dtype=jnp.floataa      )
												a__             =     (1, 3, self.sample_size * 8, self.sample_size * 8)
												a__             =     jnp.zeros(UpperCamelCase_			,dtype=jnp.floataa      )

												a__      ,					a__             =     jax.random.split(UpperCamelCase_      )
												a__             =     {'params': params_rng, 'dropout': dropout_rng}

												return self.init(UpperCamelCase_			,UpperCamelCase_			,UpperCamelCase_			,UpperCamelCase_			,UpperCamelCase_      )["params"]
					def  lowerCamelCase__(      self   :Optional[int]      )					->    List[str]:
												a__             =     self.block_out_channels
												a__             =     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.
												a__             =     self.num_attention_heads or self.attention_head_dim

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

												# time
												a__             =     FlaxTimesteps(
												    block_out_channels[0]			,flip_sin_to_cos=self.flip_sin_to_cos			,freq_shift=self.config.freq_shift      )
												a__             =     FlaxTimestepEmbedding(UpperCamelCase_			,dtype=self.dtype      )

												a__             =     FlaxControlNetConditioningEmbedding(
												    conditioning_embedding_channels=block_out_channels[0]			,block_out_channels=self.conditioning_embedding_out_channels			,)

												a__             =     self.only_cross_attention
												if isinstance(UpperCamelCase_			,UpperCamelCase_      ):
																			a__             =     (only_cross_attention,) * len(self.down_block_types      )

												if isinstance(UpperCamelCase_			,UpperCamelCase_      ):
																			a__             =     (num_attention_heads,) * len(self.down_block_types      )

												# down
												a__             =     []
												a__             =     []

												a__             =     block_out_channels[0]

												a__             =     nn.Conv(
												    UpperCamelCase_			,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(UpperCamelCase_      )

												for i, down_block_type in enumerate(self.down_block_types      ):
																			a__             =     output_channel
																			a__             =     block_out_channels[i]
																			a__             =     i == len(UpperCamelCase_      ) - 1

																			if down_block_type == "CrossAttnDownBlock2D":
																										a__             =     FlaxCrossAttnDownBlockaD(
																										    in_channels=UpperCamelCase_			,out_channels=UpperCamelCase_			,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:
																										a__             =     FlaxDownBlockaD(
																										    in_channels=UpperCamelCase_			,out_channels=UpperCamelCase_			,dropout=self.dropout			,num_layers=self.layers_per_block			,add_downsample=not is_final_block			,dtype=self.dtype			,)

																			down_blocks.append(UpperCamelCase_      )

																			for _ in range(self.layers_per_block      ):
																										a__             =     nn.Conv(
																										    UpperCamelCase_			,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(UpperCamelCase_      )

																			if not is_final_block:
																										a__             =     nn.Conv(
																										    UpperCamelCase_			,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(UpperCamelCase_      )

												a__             =     down_blocks
												a__             =     controlnet_down_blocks

												# mid
												a__             =     block_out_channels[-1]
												a__             =     FlaxUNetMidBlockaDCrossAttn(
												    in_channels=UpperCamelCase_			,dropout=self.dropout			,num_attention_heads=num_attention_heads[-1]			,use_linear_projection=self.use_linear_projection			,dtype=self.dtype			,)

												a__             =     nn.Conv(
												    UpperCamelCase_			,kernel_size=(1, 1)			,padding='VALID'			,kernel_init=nn.initializers.zeros_init()			,bias_init=nn.initializers.zeros_init()			,dtype=self.dtype			,)



					def __call__(      self   :Dict			,__snake_case   :Union[str, Any]			,__snake_case   :Any			,__snake_case   :Optional[Any]			,__snake_case   :List[Any]			,__snake_case   :float = 1.0			,__snake_case   :bool = True			,__snake_case   :bool = False			,)					->    Union[FlaxControlNetOutput, Tuple]:
												a__             =     self.controlnet_conditioning_channel_order
												if channel_order == "bgr":
																			a__             =     jnp.flip(UpperCamelCase_			,axis=1      )

												# 1. time
												if not isinstance(UpperCamelCase_			,jnp.ndarray      ):
																			a__             =     jnp.array([timesteps]			,dtype=jnp.intaa      )
												elif isinstance(UpperCamelCase_			,jnp.ndarray      ) and len(timesteps.shape      ) == 0:
																			a__             =     timesteps.astype(dtype=jnp.floataa      )
																			a__             =     jnp.expand_dims(UpperCamelCase_			,0      )

												a__             =     self.time_proj(UpperCamelCase_      )
												a__             =     self.time_embedding(UpperCamelCase_      )

												# 2. pre-process
												a__             =     jnp.transpose(UpperCamelCase_			,(0, 2, 3, 1)      )
												a__             =     self.conv_in(UpperCamelCase_      )

												a__             =     jnp.transpose(UpperCamelCase_			,(0, 2, 3, 1)      )
												a__             =     self.controlnet_cond_embedding(UpperCamelCase_      )
												sample += controlnet_cond

												# 3. down
												a__             =     (sample,)
												for down_block in self.down_blocks:
																			if isinstance(UpperCamelCase_			,UpperCamelCase_      ):
																										a__      ,					a__             =     down_block(UpperCamelCase_			,UpperCamelCase_			,UpperCamelCase_			,deterministic=not train      )
																			else:
																										a__      ,					a__             =     down_block(UpperCamelCase_			,UpperCamelCase_			,deterministic=not train      )
																			down_block_res_samples += res_samples

												# 4. mid
												a__             =     self.mid_block(UpperCamelCase_			,UpperCamelCase_			,UpperCamelCase_			,deterministic=not train      )

												# 5. contronet blocks
												a__             =     ()
												for down_block_res_sample, controlnet_block in zip(UpperCamelCase_			,self.controlnet_down_blocks      ):
																			a__             =     controlnet_block(UpperCamelCase_      )
																			controlnet_down_block_res_samples += (down_block_res_sample,)

												a__             =     controlnet_down_block_res_samples

												a__             =     self.controlnet_mid_block(UpperCamelCase_      )

												# 6. scaling
												a__             =     [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=UpperCamelCase_			,mid_block_res_sample=UpperCamelCase_      )








import inspect
import unittest

from transformers import DecisionTransformerConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
			import torch

			from transformers import DecisionTransformerModel
			from transformers.models.decision_transformer.modeling_decision_transformer import (
			    DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
			)


class        _a							:

							def __init__(							self:	Tuple    ,						UpperCamelCase_:	int    ,						UpperCamelCase_:	Any=13    ,						UpperCamelCase_:	Optional[Any]=7    ,						UpperCamelCase_:	Optional[Any]=6    ,						UpperCamelCase_:	Any=17    ,						UpperCamelCase_:	str=23    ,						UpperCamelCase_:	List[Any]=11    ,						UpperCamelCase_:	Optional[int]=True    ,						)			->   List[Any]:





									"""simple docstring"""







									lowercase__            =      parent
									lowercase__            =      batch_size
									lowercase__            =      seq_length
									lowercase__            =      act_dim
									lowercase__            =      state_dim
									lowercase__            =      hidden_size
									lowercase__            =      max_length
									lowercase__            =      is_training

							def       lowerCamelCase_ (							self:	Optional[Any]							)			->   Tuple:





									"""simple docstring"""







									lowercase__            =      floats_tensor((self.batch_size, self.seq_length, self.state_dim)							)
									lowercase__            =      floats_tensor((self.batch_size, self.seq_length, self.act_dim)							)
									lowercase__            =      floats_tensor((self.batch_size, self.seq_length, 1)							)
									lowercase__            =      floats_tensor((self.batch_size, self.seq_length, 1)							)
									lowercase__            =      ids_tensor((self.batch_size, self.seq_length)    ,						vocab_size=1_000							)
									lowercase__            =      random_attention_mask((self.batch_size, self.seq_length)							)

									lowercase__            =      self.get_config()

									return (
									    config,
									    states,
									    actions,
									    rewards,
									    returns_to_go,
									    timesteps,
									    attention_mask,
									)

							def       lowerCamelCase_ (							self:	Dict							)			->   List[Any]:





									"""simple docstring"""







									return DecisionTransformerConfig(
									    batch_size=self.batch_size    ,						seq_length=self.seq_length    ,						act_dim=self.act_dim    ,						state_dim=self.state_dim    ,						hidden_size=self.hidden_size    ,						max_length=self.max_length    ,						)

							def       lowerCamelCase_ (							self:	List[str]    ,						UpperCamelCase_:	int    ,						UpperCamelCase_:	List[str]    ,						UpperCamelCase_:	Any    ,						UpperCamelCase_:	List[Any]    ,						UpperCamelCase_:	Any    ,						UpperCamelCase_:	List[Any]    ,						UpperCamelCase_:	int    ,						)			->   Dict:





									"""simple docstring"""







									lowercase__            =      DecisionTransformerModel(config=UpperCamelCase_							)
									model.to(UpperCamelCase_							)
									model.eval()
									lowercase__            =      model(UpperCamelCase_    ,						UpperCamelCase_    ,						UpperCamelCase_    ,						UpperCamelCase_    ,						UpperCamelCase_    ,						UpperCamelCase_							)

									self.parent.assertEqual(result.state_preds.shape    ,						states.shape							)
									self.parent.assertEqual(result.action_preds.shape    ,						actions.shape							)
									self.parent.assertEqual(result.return_preds.shape    ,						returns_to_go.shape							)
									self.parent.assertEqual(
									    result.last_hidden_state.shape    ,						(self.batch_size, self.seq_length * 3, self.hidden_size)							)  # seq length *3 as there are 3 modelities: states, returns and actions


							def       lowerCamelCase_ (							self:	List[Any]							)			->   str:





									"""simple docstring"""







									lowercase__            =      self.prepare_config_and_inputs()
									(
									    (
									    lowercase__
									)     ,		(
									    lowercase__
									)     ,		(
									    lowercase__
									)     ,		(
									    lowercase__
									)     ,		(
									    lowercase__
									)     ,		(
									    lowercase__
									)     ,		(
									    lowercase__
									)     ,		
									)            =      config_and_inputs
									lowercase__            =      {
									    '''states''': states,
									    '''actions''': actions,
									    '''rewards''': rewards,
									    '''returns_to_go''': returns_to_go,
									    '''timesteps''': timesteps,
									    '''attention_mask''': attention_mask,
									}
									return config, inputs_dict


@require_torch
class        _a							(			UpperCamelCase__     ,	UpperCamelCase__     ,	UpperCamelCase__     ,	unittest.TestCase			):
							_lowercase	:			str        = (DecisionTransformerModel,) if is_torch_available() else ()
							_lowercase	:			List[str]        = ()
							_lowercase	:			List[Any]        = {'''feature-extraction''': DecisionTransformerModel} if is_torch_available() else {}

							# Ignoring of a failing test from GenerationTesterMixin, as the model does not use inputs_ids
							_lowercase	:			Any        = False

							# Ignoring of a failing tests from ModelTesterMixin, as the model does not implement these features
							_lowercase	:			Tuple        = False
							_lowercase	:			str        = False
							_lowercase	:			Tuple        = False
							_lowercase	:			Optional[Any]        = False
							_lowercase	:			Tuple        = False
							_lowercase	:			Dict        = False
							_lowercase	:			Tuple        = False
							_lowercase	:			Optional[Any]        = False
							_lowercase	:			Optional[int]        = False

							def       lowerCamelCase_ (							self:	Union[str, Any]							)			->   Optional[Any]:





									"""simple docstring"""







									lowercase__            =      DecisionTransformerModelTester(self							)
									lowercase__            =      ConfigTester(self    ,						config_class=UpperCamelCase_    ,						hidden_size=37							)

							def       lowerCamelCase_ (							self:	Any							)			->   Optional[Any]:





									"""simple docstring"""







									self.config_tester.run_common_tests()

							def       lowerCamelCase_ (							self:	int							)			->   Optional[int]:





									"""simple docstring"""







									lowercase__            =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*UpperCamelCase_							)

							@slow
							def       lowerCamelCase_ (							self:	Dict							)			->   Optional[Any]:





									"""simple docstring"""







									for model_name in DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											lowercase__            =      DecisionTransformerModel.from_pretrained(UpperCamelCase_							)
											self.assertIsNotNone(UpperCamelCase_							)


							def       lowerCamelCase_ (							self:	List[str]							)			->   Dict:





									"""simple docstring"""







									lowercase__     ,		lowercase__            =      self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
											lowercase__            =      model_class(UpperCamelCase_							)
											lowercase__            =      inspect.signature(model.forward							)
											# signature.parameters is an OrderedDict => so arg_names order is deterministic
											lowercase__            =      [*signature.parameters.keys()]

											lowercase__            =      [
											    '''states''',
											    '''actions''',
											    '''rewards''',
											    '''returns_to_go''',
											    '''timesteps''',
											    '''attention_mask''',
											]

											self.assertListEqual(arg_names[: len(UpperCamelCase_							)]    ,						UpperCamelCase_							)


@require_torch
class        _a							(			unittest.TestCase			):

							@slow
							def       lowerCamelCase_ (							self:	Tuple							)			->   Any:





									"""simple docstring"""







									lowercase__            =      2  # number of steps of autoregressive prediction we will perform
									lowercase__            =      10  # defined by the RL environment, may be normalized
									lowercase__            =      DecisionTransformerModel.from_pretrained('''edbeeching/decision-transformer-gym-hopper-expert'''							)
									lowercase__            =      model.to(UpperCamelCase_							)
									lowercase__            =      model.config
									torch.manual_seed(0							)
									lowercase__            =      torch.randn(1    ,						1    ,						config.state_dim							).to(device=UpperCamelCase_    ,						dtype=torch.floataa							)  # env.reset()

									lowercase__            =      torch.tensor(
									    [[0.242793, -0.28693074, 0.8742613], [0.67815274, -0.08101085, -0.12952147]]    ,						device=UpperCamelCase_							)

									lowercase__            =      torch.tensor(UpperCamelCase_    ,						device=UpperCamelCase_    ,						dtype=torch.floataa							).reshape(1    ,						1    ,						1							)
									lowercase__            =      state
									lowercase__            =      torch.zeros(1    ,						0    ,						config.act_dim    ,						device=UpperCamelCase_    ,						dtype=torch.floataa							)
									lowercase__            =      torch.zeros(1    ,						0    ,						device=UpperCamelCase_    ,						dtype=torch.floataa							)
									lowercase__            =      torch.tensor(0    ,						device=UpperCamelCase_    ,						dtype=torch.long							).reshape(1    ,						1							)

									for step in range(UpperCamelCase_							):
											lowercase__            =      torch.cat([actions, torch.zeros(1    ,						1    ,						config.act_dim    ,						device=UpperCamelCase_							)]    ,						dim=1							)
											lowercase__            =      torch.cat([rewards, torch.zeros(1    ,						1    ,						device=UpperCamelCase_							)]    ,						dim=1							)

											lowercase__            =      torch.ones(1    ,						states.shape[1]							).to(dtype=torch.long    ,						device=states.device							)

											with torch.no_grad():
													lowercase__     ,		lowercase__     ,		lowercase__            =      model(
													    states=UpperCamelCase_    ,						actions=UpperCamelCase_    ,						rewards=UpperCamelCase_    ,						returns_to_go=UpperCamelCase_    ,						timesteps=UpperCamelCase_    ,						attention_mask=UpperCamelCase_    ,						return_dict=UpperCamelCase_    ,						)

											self.assertEqual(action_pred.shape    ,						actions.shape							)
											self.assertTrue(torch.allclose(action_pred[0, -1]    ,						expected_outputs[step]    ,						atol=1E-4							)							)
											lowercase__     ,		lowercase__     ,		lowercase__     ,		lowercase__            =      (  # env.step(action)
											    torch.randn(1    ,						1    ,						config.state_dim							).to(device=UpperCamelCase_    ,						dtype=torch.floataa							),
											    1.0,
											    False,
											    {},
											)

											lowercase__            =      action_pred[0, -1]
											lowercase__            =      torch.cat([states, state]    ,						dim=1							)
											lowercase__            =      returns_to_go[0, -1] - reward
											lowercase__            =      torch.cat([returns_to_go, pred_return.reshape(1    ,						1    ,						1							)]    ,						dim=1							)
											lowercase__            =      torch.cat(
											    [timesteps, torch.ones((1, 1)    ,						device=UpperCamelCase_    ,						dtype=torch.long							) * (step + 1)]    ,						dim=1							)


