





from __future__ import annotations

import random
import unittest

from transformers import TransfoXLConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

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


if is_tf_available():
      import tensorflow as tf

      from transformers import (
          TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST,
          TFTransfoXLForSequenceClassification,
          TFTransfoXLLMHeadModel,
          TFTransfoXLModel,
      )
class        SCREAMING_SNAKE_CASE     :

   '''simple docstring'''





   def __init__(					self    :       Tuple    ,  a    :       List[str]    ,  )->     Any:





     """simple docstring"""







     lowercase__			    =						parent
     lowercase__			    =						13
     lowercase__			    =						7
     lowercase__			    =						30
     lowercase__			    =						self.seq_length + self.mem_len
     lowercase__			    =						15
     lowercase__			    =						True
     lowercase__			    =						True
     lowercase__			    =						99
     lowercase__			    =						[10, 50, 80]
     lowercase__			    =						32
     lowercase__			    =						32
     lowercase__			    =						4
     lowercase__			    =						8
     lowercase__			    =						128
     lowercase__			    =						2
     lowercase__			    =						2
     lowercase__			    =						None
     lowercase__			    =						1
     lowercase__			    =						0
     lowercase__			    =						3
     lowercase__			    =						self.vocab_size - 1
     lowercase__			    =						0.01


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





     """simple docstring"""







     lowercase__			    =						ids_tensor([self.batch_size, self.seq_length]    ,  self.vocab_size )
     lowercase__			    =						ids_tensor([self.batch_size, self.seq_length]    ,  self.vocab_size )

     lowercase__			    =						None
     if self.use_labels:
       lowercase__			    =						ids_tensor([self.batch_size, self.seq_length]    ,  self.vocab_size )

     lowercase__			    =						TransfoXLConfig(
         vocab_size=self.vocab_size    ,  mem_len=self.mem_len    ,  clamp_len=self.clamp_len    ,  cutoffs=self.cutoffs    ,  d_model=self.hidden_size    ,  d_embed=self.d_embed    ,  n_head=self.num_attention_heads    ,  d_head=self.d_head    ,  d_inner=self.d_inner    ,  div_val=self.div_val    ,  n_layer=self.num_hidden_layers    ,  eos_token_id=self.eos_token_id    ,  pad_token_id=self.vocab_size - 1    ,  init_range=self.init_range    ,  num_labels=self.num_labels    ,  )

     return (config, input_ids_a, input_ids_a, lm_labels)


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





     """simple docstring"""







     random.seed(self.seed )
     tf.random.set_seed(self.seed )


   def  SCREAMING_SNAKE_CASE_							(					self    :       Union[str, Any]    ,  a    :       Optional[Any]    ,  a    :       List[Any]    ,  a    :       Dict    ,  a    :       Optional[Any] )->     List[Any]:





     """simple docstring"""







     lowercase__			    =						TFTransfoXLModel(UpperCamelCase__ )

     lowercase__			    =						model(UpperCamelCase__ ).to_tuple()

     lowercase__			    =						{'''input_ids''': input_ids_a, '''mems''': mems_a}

     lowercase__			    =						model(UpperCamelCase__ ).to_tuple()

     self.parent.assertEqual(hidden_states_a.shape    ,  (self.batch_size, self.seq_length, self.hidden_size) )
     self.parent.assertEqual(hidden_states_a.shape    ,  (self.batch_size, self.seq_length, self.hidden_size) )
     self.parent.assertListEqual(
         [mem.shape for mem in mems_a]    ,  [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers    ,  )
     self.parent.assertListEqual(
         [mem.shape for mem in mems_a]    ,  [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers    ,  )


   def  SCREAMING_SNAKE_CASE_							(					self    :       Dict    ,  a    :       List[Any]    ,  a    :       Tuple    ,  a    :       Tuple    ,  a    :       Any )->     Union[str, Any]:





     """simple docstring"""







     lowercase__			    =						TFTransfoXLLMHeadModel(UpperCamelCase__ )

     lowercase__			    =						model(UpperCamelCase__ ).to_tuple()

     lowercase__			    =						{'''input_ids''': input_ids_a, '''labels''': lm_labels}
     lowercase__			    =						model(UpperCamelCase__ ).to_tuple()

     lowercase__			    =						model([input_ids_a, mems_a] ).to_tuple()

     lowercase__			    =						{'''input_ids''': input_ids_a, '''mems''': mems_a, '''labels''': lm_labels}

     lowercase__			    =						model(UpperCamelCase__ ).to_tuple()

     self.parent.assertEqual(lm_logits_a.shape    ,  (self.batch_size, self.seq_length, self.vocab_size) )
     self.parent.assertListEqual(
         [mem.shape for mem in mems_a]    ,  [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers    ,  )

     self.parent.assertEqual(lm_logits_a.shape    ,  (self.batch_size, self.seq_length, self.vocab_size) )
     self.parent.assertListEqual(
         [mem.shape for mem in mems_a]    ,  [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers    ,  )


   def  SCREAMING_SNAKE_CASE_							(					self    :       int    ,  a    :       Optional[int]    ,  a    :       List[Any]    ,  a    :       Dict    ,  a    :       Optional[int] )->     Optional[Any]:





     """simple docstring"""







     lowercase__			    =						TFTransfoXLForSequenceClassification(UpperCamelCase__ )
     lowercase__			    =						model(UpperCamelCase__ )
     self.parent.assertEqual(result.logits.shape    ,  (self.batch_size, self.num_labels) )


   def  SCREAMING_SNAKE_CASE_							(					self    :       Any )->     int:





     """simple docstring"""







     lowercase__			    =						self.prepare_config_and_inputs()
     (lowercase__)			    =						config_and_inputs
     lowercase__			    =						{'''input_ids''': input_ids_a}
     return config, inputs_dict






@require_tf
class        SCREAMING_SNAKE_CASE     (SCREAMING_SNAKE_CASE_    ,			SCREAMING_SNAKE_CASE_    ,			unittest.TestCase   ):

   '''simple docstring'''





   _UpperCamelCase					:	Dict				      = (
       (TFTransfoXLModel, TFTransfoXLLMHeadModel, TFTransfoXLForSequenceClassification) if is_tf_available() else ()
   )
   _UpperCamelCase					:	Union[str, Any]				      = () if is_tf_available() else ()
   _UpperCamelCase					:	str				      = (
       {
           'feature-extraction': TFTransfoXLModel,
           'text-classification': TFTransfoXLForSequenceClassification,
           'text-generation': TFTransfoXLLMHeadModel,
           'zero-shot': TFTransfoXLForSequenceClassification,
       }
       if is_tf_available()
       else {}
   )
   # TODO: add this test when TFTransfoXLLMHead has a linear output layer implemented
   _UpperCamelCase					:	Optional[int]				      = False
   _UpperCamelCase					:	Dict				      = False
   _UpperCamelCase					:	str				      = False
   _UpperCamelCase					:	List[Any]				      = False


   def  SCREAMING_SNAKE_CASE_							(					self    :       Optional[int]    ,  a    :       Union[str, Any]    ,  a    :       Any    ,  a    :       Tuple    ,  a    :       int    ,  a    :       List[str] )->     Dict:





     """simple docstring"""







     if pipeline_test_casse_name == "TextGenerationPipelineTests":
       # Get `ValueError: AttributeError: 'NoneType' object has no attribute 'new_ones'` or `AssertionError`.
       # `TransfoXLConfig` was never used in pipeline tests: cannot create a simple
       # tokenizer.
       return True

     return False


   def  SCREAMING_SNAKE_CASE_							(					self    :       Tuple )->     str:





     """simple docstring"""







     lowercase__			    =						TFTransfoXLModelTester(self )
     lowercase__			    =						ConfigTester(self    ,  config_class=UpperCamelCase__    ,  d_embed=37 )


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





     """simple docstring"""







     self.config_tester.run_common_tests()


   def  SCREAMING_SNAKE_CASE_							(					self    :       Tuple )->     Union[str, Any]:





     """simple docstring"""







     self.model_tester.set_seed()
     lowercase__			    =						self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_transfo_xl_model(*UpperCamelCase__ )


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





     """simple docstring"""







     self.model_tester.set_seed()
     lowercase__			    =						self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_transfo_xl_lm_head(*UpperCamelCase__ )


   def  SCREAMING_SNAKE_CASE_							(					self    :       Dict )->     Union[str, Any]:





     """simple docstring"""







     lowercase__			    =						self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_transfo_xl_for_sequence_classification(*UpperCamelCase__ )


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





     """simple docstring"""







     lowercase__			    =						self.model_tester.prepare_config_and_inputs_for_common()
     lowercase__			    =						[TFTransfoXLForSequenceClassification]

     for model_class in self.all_model_classes:
       lowercase__			    =						model_class(UpperCamelCase__ )
       assert isinstance(model.get_input_embeddings()    ,  tf.keras.layers.Layer )
       if model_class in list_other_models_with_output_ebd:
         lowercase__			    =						model.get_output_embeddings()
         assert isinstance(UpperCamelCase__    ,  tf.keras.layers.Layer )
         lowercase__			    =						model.get_bias()
         assert name is None
       else:
         lowercase__			    =						model.get_output_embeddings()
         assert x is None
         lowercase__			    =						model.get_bias()
         assert name is None


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





     """simple docstring"""







     pass


   @slow
   def  SCREAMING_SNAKE_CASE_							(					self    :       int )->     str:





     """simple docstring"""







     for model_name in TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
       lowercase__			    =						TFTransfoXLModel.from_pretrained(UpperCamelCase__ )
       self.assertIsNotNone(UpperCamelCase__ )


   @unittest.skip(reason='This model doesn\'t play well with fit() due to not returning a single loss.' )
   def  SCREAMING_SNAKE_CASE_							(					self    :       Union[str, Any] )->     Any:





     """simple docstring"""







     pass






@require_tf
class        SCREAMING_SNAKE_CASE     (unittest.TestCase   ):

   '''simple docstring'''





   @unittest.skip('Skip test until #12651 is resolved.' )
   @slow
   def  SCREAMING_SNAKE_CASE_							(					self    :       List[str] )->     List[str]:





     """simple docstring"""







     lowercase__			    =						TFTransfoXLLMHeadModel.from_pretrained('transfo-xl-wt103' )
     # fmt: off
     lowercase__			    =						tf.convert_to_tensor([[33,1_297,2,1,1_009,4,1_109,11_739,4_762,358,5,25,245,22,1_706,17,20_098,5,3_215,21,37,1_110,3,13,1_041,4,24,603,490,2,71_477,20_098,104_447,2,20_961,1,2_604,4,1,329,3,6_224,831,16_002,2,8,603,78_967,29_546,23,803,20,25,416,5,8,232,4,277,6,1_855,4_601,3,29_546,54,8,3_609,5,57_211,49,4,1,277,18,8,1_755,15_691,3,341,25,416,693,42_573,71,17,401,94,31,17_919,2,29_546,7_873,18,1,435,23,11_011,755,5,5_167,3,7_983,98,84,2,29_546,3_267,8,3_609,4,1,4_865,1_075,2,6_087,71,6,346,8,5_854,3,29_546,824,1_400,1_868,2,19,160,2,311,8,5_496,2,20_920,17,25,15_097,3,24,24,0]]    ,  dtype=tf.intaa )  # noqa: E231
     # fmt: on
     #  In 1991 , the remains of Russian Tsar Nicholas II and his family
     #  ( except for Alexei and Maria ) are discovered .
     #  The voice of Nicholas's young son , Tsarevich Alexei Nikolaevich , narrates the
     #  remainder of the story . 1883 Western Siberia ,
     #  a young Grigori Rasputin is asked by his father and a group of men to perform magic .
     #  Rasputin has a vision and denounces one of the men as a horse thief . Although his
     #  father initially slaps him for making such an accusation , Rasputin watches as the
     #  man is chased outside and beaten . Twenty years later , Rasputin sees a vision of
     #  the Virgin Mary , prompting him to become a priest . Rasputin quickly becomes famous ,
     #  with people , even a bishop , begging for his blessing . <eod> </s> <eos>

     # fmt: off
     lowercase__			    =						[33,1_297,2,1,1_009,4,1_109,11_739,4_762,358,5,25,245,22,1_706,17,20_098,5,3_215,21,37,1_110,3,13,1_041,4,24,603,490,2,71_477,20_098,104_447,2,20_961,1,2_604,4,1,329,3,6_224,831,16_002,2,8,603,78_967,29_546,23,803,20,25,416,5,8,232,4,277,6,1_855,4_601,3,29_546,54,8,3_609,5,57_211,49,4,1,277,18,8,1_755,15_691,3,341,25,416,693,42_573,71,17,401,94,31,17_919,2,29_546,7_873,18,1,435,23,11_011,755,5,5_167,3,7_983,98,84,2,29_546,3_267,8,3_609,4,1,4_865,1_075,2,6_087,71,6,346,8,5_854,3,29_546,824,1_400,1_868,2,19,160,2,311,8,5_496,2,20_920,17,25,15_097,3,24,24,0,33,1,1_857,2,1,1_009,4,1_109,11_739,4_762,358,5,25,245,28,1_110,3,13,1_041,4,24,603,490,2,71_477,20_098,104_447,2,20_961,1,2_604,4,1,329,3,0]  # noqa: E231
     # fmt: on
     #  In 1991, the remains of Russian Tsar Nicholas II and his family (
     #  except for Alexei and Maria ) are discovered. The voice of young son,
     #  Tsarevich Alexei Nikolaevich, narrates the remainder of the story.
     #  1883 Western Siberia, a young Grigori Rasputin is asked by his father
     #  and a group of men to perform magic. Rasputin has a vision and
     #  denounces one of the men as a horse thief. Although his father initially
     #  slaps him for making such an accusation, Rasputin watches as the man
     #  is chased outside and beaten. Twenty years later, Rasputin sees a vision
     #  of the Virgin Mary, prompting him to become a priest.
     #  Rasputin quickly becomes famous, with people, even a bishop, begging for
     #  his blessing. <unk> <unk> <eos> In the 1990s, the remains of Russian Tsar
     # Nicholas II and his family were discovered. The voice of <unk> young son,
     # Tsarevich Alexei Nikolaevich, narrates the remainder of the story.<eos>

     lowercase__			    =						model.generate(UpperCamelCase__    ,  max_length=200    ,  do_sample=UpperCamelCase__ )
     self.assertListEqual(output_ids[0].numpy().tolist()    ,  UpperCamelCase__ )









import numpy as np
import torch
import tqdm

from ...models.unet_ad import UNetaDModel
from ...pipelines import DiffusionPipeline
from ...utils import randn_tensor
from ...utils.dummy_pt_objects import DDPMScheduler
class        SCREAMING_SNAKE_CASE     (UpperCAmelCase   ):


   def __init__(					self    :       Optional[Any]    ,  a    :       UNetaDModel    ,  a    :       UNetaDModel    ,  a    :       DDPMScheduler    ,  a    :       Any    ,  )->     Dict:





     """simple docstring"""







     super().__init__()
     lowercase__			    =						value_function
     lowercase__			    =						unet
     lowercase__			    =						scheduler
     lowercase__			    =						env
     lowercase__			    =						env.get_dataset()
     lowercase__			    =						{}
     for key in self.data.keys():
       try:
         lowercase__			    =						self.data[key].mean()
       except:  # noqa: E722
         pass
     lowercase__			    =						{}
     for key in self.data.keys():
       try:
         lowercase__			    =						self.data[key].std()
       except:  # noqa: E722
         pass
     lowercase__			    =						env.observation_space.shape[0]
     lowercase__			    =						env.action_space.shape[0]


   def  SCREAMING_SNAKE_CASE_							(					self    :       Optional[Any]    ,  a    :       Any    ,  a    :       int )->     Dict:





     """simple docstring"""







     return (x_in - self.means[key]) / self.stds[key]


   def  SCREAMING_SNAKE_CASE_							(					self    :       Optional[int]    ,  a    :       str    ,  a    :       List[str] )->     str:





     """simple docstring"""







     return x_in * self.stds[key] + self.means[key]


   def  SCREAMING_SNAKE_CASE_							(					self    :       List[str]    ,  a    :       Tuple )->     Tuple:





     """simple docstring"""







     if type(a ) is dict:
       return {k: self.to_torch(a ) for k, v in x_in.items()}
     elif torch.is_tensor(a ):
       return x_in.to(self.unet.device )
     return torch.tensor(a    ,  device=self.unet.device )


   def  SCREAMING_SNAKE_CASE_							(					self    :       Optional[Any]    ,  a    :       Optional[int]    ,  a    :       Dict    ,  a    :       Optional[Any] )->     List[Any]:





     """simple docstring"""







     for key, val in cond.items():
       lowercase__			    =						val.clone()
     return x_in


   def  SCREAMING_SNAKE_CASE_							(					self    :       Tuple    ,  a    :       Optional[Any]    ,  a    :       Any    ,  a    :       Optional[Any]    ,  a    :       Optional[int] )->     List[Any]:





     """simple docstring"""







     lowercase__			    =						x.shape[0]
     lowercase__			    =						None
     for i in tqdm.tqdm(self.scheduler.timesteps ):
       # create batch of timesteps to pass into model
       lowercase__			    =						torch.full((batch_size,)    ,  a    ,  device=self.unet.device    ,  dtype=torch.long )
       for _ in range(a ):
         with torch.enable_grad():
           x.requires_grad_()

           # permute to match dimension for pre-trained models
           lowercase__			    =						self.value_function(x.permute(0    ,  2    ,  1 )    ,  a ).sample
           lowercase__			    =						torch.autograd.grad([y.sum()]    ,  [x] )[0]

           lowercase__			    =						self.scheduler._get_variance(a )
           lowercase__			    =						torch.exp(0.5 * posterior_variance )
           lowercase__			    =						model_std * grad

         lowercase__			    =						0
         lowercase__			    =						x.detach()
         lowercase__			    =						x + scale * grad
         lowercase__			    =						self.reset_xa(a    ,  a    ,  self.action_dim )

       lowercase__			    =						self.unet(x.permute(0    ,  2    ,  1 )    ,  a ).sample.permute(0    ,  2    ,  1 )

       # TODO: verify deprecation of this kwarg
       lowercase__			    =						self.scheduler.step(a    ,  a    ,  a    ,  predict_epsilon=a )['prev_sample']

       # apply conditions to the trajectory (set the initial state)
       lowercase__			    =						self.reset_xa(a    ,  a    ,  self.action_dim )
       lowercase__			    =						self.to_torch(a )
     return x, y


   def __call__(					self    :       Any    ,  a    :       Tuple    ,  a    :       int=64    ,  a    :       Tuple=32    ,  a    :       List[Any]=2    ,  a    :       List[str]=0.1 )->     List[Any]:





     """simple docstring"""







     lowercase__			    =						self.normalize(a    ,  'observations' )
     lowercase__			    =						obs[None].repeat(a    ,  axis=0 )

     lowercase__			    =						{0: self.to_torch(a )}
     lowercase__			    =						(batch_size, planning_horizon, self.state_dim + self.action_dim)

     # generate initial noise and apply our conditions (to make the trajectories start at current state)
     lowercase__			    =						randn_tensor(a    ,  device=self.unet.device )
     lowercase__			    =						self.reset_xa(a    ,  a    ,  self.action_dim )
     lowercase__			    =						self.to_torch(a )

     # run the diffusion process
     lowercase__    ,					lowercase__			    =						self.run_diffusion(a    ,  a    ,  a    ,  a )

     # sort output trajectories by value
     lowercase__			    =						y.argsort(0    ,  descending=a ).squeeze()
     lowercase__			    =						x[sorted_idx]
     lowercase__			    =						sorted_values[:, :, : self.action_dim]
     lowercase__			    =						actions.detach().cpu().numpy()
     lowercase__			    =						self.de_normalize(a    ,  key='actions' )

     # select the action with the highest value
     if y is not None:
       lowercase__			    =						0
     else:
       # if we didn't run value guiding, select a random action
       lowercase__			    =						np.random.randint(0    ,  a )

     lowercase__			    =						denorm_actions[selected_index, 0]
     return denorm_actions


