






"""simple docstring"""


import tempfile
import unittest

import numpy as np

import transformers
from transformers import GPTaConfig, GPTaTokenizer, is_flax_available, is_torch_available
from transformers.testing_utils import is_pt_flax_cross_test, require_flax, slow

from ...generation.test_flax_utils import FlaxGenerationTesterMixin
from ...test_modeling_flax_common import FlaxModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask


if is_flax_available():
   import jax
   import jax.numpy as jnp

   from transformers.modeling_flax_pytorch_utils import (
       convert_pytorch_state_dict_to_flax,
       load_flax_weights_in_pytorch_model,
   )
   from transformers.models.gpta.modeling_flax_gpta import FlaxGPTaLMHeadModel, FlaxGPTaModel

if is_torch_available():
   import torch






class __a :



  def __init__(self	:		List[Any]					,      lowercase_	:		Any					,      lowercase_	:		Tuple=14					,      lowercase_	:		List[Any]=7					,      lowercase_	:		Optional[int]=True					,      lowercase_	:		str=True					,      lowercase_	:		Optional[Any]=False					,      lowercase_	:		Union[str, Any]=True					,      lowercase_	:		Any=99					,      lowercase_	:		Optional[Any]=32					,      lowercase_	:		Union[str, Any]=5					,      lowercase_	:		Dict=4					,      lowercase_	:		List[str]=37					,      lowercase_	:		Tuple="gelu"					,      lowercase_	:		Optional[int]=0.1					,      lowercase_	:		int=0.1					,      lowercase_	:		Optional[Any]=512					,      lowercase_	:		Optional[int]=0.02					,      )		->				List[str]:
   __A						:							Any      =						parent
   __A						:							List[str]      =						batch_size
   __A						:							List[Any]      =						seq_length
   __A						:							int      =						is_training
   __A						:							str      =						use_input_mask
   __A						:							Union[str, Any]      =						use_token_type_ids
   __A						:							Optional[Any]      =						use_labels
   __A						:							str      =						vocab_size
   __A						:							List[Any]      =						hidden_size
   __A						:							Any      =						num_hidden_layers
   __A						:							Optional[Any]      =						num_attention_heads
   __A						:							List[Any]      =						intermediate_size
   __A						:							List[str]      =						hidden_act
   __A						:							Optional[int]      =						hidden_dropout_prob
   __A						:							Optional[int]      =						attention_probs_dropout_prob
   __A						:							Union[str, Any]      =						max_position_embeddings
   __A						:							Dict      =						initializer_range
   __A						:							List[str]      =						None
   __A						:							Optional[Any]      =						vocab_size - 1
   __A						:							int      =						vocab_size - 1
   __A						:							Tuple      =						vocab_size - 1



  def     A_					(self	:		List[Any] )		->				Optional[Any]:
   __A						:							Tuple      =						ids_tensor([self.batch_size, self.seq_length]					,      self.vocab_size )

   __A						:							Optional[Any]      =						None
   if self.use_input_mask:
    __A						:							Dict      =						random_attention_mask([self.batch_size, self.seq_length] )

   __A						:							Any      =						GPTaConfig(
       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					,      use_cache=lowercase_					,      bos_token_id=self.bos_token_id					,      eos_token_id=self.eos_token_id					,      pad_token_id=self.pad_token_id					,      )

   return (config, input_ids, input_mask)



  def     A_					(self	:		Optional[Any] )		->				Tuple:
   __A						:							Union[str, Any]      =						self.prepare_config_and_inputs()
   __A     ,       __A     ,       __A						:							Tuple      =						config_and_inputs
   __A						:							List[str]      =						{'input_ids': input_ids, 'attention_mask': attention_mask}
   return config, inputs_dict



  def     A_					(self	:		Any )		->				List[str]:
   __A     ,       __A     ,       __A						:							str      =						self.prepare_config_and_inputs()

   __A						:							Any      =						floats_tensor([self.batch_size, self.seq_length, self.hidden_size] )
   __A						:							Optional[Any]      =						ids_tensor([self.batch_size, self.seq_length]					,      vocab_size=2 )

   return (
       config,
       input_ids,
       attention_mask,
       encoder_hidden_states,
       encoder_attention_mask,
   )



  def     A_					(self	:		Dict					,      lowercase_	:		Optional[Any]					,      lowercase_	:		Union[str, Any]					,      lowercase_	:		Optional[Any]					,      lowercase_	:		str )		->				str:
   __A						:							Dict      =						20
   __A						:							Any      =						model_class_name(lowercase_ )

   __A						:							Dict      =						model.init_cache(input_ids.shape[0]					,      lowercase_ )
   __A						:							Optional[int]      =						jnp.ones((input_ids.shape[0], max_decoder_length)					,      dtype='i4' )

   __A						:							Optional[int]      =						jnp.broadcast_to(
       jnp.arange(input_ids.shape[-1] - 1 )[None, :]					,      (input_ids.shape[0], input_ids.shape[-1] - 1) )
   __A						:							List[Any]      =						model(
       input_ids[:, :-1]					,      attention_mask=lowercase_					,      past_key_values=lowercase_					,      position_ids=lowercase_					,      )

   __A						:							List[str]      =						jnp.array(input_ids.shape[0] * [[input_ids.shape[-1] - 1]]					,      dtype='i4' )
   __A						:							str      =						model(
       input_ids[:, -1:]					,      attention_mask=lowercase_					,      past_key_values=outputs_cache.past_key_values					,      position_ids=lowercase_					,      )

   __A						:							Optional[Any]      =						model(lowercase_ )

   __A						:							Optional[int]      =						np.max(np.abs((outputs_cache_next[0][:, -1, :5] - outputs[0][:, -1, :5]) ) )
   self.parent.assertTrue(diff < 1E-3					,      msg=f"Max diff is {diff}" )



  def     A_					(self	:		List[Any]					,      lowercase_	:		str					,      lowercase_	:		int					,      lowercase_	:		Dict					,      lowercase_	:		Dict )		->				List[str]:
   __A						:							Optional[int]      =						20
   __A						:							Dict      =						model_class_name(lowercase_ )

   __A						:							int      =						jnp.concatenate(
       [attention_mask, jnp.zeros((attention_mask.shape[0], max_decoder_length - attention_mask.shape[1]) )]					,      axis=-1					,      )

   __A						:							Union[str, Any]      =						model.init_cache(input_ids.shape[0]					,      lowercase_ )
   __A						:							Optional[int]      =						jnp.broadcast_to(
       jnp.arange(input_ids.shape[-1] - 1 )[None, :]					,      (input_ids.shape[0], input_ids.shape[-1] - 1) )

   __A						:							List[Any]      =						model(
       input_ids[:, :-1]					,      attention_mask=lowercase_					,      past_key_values=lowercase_					,      position_ids=lowercase_					,      )
   __A						:							List[Any]      =						jnp.array(input_ids.shape[0] * [[input_ids.shape[-1] - 1]]					,      dtype='i4' )
   __A						:							List[str]      =						model(
       input_ids[:, -1:]					,      past_key_values=outputs_cache.past_key_values					,      attention_mask=lowercase_					,      position_ids=lowercase_					,      )

   __A						:							int      =						model(lowercase_					,      attention_mask=lowercase_ )

   __A						:							Dict      =						np.max(np.abs((outputs_cache_next[0][:, -1, :5] - outputs[0][:, -1, :5]) ) )
   self.parent.assertTrue(diff < 1E-3					,      msg=f"Max diff is {diff}" )






@require_flax
class __a ( snake_case_			,snake_case_			,unittest.TestCase					):
  __lowercase     :     Union[str, Any]       						=       (FlaxGPTaModel, FlaxGPTaLMHeadModel) if is_flax_available() else ()
  __lowercase     :     str       						=       (FlaxGPTaLMHeadModel,) if is_flax_available() else ()



  def     A_					(self	:		List[Any] )		->				List[str]:
   __A						:							Dict      =						FlaxGPTaModelTester(self )



  def     A_					(self	:		Union[str, Any] )		->				List[Any]:
   for model_class_name in self.all_model_classes:
    __A     ,       __A     ,       __A						:							Union[str, Any]      =						self.model_tester.prepare_config_and_inputs()
    self.model_tester.check_use_cache_forward(lowercase_					,      lowercase_					,      lowercase_					,      lowercase_ )



  def     A_					(self	:		Tuple )		->				Any:
   for model_class_name in self.all_model_classes:
    __A     ,       __A     ,       __A						:							Union[str, Any]      =						self.model_tester.prepare_config_and_inputs()
    self.model_tester.check_use_cache_forward_with_attn_mask(
        lowercase_					,      lowercase_					,      lowercase_					,      lowercase_ )



  @slow
  def     A_					(self	:		str )		->				Dict:
   __A						:							Dict      =						GPTaTokenizer.from_pretrained('gpt2'					,      pad_token='</s>'					,      padding_side='left' )
   __A						:							List[str]      =						tokenizer(['Hello this is a long string', 'Hey']					,      return_tensors='np'					,      padding=lowercase_					,      truncation=lowercase_ )

   __A						:							Any      =						FlaxGPTaLMHeadModel.from_pretrained('gpt2' )
   __A						:							Union[str, Any]      =						False
   __A						:							List[Any]      =						model.config.eos_token_id

   __A						:							Optional[int]      =						jax.jit(model.generate )

   __A						:							Dict      =						jit_generate(inputs['input_ids']					,      attention_mask=inputs['attention_mask'] ).sequences

   __A						:							Tuple      =						tokenizer.batch_decode(lowercase_					,      skip_special_tokens=lowercase_ )

   __A						:							Optional[int]      =						[
       'Hello this is a long string of words. I\'m going to start with the first one.\n',
       'Hey, I\'m not sure if I\'m going to be able to do',
   ]

   self.assertListEqual(lowercase_					,      lowercase_ )



  @is_pt_flax_cross_test
  def     A_					(self	:		Optional[Any] )		->				Dict:
   __A     ,       __A						:							Dict      =						self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
    with self.subTest(model_class.__name__ ):
     # prepare inputs
     __A						:							List[Any]      =						self._prepare_for_class(lowercase_					,      lowercase_ )
     __A						:							Tuple      =						{k: torch.tensor(v.tolist() ) for k, v in prepared_inputs_dict.items()}

     # load corresponding PyTorch class
     __A						:							int      =						model_class.__name__[4:]  # Skip the "Flax" at the beginning
     __A						:							str      =						getattr(lowercase_					,      lowercase_ )

     __A     ,       __A						:							Optional[int]      =						pt_inputs['input_ids'].shape
     __A						:							Optional[int]      =						np.random.randint(0					,      seq_length - 1					,      size=(batch_size,) )
     for batch_idx, start_index in enumerate(lowercase_ ):
      __A						:							List[Any]      =						0
      __A						:							Union[str, Any]      =						1
      __A						:							Optional[Any]      =						0
      __A						:							Optional[Any]      =						1
     __A						:							Tuple      =						pt_model_class(lowercase_ ).eval()
     __A						:							List[str]      =						model_class(lowercase_					,      dtype=jnp.floataa )

     __A						:							Dict      =						convert_pytorch_state_dict_to_flax(pt_model.state_dict()					,      lowercase_ )
     __A						:							Union[str, Any]      =						fx_state

     with torch.no_grad():
      __A						:							Tuple      =						pt_model(**lowercase_ ).to_tuple()

     __A						:							Union[str, Any]      =						fx_model(**lowercase_ ).to_tuple()
     self.assertEqual(len(lowercase_ )					,      len(lowercase_ )					,      'Output lengths differ between Flax and PyTorch' )
     for fx_output, pt_output in zip(lowercase_					,      lowercase_ ):
      self.assert_almost_equals(fx_output[:, -1]					,      pt_output[:, -1].numpy()					,      4E-2 )

     with tempfile.TemporaryDirectory() as tmpdirname:
      pt_model.save_pretrained(lowercase_ )
      __A						:							Any      =						model_class.from_pretrained(lowercase_					,      from_pt=lowercase_ )

     __A						:							Dict      =						fx_model_loaded(**lowercase_ ).to_tuple()
     self.assertEqual(
         len(lowercase_ )					,      len(lowercase_ )					,      'Output lengths differ between Flax and PyTorch' )
     for fx_output_loaded, pt_output in zip(lowercase_					,      lowercase_ ):
      self.assert_almost_equals(fx_output_loaded[:, -1]					,      pt_output[:, -1].numpy()					,      4E-2 )



  @is_pt_flax_cross_test
  def     A_					(self	:		Optional[int] )		->				Tuple:
   __A     ,       __A						:							Tuple      =						self.model_tester.prepare_config_and_inputs_for_common()
   for model_class in self.all_model_classes:
    with self.subTest(model_class.__name__ ):
     # prepare inputs
     __A						:							Any      =						self._prepare_for_class(lowercase_					,      lowercase_ )
     __A						:							int      =						{k: torch.tensor(v.tolist() ) for k, v in prepared_inputs_dict.items()}

     # load corresponding PyTorch class
     __A						:							List[Any]      =						model_class.__name__[4:]  # Skip the "Flax" at the beginning
     __A						:							Union[str, Any]      =						getattr(lowercase_					,      lowercase_ )

     __A						:							Optional[Any]      =						pt_model_class(lowercase_ ).eval()
     __A						:							List[str]      =						model_class(lowercase_					,      dtype=jnp.floataa )

     __A						:							int      =						load_flax_weights_in_pytorch_model(lowercase_					,      fx_model.params )
     __A     ,       __A						:							Optional[Any]      =						pt_inputs['input_ids'].shape
     __A						:							List[str]      =						np.random.randint(0					,      seq_length - 1					,      size=(batch_size,) )
     for batch_idx, start_index in enumerate(lowercase_ ):
      __A						:							Any      =						0
      __A						:							Tuple      =						1
      __A						:							int      =						0
      __A						:							Optional[int]      =						1

     # make sure weights are tied in PyTorch
     pt_model.tie_weights()

     with torch.no_grad():
      __A						:							Optional[Any]      =						pt_model(**lowercase_ ).to_tuple()

     __A						:							Optional[int]      =						fx_model(**lowercase_ ).to_tuple()
     self.assertEqual(len(lowercase_ )					,      len(lowercase_ )					,      'Output lengths differ between Flax and PyTorch' )
     for fx_output, pt_output in zip(lowercase_					,      lowercase_ ):
      self.assert_almost_equals(fx_output[:, -1]					,      pt_output[:, -1].numpy()					,      4E-2 )

     with tempfile.TemporaryDirectory() as tmpdirname:
      fx_model.save_pretrained(lowercase_ )
      __A						:							int      =						pt_model_class.from_pretrained(lowercase_					,      from_flax=lowercase_ )

     with torch.no_grad():
      __A						:							List[str]      =						pt_model_loaded(**lowercase_ ).to_tuple()

     self.assertEqual(
         len(lowercase_ )					,      len(lowercase_ )					,      'Output lengths differ between Flax and PyTorch' )
     for fx_output, pt_output in zip(lowercase_					,      lowercase_ ):
      self.assert_almost_equals(fx_output[:, -1]					,      pt_output[:, -1].numpy()					,      4E-2 )



  @slow
  def     A_					(self	:		Tuple )		->				Union[str, Any]:
   for model_class_name in self.all_model_classes:
    __A						:							str      =						model_class_name.from_pretrained('gpt2'					,      from_pt=lowercase_ )
    __A						:							Dict      =						model(np.ones((1, 1) ) )
    self.assertIsNotNone(lowercase_ )













"""simple docstring"""


import importlib.metadata
import subprocess
import sys




def 					a			(  UpperCamelCase_				:					bool = False		)	->		Optional[Any]:
 __A						:							List[Any]      =						False
 if "IPython" in sys.modules:
  __A						:							List[Any]      =						'google.colab' in str(sys.modules['IPython'].get_ipython()		)

 if in_colab:
  if upgrade:
   __A						:							Dict      =						['pip', 'install', '-U', 'torch']
   subprocess.run(UpperCamelCase_  ,						check=UpperCamelCase_		)
  # get the current version of torch
  __A						:							int      =						importlib.metadata.version('torch'		)
  __A						:							int      =						torch_version[: torch_version.rindex('.'		)]
  __A						:							Any      =						F"https://storage.googleapis.com/tpu-pytorch/wheels/colab/torch_xla-{torch_version_trunc}-cp37-cp37m-linux_x86_64.whl"
  __A						:							int      =						['pip', 'install', xla_wheel]
  subprocess.run(UpperCamelCase_  ,						check=UpperCamelCase_		)
 else:
  raise RuntimeError('`install_xla` utility works only on google colab.'		)





