

'''simple docstring'''

from __future__ import annotations

import unittest

from transformers import CTRLConfig, 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, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
						import tensorflow as tf

						from transformers.models.ctrl.modeling_tf_ctrl import (
						    TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
						    TFCTRLForSequenceClassification,
						    TFCTRLLMHeadModel,
						    TFCTRLModel,
						)







class lowerCamelCase   (    __lowerCAmelCase				):

				def __init__(       self,						lowercase_,						)    ->							Any:
					snake_case							=					parent
					snake_case							=					13
					snake_case							=					7
					snake_case							=					True
					snake_case							=					True
					snake_case							=					True
					snake_case							=					True
					snake_case							=					True
					snake_case							=					99
					snake_case							=					32
					snake_case							=					2
					snake_case							=					4
					snake_case							=					37
					snake_case							=					'gelu'
					snake_case							=					0.1
					snake_case							=					0.1
					snake_case							=					512
					snake_case							=					16
					snake_case							=					2
					snake_case							=					0.02
					snake_case							=					3
					snake_case							=					4
					snake_case							=					None
					snake_case							=					self.vocab_size - 1

				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					ids_tensor([self.batch_size, self.seq_length],						self.vocab_size )

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

					snake_case							=					None
					if self.use_token_type_ids:
						snake_case							=					ids_tensor([self.batch_size, self.seq_length],						self.type_vocab_size )

					snake_case							=					None
					if self.use_mc_token_ids:
						snake_case							=					ids_tensor([self.batch_size, self.num_choices],						self.seq_length )

					snake_case							=					None
					snake_case							=					None
					snake_case							=					None
					if self.use_labels:
						snake_case							=					ids_tensor([self.batch_size],						self.type_sequence_label_size )
						snake_case							=					ids_tensor([self.batch_size, self.seq_length],						self.num_labels )
						snake_case							=					ids_tensor([self.batch_size],						self.num_choices )

					snake_case							=					CTRLConfig(
					    vocab_size=self.vocab_size,						n_embd=self.hidden_size,						n_layer=self.num_hidden_layers,						n_head=self.num_attention_heads,						n_positions=self.max_position_embeddings,						pad_token_id=self.pad_token_id,						)

					snake_case							=					ids_tensor([self.num_hidden_layers, self.num_attention_heads],						2 )

					return (
					    config,
					    input_ids,
					    input_mask,
					    head_mask,
					    token_type_ids,
					    mc_token_ids,
					    sequence_labels,
					    token_labels,
					    choice_labels,
					)

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						*lowercase_ )    ->							Dict:
					snake_case							=					TFCTRLModel(config=lowercase_ )
					snake_case							=					{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
					snake_case							=					model(lowercase_ )

					snake_case							=					[input_ids, None, input_mask]  # None is the input for 'past'
					snake_case							=					model(lowercase_ )

					snake_case							=					model(lowercase_ )

					self.parent.assertEqual(result.last_hidden_state.shape,						(self.batch_size, self.seq_length, self.hidden_size) )

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						*lowercase_ )    ->							Tuple:
					snake_case							=					TFCTRLLMHeadModel(config=lowercase_ )
					snake_case							=					{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
					snake_case							=					model(lowercase_ )
					self.parent.assertEqual(result.logits.shape,						(self.batch_size, self.seq_length, self.vocab_size) )

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						*lowercase_ )    ->							Optional[Any]:
					snake_case							=					self.num_labels
					snake_case							=					ids_tensor([self.batch_size],						self.type_sequence_label_size )
					snake_case							=					{
					    'input_ids': input_ids,
					    'token_type_ids': token_type_ids,
					    'labels': sequence_labels,
					}
					snake_case							=					TFCTRLForSequenceClassification(lowercase_ )
					snake_case							=					model(lowercase_ )
					self.parent.assertEqual(result.logits.shape,						(self.batch_size, self.num_labels) )






				def 					_lowerCamelCase    (       self )    ->							Dict:
					snake_case							=					self.prepare_config_and_inputs()

					(
					    (
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			(
					    snake_case
					)			,			
					)							=					config_and_inputs

					snake_case							=					{'input_ids': input_ids, 'token_type_ids': token_type_ids, 'attention_mask': input_mask}
					return config, inputs_dict







@require_tf
class lowerCamelCase   (    __lowerCAmelCase				,				__lowerCAmelCase				,				unittest.TestCase				):
				snake_case_							     = (TFCTRLModel, TFCTRLLMHeadModel, TFCTRLForSequenceClassification) if is_tf_available() else ()
				snake_case_							     = (TFCTRLLMHeadModel,) if is_tf_available() else ()
				snake_case_							     = (
				    {
				        '''feature-extraction''': TFCTRLModel,
				        '''text-classification''': TFCTRLForSequenceClassification,
				        '''text-generation''': TFCTRLLMHeadModel,
				        '''zero-shot''': TFCTRLForSequenceClassification,
				    }
				    if is_tf_available()
				    else {}
				)
				snake_case_							     = False
				snake_case_							     = False

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_,						lowercase_,						lowercase_ )    ->							Tuple:
					if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
						# Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
						# `CTRLConfig` was never used in pipeline tests, either because of a missing checkpoint or because a tiny
						# config could not be created.
						return True

					return False

				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					snake_case							=					TFCTRLModelTester(self )
					snake_case							=					ConfigTester(self,						config_class=lowercase_,						n_embd=37 )

				def 					_lowerCamelCase    (       self )    ->							Tuple:
					self.config_tester.run_common_tests()

				def 					_lowerCamelCase    (       self )    ->							Optional[Any]:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_ctrl_model(*lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							Any:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_ctrl_lm_head(*lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							List[str]:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_ctrl_for_sequence_classification(*lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()
					snake_case							=					[TFCTRLLMHeadModel]
					snake_case							=					[TFCTRLForSequenceClassification]

					for model_class in self.all_model_classes:
						snake_case							=					model_class(lowercase_ )
						model.build()  # may be needed for the get_bias() call below
						assert isinstance(model.get_input_embeddings(),						tf.keras.layers.Layer )

						if model_class in list_lm_models:
							snake_case							=					model.get_output_embeddings()
							assert isinstance(lowercase_,						tf.keras.layers.Layer )
							snake_case							=					model.get_bias()
							assert isinstance(lowercase_,						lowercase_ )
							for k, v in name.items():
								assert isinstance(lowercase_,						tf.Variable )
						elif model_class in list_other_models_with_output_ebd:
							snake_case							=					model.get_output_embeddings()
							assert isinstance(lowercase_,						tf.keras.layers.Layer )
							snake_case							=					model.get_bias()
							assert name is None
						else:
							snake_case							=					model.get_output_embeddings()
							assert x is None
							snake_case							=					model.get_bias()
							assert name is None






				@slow
				def 					_lowerCamelCase    (       self )    ->							Tuple:
					for model_name in TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
						snake_case							=					TFCTRLModel.from_pretrained(lowercase_ )
						self.assertIsNotNone(lowercase_ )







@require_tf
class lowerCamelCase   (    unittest.TestCase				):

				@slow
				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					snake_case							=					TFCTRLLMHeadModel.from_pretrained('ctrl' )
					snake_case							=					tf.convert_to_tensor([[11859, 0, 1611, 8]],						dtype=tf.intaa )  # Legal the president is
					snake_case							=					[
					    11859,
					    0,
					    1611,
					    8,
					    5,
					    150,
					    26449,
					    2,
					    19,
					    348,
					    469,
					    3,
					    2595,
					    48,
					    20740,
					    246533,
					    246533,
					    19,
					    30,
					    5,
					]  # Legal the president is a good guy and I don't want to lose my job. \n \n I have a

					snake_case							=					model.generate(lowercase_,						do_sample=lowercase_ )
					self.assertListEqual(output_ids[0].numpy().tolist(),						lowercase_ )










import inspect
import warnings

import torch

from .state import AcceleratorState, GradientState
from .utils import DistributedType, honor_type, is_tpu_available


if is_tpu_available(check_device=False):
      import torch_xla.core.xla_model as xm






def 						_lowerCAmelCase    (					__lowerCamelCase :  Any		,						__lowerCamelCase :  str  ):



 """simple docstring"""





 if isinstance(__lowerCamelCase		,						(list, tuple)  ):
  return honor_type(__lowerCamelCase		,						(move_to_device(__lowerCamelCase		,						__lowerCamelCase  ) for t in state)  )
 elif isinstance(__lowerCamelCase		,						__lowerCamelCase  ):
  return type(__lowerCamelCase  )({k: move_to_device(__lowerCamelCase		,						__lowerCamelCase  ) for k, v in state.items()}  )
 elif isinstance(__lowerCamelCase		,						torch.Tensor  ):
  return state.to(__lowerCamelCase  )
 return state






class    _SCREAMING_SNAKE_CASE						(torch.optim.Optimizer			):





     def __init__(							self			:				int  ,	UpperCamelCase			:				Any  ,	UpperCamelCase			:				List[str]=True  ,	UpperCamelCase			:				List[Any]=None					)->Optional[int]:
      __SCREAMING_SNAKE_CASE							:     str							=						optimizer
      __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						scaler
      __SCREAMING_SNAKE_CASE							:     Any							=						AcceleratorState()
      __SCREAMING_SNAKE_CASE							:     int							=						GradientState()
      __SCREAMING_SNAKE_CASE							:     str							=						device_placement
      __SCREAMING_SNAKE_CASE							:     int							=						False
      __SCREAMING_SNAKE_CASE							:     Dict							=						None

      # Handle device placement
      if device_placement:
       __SCREAMING_SNAKE_CASE							:     List[str]							=						self.optimizer.state_dict()
       if self.accelerator_state.distributed_type == DistributedType.TPU:
        xm.send_cpu_data_to_device(UpperCamelCase  ,	self.accelerator_state.device					)
       else:
        __SCREAMING_SNAKE_CASE							:     Union[str, Any]							=						move_to_device(UpperCamelCase  ,	self.accelerator_state.device					)
       self.optimizer.load_state_dict(UpperCamelCase					)





     @property
     def 						__snake_case     (							self			:				Optional[Any]					)->List[str]:
      return self.optimizer.state





     @state.setter
     def 						__snake_case     (							self			:				Optional[int]  ,	UpperCamelCase			:				List[str]					)->Dict:
      __SCREAMING_SNAKE_CASE							:     List[str]							=						state





     @property
     def 						__snake_case     (							self			:				List[Any]					)->str:
      return self.optimizer.param_groups





     @param_groups.setter
     def 						__snake_case     (							self			:				List[str]  ,	UpperCamelCase			:				Any					)->Dict:
      __SCREAMING_SNAKE_CASE							:     Optional[int]							=						param_groups





     @property
     def 						__snake_case     (							self			:				str					)->str:
      return self.optimizer.defaults





     @defaults.setter
     def 						__snake_case     (							self			:				int  ,	UpperCamelCase			:				Tuple					)->List[str]:
      __SCREAMING_SNAKE_CASE							:     List[str]							=						defaults





     def 						__snake_case     (							self			:				Any  ,	UpperCamelCase			:				Tuple					)->Optional[int]:
      self.optimizer.add_param_group(UpperCamelCase					)





     def 						__snake_case     (							self			:				Union[str, Any]  ,	UpperCamelCase			:				Union[str, Any]					)->Dict:
      if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
       xm.send_cpu_data_to_device(UpperCamelCase  ,	self.accelerator_state.device					)
      self.optimizer.load_state_dict(UpperCamelCase					)





     def 						__snake_case     (							self			:				List[str]					)->List[Any]:
      return self.optimizer.state_dict()





     def 						__snake_case     (							self			:				Optional[int]  ,	UpperCamelCase			:				Any=None					)->Any:
      if self.gradient_state.sync_gradients:
       __SCREAMING_SNAKE_CASE							:     Optional[int]							=						"set_to_none" in inspect.signature(self.optimizer.zero_grad					).parameters
       if accept_arg:
        if set_to_none is None:
         __SCREAMING_SNAKE_CASE							:     Optional[int]							=						False
        self.optimizer.zero_grad(set_to_none=UpperCamelCase					)
       else:
        if set_to_none is not None:
         raise ValueError("`set_to_none` for Optimizer.zero_grad` is not supported by this optimizer."					)
        self.optimizer.zero_grad()





     def 						__snake_case     (							self			:				Dict  ,	UpperCamelCase			:				Tuple=None					)->List[Any]:
      if self.gradient_state.sync_gradients:
       if self.accelerator_state.distributed_type == DistributedType.TPU:
        __SCREAMING_SNAKE_CASE							:     List[str]							=						{"closure": closure} if closure is not None else {}
        xm.optimizer_step(self.optimizer  ,	optimizer_args=UpperCamelCase					)
       elif self.scaler is not None:
        __SCREAMING_SNAKE_CASE							:     Optional[int]							=						False
        if self._last_scale is None:
         # `get_scale` is an async operation requiring full synchronization
         # on CPU and GPUs before finishing. As a result, we store away
         # the prior one to reduce the call overhead
         __SCREAMING_SNAKE_CASE							:     List[Any]							=						self.scaler.get_scale()
         __SCREAMING_SNAKE_CASE							:     List[Any]							=						True
        self.scaler.step(self.optimizer  ,	UpperCamelCase					)
        self.scaler.update()
        __SCREAMING_SNAKE_CASE							:     Dict							=						self.scaler.get_scale()
        if not new_scale:
         # If we reduced the loss scale, it means the optimizer step was skipped because of gradient overflow.
         __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						scale_after < self._last_scale
        __SCREAMING_SNAKE_CASE							:     List[Any]							=						scale_after

       else:
        self.optimizer.step(UpperCamelCase					)





     def 						__snake_case     (							self			:				Optional[int]  ,	UpperCamelCase			:				Optional[Any]					)->Any:
      for param_group in self.optimizer.param_groups:
       __SCREAMING_SNAKE_CASE							:     Tuple							=						[parameters_map.get(UpperCamelCase  ,	UpperCamelCase					) for p in param_group["params"]]





     @property
     def 						__snake_case     (							self			:				List[Any]					)->int:
      warnings.warn(
          "The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use "
          "`optimizer.step_was_skipped` instead."  ,	UpperCamelCase  ,	)
      return self._is_overflow





     @property
     def 						__snake_case     (							self			:				Optional[int]					)->List[Any]:
      return self._is_overflow





     def __getstate__(							self			:				List[str]					)->List[str]:
      return self.__dict__.copy()






     def __setstate__(							self			:				Union[str, Any]  ,	UpperCamelCase			:				str					)->int:
      self.__dict__.update(UpperCamelCase					)






