





'''simple docstring'''





SCREAMING_SNAKE_CASE_		     = {
    """Pillow""": """Pillow<10.0.0""",
    """accelerate""": """accelerate>=0.20.3""",
    """av""": """av==9.2.0""",
    """beautifulsoup4""": """beautifulsoup4""",
    """black""": """black~=23.1""",
    """codecarbon""": """codecarbon==1.2.0""",
    """cookiecutter""": """cookiecutter==1.7.3""",
    """dataclasses""": """dataclasses""",
    """datasets""": """datasets!=2.5.0""",
    """decord""": """decord==0.6.0""",
    """deepspeed""": """deepspeed>=0.9.3""",
    """diffusers""": """diffusers""",
    """dill""": """dill<0.3.5""",
    """evaluate""": """evaluate>=0.2.0""",
    """fairscale""": """fairscale>0.3""",
    """faiss-cpu""": """faiss-cpu""",
    """fastapi""": """fastapi""",
    """filelock""": """filelock""",
    """flax""": """flax>=0.4.1,<=0.7.0""",
    """ftfy""": """ftfy""",
    """fugashi""": """fugashi>=1.0""",
    """GitPython""": """GitPython<3.1.19""",
    """hf-doc-builder""": """hf-doc-builder>=0.3.0""",
    """huggingface-hub""": """huggingface-hub>=0.14.1,<1.0""",
    """importlib_metadata""": """importlib_metadata""",
    """ipadic""": """ipadic>=1.0.0,<2.0""",
    """isort""": """isort>=5.5.4""",
    """jax""": """jax>=0.2.8,!=0.3.2,<=0.4.13""",
    """jaxlib""": """jaxlib>=0.1.65,<=0.4.13""",
    """jieba""": """jieba""",
    """kenlm""": """kenlm""",
    """keras-nlp""": """keras-nlp>=0.3.1""",
    """librosa""": """librosa""",
    """nltk""": """nltk""",
    """natten""": """natten>=0.14.6""",
    """numpy""": """numpy>=1.17""",
    """onnxconverter-common""": """onnxconverter-common""",
    """onnxruntime-tools""": """onnxruntime-tools>=1.4.2""",
    """onnxruntime""": """onnxruntime>=1.4.0""",
    """opencv-python""": """opencv-python""",
    """optuna""": """optuna""",
    """optax""": """optax>=0.0.8,<=0.1.4""",
    """packaging""": """packaging>=20.0""",
    """parameterized""": """parameterized""",
    """phonemizer""": """phonemizer""",
    """protobuf""": """protobuf""",
    """psutil""": """psutil""",
    """pyyaml""": """pyyaml>=5.1""",
    """pydantic""": """pydantic<2""",
    """pytest""": """pytest>=7.2.0""",
    """pytest-timeout""": """pytest-timeout""",
    """pytest-xdist""": """pytest-xdist""",
    """python""": """python>=3.8.0""",
    """ray[tune]""": """ray[tune]""",
    """regex""": """regex!=2019.12.17""",
    """requests""": """requests""",
    """rhoknp""": """rhoknp>=1.1.0,<1.3.1""",
    """rjieba""": """rjieba""",
    """rouge-score""": """rouge-score!=0.0.7,!=0.0.8,!=0.1,!=0.1.1""",
    """ruff""": """ruff>=0.0.241,<=0.0.259""",
    """sacrebleu""": """sacrebleu>=1.4.12,<2.0.0""",
    """sacremoses""": """sacremoses""",
    """safetensors""": """safetensors>=0.3.1""",
    """sagemaker""": """sagemaker>=2.31.0""",
    """scikit-learn""": """scikit-learn""",
    """sentencepiece""": """sentencepiece>=0.1.91,!=0.1.92""",
    """sigopt""": """sigopt""",
    """starlette""": """starlette""",
    """sudachipy""": """sudachipy>=0.6.6""",
    """sudachidict_core""": """sudachidict_core>=20220729""",
    """tensorflow-cpu""": """tensorflow-cpu>=2.6,<2.14""",
    """tensorflow""": """tensorflow>=2.6,<2.14""",
    """tensorflow-text""": """tensorflow-text<2.14""",
    """tf2onnx""": """tf2onnx""",
    """timeout-decorator""": """timeout-decorator""",
    """timm""": """timm""",
    """tokenizers""": """tokenizers>=0.11.1,!=0.11.3,<0.14""",
    """torch""": """torch>=1.9,!=1.12.0""",
    """torchaudio""": """torchaudio""",
    """torchvision""": """torchvision""",
    """pyctcdecode""": """pyctcdecode>=0.4.0""",
    """tqdm""": """tqdm>=4.27""",
    """unidic""": """unidic>=1.0.2""",
    """unidic_lite""": """unidic_lite>=1.0.7""",
    """urllib3""": """urllib3<2.0.0""",
    """uvicorn""": """uvicorn""",
}









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 	UpperCamelCase__				:



     '''simple docstring'''

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




           '''simple docstring'''


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






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




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		floats_tensor((self.batch_size, self.seq_length, self.state_dim) )
           SCREAMING_SNAKE_CASE	=		floats_tensor((self.batch_size, self.seq_length, self.act_dim) )
           SCREAMING_SNAKE_CASE	=		floats_tensor((self.batch_size, self.seq_length, 1) )
           SCREAMING_SNAKE_CASE	=		floats_tensor((self.batch_size, self.seq_length, 1) )
           SCREAMING_SNAKE_CASE	=		ids_tensor((self.batch_size, self.seq_length)		,vocab_size=1000 )
           SCREAMING_SNAKE_CASE	=		random_attention_mask((self.batch_size, self.seq_length) )

           SCREAMING_SNAKE_CASE	=		self.get_config()

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






     def        SCREAMING_SNAKE_CASE__						(							self		:					Tuple )  ->	List[str]:




           '''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        SCREAMING_SNAKE_CASE__						(							self		:					Optional[Any]		,lowerCamelCase__		:					int		,lowerCamelCase__		:					Optional[int]		,lowerCamelCase__		:					Union[str, Any]		,lowerCamelCase__		:					Any		,lowerCamelCase__		:					List[Any]		,lowerCamelCase__		:					Any		,lowerCamelCase__		:					Union[str, Any]		,)  ->	Dict:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		DecisionTransformerModel(config=lowerCamelCase__ )
           model.to(lowerCamelCase__ )
           model.eval()
           SCREAMING_SNAKE_CASE	=		model(lowerCamelCase__		,lowerCamelCase__		,lowerCamelCase__		,lowerCamelCase__		,lowerCamelCase__		,lowerCamelCase__ )

           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        SCREAMING_SNAKE_CASE__						(							self		:					Optional[Any] )  ->	Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		self.prepare_config_and_inputs()
           (
               (
               SCREAMING_SNAKE_CASE
           ),      (
               SCREAMING_SNAKE_CASE
           ),      (
               SCREAMING_SNAKE_CASE
           ),      (
               SCREAMING_SNAKE_CASE
           ),      (
               SCREAMING_SNAKE_CASE
           ),      (
               SCREAMING_SNAKE_CASE
           ),      (
               SCREAMING_SNAKE_CASE
           ),      
           )	=		config_and_inputs
           SCREAMING_SNAKE_CASE	=		{
               """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 	UpperCamelCase__				(   lowerCAmelCase_							,   lowerCAmelCase_							,   lowerCAmelCase_							,   unittest.TestCase     ):



     '''simple docstring'''

     __snake_case					:			Tuple				=    (DecisionTransformerModel,) if is_torch_available() else ()
     __snake_case					:			Dict				=    ()
     __snake_case					:			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
     __snake_case					:			int				=    False

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






     def        SCREAMING_SNAKE_CASE__						(							self		:					Optional[int] )  ->	Dict:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		DecisionTransformerModelTester(self )
           SCREAMING_SNAKE_CASE	=		ConfigTester(self		,config_class=lowerCamelCase__		,hidden_size=37 )






     def        SCREAMING_SNAKE_CASE__						(							self		:					Dict )  ->	Dict:




           '''simple docstring'''


           self.config_tester.run_common_tests()






     def        SCREAMING_SNAKE_CASE__						(							self		:					int )  ->	List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*lowerCamelCase__ )






     @slow
     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	Union[str, Any]:




           '''simple docstring'''


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






     def        SCREAMING_SNAKE_CASE__						(							self		:					Union[str, Any] )  ->	List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE,      SCREAMING_SNAKE_CASE	=		self.model_tester.prepare_config_and_inputs_for_common()

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

                 SCREAMING_SNAKE_CASE	=		[
                     """states""",
                     """actions""",
                     """rewards""",
                     """returns_to_go""",
                     """timesteps""",
                     """attention_mask""",
                 ]

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

@require_torch
class 	UpperCamelCase__				(   unittest.TestCase     ):



     '''simple docstring'''

     @slow
     def        SCREAMING_SNAKE_CASE__						(							self		:					Dict )  ->	Optional[int]:




           '''simple docstring'''


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

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

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

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

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

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

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

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


