





import copy
import logging
import random

import joblib
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from tqdm import tqdm

from transformers import AdamW, GPTaLMHeadModel, get_linear_schedule_with_warmup


__UpperCAmelCase			:  Optional[int]										=				logging.getLogger(__name__)







def      A__				(       SCREAMING_SNAKE_CASE__)  ->       List[str]:
  random.seed(SCREAMING_SNAKE_CASE__)
  np.random.seed(SCREAMING_SNAKE_CASE__)
  torch.manual_seed(SCREAMING_SNAKE_CASE__)
  torch.cuda.manual_seed_all(SCREAMING_SNAKE_CASE__)







def      A__				(       SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)  ->       Any:
  model.eval()
  __snake_case:		Union[str, Any]					    =		next(model.parameters()).device
  __snake_case:		Any					    =		1
  __snake_case:		str					    =		torch.zeros((eval_batch_size, context_len)   ,   dtype=torch.long   ,   device=SCREAMING_SNAKE_CASE__)
  __snake_case:		List[Any]					    =		DataLoader(SCREAMING_SNAKE_CASE__   ,   shuffle=SCREAMING_SNAKE_CASE__   ,   batch_size=SCREAMING_SNAKE_CASE__)
  __snake_case:		Any					    =		torch.zeros(1   ,   device=SCREAMING_SNAKE_CASE__)
  __snake_case:		str					    =		0
  for batch in eval_dataloader:
    batch.to(SCREAMING_SNAKE_CASE__)
    # pad
    context.zero_()
    for i in range(SCREAMING_SNAKE_CASE__):
      __snake_case:		str					    =		batch[i]
    __snake_case:		str					    =		model(SCREAMING_SNAKE_CASE__   ,   labels=SCREAMING_SNAKE_CASE__)
    eval_loss += outputs[0].sum().item()
    nb_eval_examples += batch.size(0)
  __snake_case:		Optional[Any]					    =		eval_loss / nb_eval_examples
  __snake_case:		Optional[int]					    =		torch.exp(SCREAMING_SNAKE_CASE__)
  model.train()
  return perplexity







def      A__				(       SCREAMING_SNAKE_CASE__="gpt2")  ->       Optional[int]:
  __snake_case:		int					    =		GPTaLMHeadModel.from_pretrained(SCREAMING_SNAKE_CASE__   ,   output_hidden_states=SCREAMING_SNAKE_CASE__)
  torch.save(model.state_dict()   ,   model_name + """local.pt""")
  return model







def      A__				(       SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)  ->       Union[str, Any]:
  __snake_case:		Dict					    =		copy.deepcopy(SCREAMING_SNAKE_CASE__)
  model.to(SCREAMING_SNAKE_CASE__)

  __snake_case:		List[Any]					    =		["""bias""", """LayerNorm.weight"""]
  __snake_case:		Optional[int]					    =		[
      {
          """params""": [p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay)],
          """weight_decay""": 0.0,
      },
      {"""params""": [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay)], """weight_decay""": 0.0},
  ]
  __snake_case:		Any					    =		AdamW(SCREAMING_SNAKE_CASE__   ,   lr=5e-5   ,   eps=1e-8)
  __snake_case:		Any					    =		get_linear_schedule_with_warmup(SCREAMING_SNAKE_CASE__   ,   0   ,   SCREAMING_SNAKE_CASE__)
  torch.cuda.empty_cache()
  return model, lm_optimizer, lm_scheduler







def      A__				(       SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)  ->       List[Any]:
  __snake_case:		List[str]					    =		np.array(SCREAMING_SNAKE_CASE__).mean()
  __snake_case:		Optional[Any]					    =		np.array(SCREAMING_SNAKE_CASE__).std()
  __snake_case:		Optional[Any]					    =		(real_perps - avg) / std
  __snake_case:		Optional[int]					    =		list(zip(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__))
  joblib.dump(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)







def      A__				(       SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__="dev.jbl"   ,   SCREAMING_SNAKE_CASE__=recopy_gpta   ,   )  ->       Any:
  __snake_case:		List[Any]					    =		[]
  __snake_case:		Tuple					    =		[]
  __snake_case:		Optional[Any]					    =		[]

  # Initialize the transformer model
  __snake_case:		Tuple					    =		copy.deepcopy(SCREAMING_SNAKE_CASE__)
  orig_model.to(device="""cpu""")
  torch.cuda.empty_cache()

  # Compute perplexity of initial transformer model for comparison
  model.train()
  __snake_case    ,  __snake_case    ,  __snake_case:		Union[str, Any]					    =		recopy_model(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)

  for step in tqdm(range(SCREAMING_SNAKE_CASE__)):
    __snake_case:		List[Any]					    =		torch.zeros((1, context_len)   ,   dtype=torch.long   ,   device=SCREAMING_SNAKE_CASE__)
    __snake_case:		Dict					    =		random.choice(SCREAMING_SNAKE_CASE__)
    __snake_case:		Any					    =		random.randint(0   ,   len(story[0]) - context_len - 1)
    __snake_case:		Optional[int]					    =		story[0][start : start + context_len]
    lm_optimizer.zero_grad()
    __snake_case:		str					    =		model(SCREAMING_SNAKE_CASE__   ,   labels=SCREAMING_SNAKE_CASE__)
    __snake_case:		Union[str, Any]					    =		outputs[0]
    __snake_case:		List[Any]					    =		compute_perplexity(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)
    model.train()
    lm_loss.backward()
    # Do LM backprop
    torch.nn.utils.clip_grad_norm_(model.parameters()   ,   3.0)
    lm_optimizer.step()
    lm_scheduler.step()  # Update learning rate schedule

    # Compute perplexity after back-propagating on the selected context
    __snake_case:		Any					    =		compute_perplexity(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)

    # Periodically save the stored (X, IG(X)) pairs
    if step % 1000 == 0 and step > 1:
      intermittent_save(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)

    # Reset the pretrained model to the original pretrained GPT-2 weights after each iteration
    __snake_case    ,  __snake_case    ,  __snake_case:		str					    =		recopy_model(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)

    past_perps.append(past_perp.item())
    real_perps.append(orig_perp - real_perp.item())
    contexts.append(np.array(context.cpu()))

  intermittent_save(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__)







def      A__				(       SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__="data/tokenized_stories_train_wikitext103.jbl"   ,   SCREAMING_SNAKE_CASE__=100   ,   SCREAMING_SNAKE_CASE__=1026   ,   SCREAMING_SNAKE_CASE__=True)  ->       int:
  __snake_case:		List[str]					    =		joblib.load(SCREAMING_SNAKE_CASE__)
  print("""data loaded""")
  __snake_case:		Tuple					    =		[]
  if trim:
    for i, example in enumerate(SCREAMING_SNAKE_CASE__):
      if len(example[0]) > min_len:
        __snake_case:		Optional[Any]					    =		random.randint(0   ,   len(example[0]) - context_len - 1)
        objective_set.append(example[0, start : start + context_len])
      if len(SCREAMING_SNAKE_CASE__) >= number:
        break
    __snake_case:		List[str]					    =		[]
    for j in range(i + 1   ,   len(SCREAMING_SNAKE_CASE__)):
      if len(data[j][0]) > min_len:
        train_data.append(data[j])
  else:
    __snake_case:		Optional[int]					    =		data[0:number]
    __snake_case:		Tuple					    =		data[number:]

  joblib.dump(SCREAMING_SNAKE_CASE__   ,   """objective_set.jbl""")
  print("""objective set saved""")
  return train_data, objective_set







def      A__				(       SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__=50   ,   SCREAMING_SNAKE_CASE__="secondary_learner.pt")  ->       List[str]:
  __snake_case:		Union[str, Any]					    =		torch.device("""cuda:0""" if torch.cuda.is_available() else """cpu""")
  # We will use the first 512 pairs from our dataset as a test set for
  # our secondary learner and the rest to train
  __snake_case:		int					    =		train_dataset[:512]
  __snake_case:		List[Any]					    =		train_dataset[512:]
  __snake_case:		str					    =		DataLoader(SCREAMING_SNAKE_CASE__   ,   shuffle=SCREAMING_SNAKE_CASE__   ,   batch_size=SCREAMING_SNAKE_CASE__)
  __snake_case:		Optional[int]					    =		DataLoader(SCREAMING_SNAKE_CASE__   ,   shuffle=SCREAMING_SNAKE_CASE__   ,   batch_size=SCREAMING_SNAKE_CASE__)

  # secondary learner model set up
  __snake_case:		int					    =		nn.MSELoss()
  __snake_case:		str					    =		nn.MSELoss(reduction="""sum""")
  secondary_learner.to(SCREAMING_SNAKE_CASE__)
  __snake_case:		Optional[Any]					    =		torch.optim.Adam(secondary_learner.parameters()   ,   lr=0.00_001)
  secondary_learner.train()

  # TODO in original code this is written as number of actual batches seen
  # not number of items seen but other places it is number of items instead.
  # improve consistency! changed this to epochs for clarity
  __snake_case:		Optional[int]					    =		float("""inf""")
  # Iterate through batches until we've used max_steps batches
  for epoch in range(int(SCREAMING_SNAKE_CASE__)):
    __snake_case:		Any					    =		0.0
    secondary_learner.train()
    for step, batch in enumerate(SCREAMING_SNAKE_CASE__):
      __snake_case:		int					    =		batch[0].to(SCREAMING_SNAKE_CASE__)
      __snake_case:		Union[str, Any]					    =		batch[1].to(SCREAMING_SNAKE_CASE__)
      __snake_case:		Any					    =		secondary_learner(SCREAMING_SNAKE_CASE__)
      q_optimizer.zero_grad()
      __snake_case:		List[Any]					    =		loss(SCREAMING_SNAKE_CASE__   ,   real_q.float())
      q_loss.backward()
      q_optimizer.step()
      tr_q_loss += q_loss.item()

      # model trains fairly quickly so we won't wait for a full epoch
      # eval is triggered at eval_freq and end of epochs
      if (step % eval_freq == 0 and step > 0) or ((step + 1) == len(SCREAMING_SNAKE_CASE__)):
        __snake_case:		List[Any]					    =		tr_q_loss / (step + 1)

        secondary_learner.eval()
        __snake_case:		Tuple					    =		0.0
        __snake_case:		Optional[Any]					    =		0.0
        __snake_case:		int					    =		[]
        __snake_case:		Dict					    =		[]
        # Compute performance of the secondary learner after this batch
        for stepa, batcha in enumerate(SCREAMING_SNAKE_CASE__):
          __snake_case:		Tuple					    =		batcha[0].to(SCREAMING_SNAKE_CASE__)
          __snake_case:		Union[str, Any]					    =		batcha[1].to(SCREAMING_SNAKE_CASE__)
          __snake_case:		List[Any]					    =		secondary_learner(SCREAMING_SNAKE_CASE__)
          q_lossa += test_loss(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__).item()
          sum_qa += torch.sum(SCREAMING_SNAKE_CASE__).item()
          for ei, i in enumerate(predicted_qa.cpu().detach().numpy()):
            predicted.append(i.item())
          for ei, i in enumerate(real_qa.cpu().detach().numpy()):
            actual.append(i.item())

        q_lossa /= len(SCREAMING_SNAKE_CASE__)
        print(
            """Epoch: """   ,   SCREAMING_SNAKE_CASE__   ,   """step: """   ,   SCREAMING_SNAKE_CASE__   ,   """Avg. q:"""   ,   sum_qa / len(SCREAMING_SNAKE_CASE__)   ,   """Train Loss: """   ,   SCREAMING_SNAKE_CASE__   ,   """Test Loss: """   ,   SCREAMING_SNAKE_CASE__   ,   )
        if q_lossa < best_test_loss:
          joblib.dump((predicted, actual)   ,   """pred_vs_actual.jbl""")
          torch.save(secondary_learner.state_dict()   ,   SCREAMING_SNAKE_CASE__)
          __snake_case:		Union[str, Any]					    =		q_lossa

      secondary_learner.train()
  return secondary_learner







class        __snake_case			(				nn.Module     ):





       '''simple docstring'''







       def __init__(    self					:       Union[str, Any]				,							A					:       Union[str, Any]						):
         super(A				,							self						).__init__()
         __snake_case:		str					    =		model.transformer.wte
         __snake_case:		Optional[int]					    =		copy.deepcopy(model.transformer.wte.weight						)
         __snake_case:		Optional[Any]					    =		nn.Convad(self.embeddings.weight.size(1						)				,							256				,							3				,							padding=1						)
         __snake_case:		Optional[int]					    =		nn.Sequential(nn.Linear(256				,							32						)				,							nn.Dropout(p=0.1						)				,							nn.Linear(32				,							32						)				,							nn.Linear(32				,							1						)						)




       def 				UpperCAmelCase__					(    self					:       Optional[int]				,							A					:       List[str]						):
         __snake_case:		Optional[int]					    =		torch.max(self.conv(self.embeddings(A						).squeeze(1						).transpose(1				,							2						)						)				,							2						)[0]
         __snake_case:		Optional[Any]					    =		self.fc(A						)
         return qs.squeeze(1						)






       @classmethod
       def 				UpperCAmelCase__					(    cls					:       str				,							A					:       Optional[Any]				,							A					:       List[Any]						):
         __snake_case:		Union[str, Any]					    =		cls(A						)  # this calls __init__
         __snake_case:		int					    =		torch.load(A						)
         secondary_learner.load_state_dict(A						)
         __snake_case:		str					    =		model.transformer.wte
         __snake_case:		Union[str, Any]					    =		copy.deepcopy(model.transformer.wte.weight						)
         return secondary_learner













import unittest

from transformers import FunnelConfig, FunnelTokenizer, is_torch_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
      import torch

      from transformers import (
          MODEL_FOR_PRETRAINING_MAPPING,
          FunnelBaseModel,
          FunnelForMaskedLM,
          FunnelForMultipleChoice,
          FunnelForPreTraining,
          FunnelForQuestionAnswering,
          FunnelForSequenceClassification,
          FunnelForTokenClassification,
          FunnelModel,
      )







class        __snake_case			:





       '''simple docstring'''







       def __init__(    self					:       str				,							A					:       int				,							A					:       List[Any]=13				,							A					:       Dict=7				,							A					:       Optional[int]=True				,							A					:       Dict=True				,							A					:       str=True				,							A					:       Union[str, Any]=True				,							A					:       Union[str, Any]=99				,							A					:       Dict=[1, 1, 2]				,							A					:       List[Any]=1				,							A					:       List[str]=32				,							A					:       Dict=4				,							A					:       Tuple=8				,							A					:       List[Any]=37				,							A					:       Dict="gelu_new"				,							A					:       Dict=0.1				,							A					:       List[Any]=0.1				,							A					:       Optional[int]=0.0				,							A					:       Tuple=512				,							A					:       Tuple=3				,							A					:       List[str]=0.02				,							A					:       str=3				,							A					:       Optional[Any]=4				,							A					:       str=None				,							A					:       Union[str, Any]=False				,							):
         __snake_case:		int					    =		parent
         __snake_case:		Optional[Any]					    =		batch_size
         __snake_case:		Optional[Any]					    =		seq_length
         __snake_case:		List[Any]					    =		is_training
         __snake_case:		List[str]					    =		use_input_mask
         __snake_case:		Any					    =		use_token_type_ids
         __snake_case:		Union[str, Any]					    =		use_labels
         __snake_case:		Any					    =		vocab_size
         __snake_case:		int					    =		block_sizes
         __snake_case:		Optional[int]					    =		num_decoder_layers
         __snake_case:		Union[str, Any]					    =		d_model
         __snake_case:		List[str]					    =		n_head
         __snake_case:		Optional[Any]					    =		d_head
         __snake_case:		str					    =		d_inner
         __snake_case:		Optional[Any]					    =		hidden_act
         __snake_case:		List[Any]					    =		hidden_dropout
         __snake_case:		List[Any]					    =		attention_dropout
         __snake_case:		Tuple					    =		activation_dropout
         __snake_case:		Optional[int]					    =		max_position_embeddings
         __snake_case:		Optional[int]					    =		type_vocab_size
         __snake_case:		Any					    =		2
         __snake_case:		Dict					    =		num_labels
         __snake_case:		List[Any]					    =		num_choices
         __snake_case:		Any					    =		scope
         __snake_case:		Optional[Any]					    =		initializer_std

         # Used in the tests to check the size of the first attention layer
         __snake_case:		Union[str, Any]					    =		n_head
         # Used in the tests to check the size of the first hidden state
         __snake_case:		Union[str, Any]					    =		self.d_model
         # Used in the tests to check the number of output hidden states/attentions
         __snake_case:		Optional[int]					    =		sum(self.block_sizes						) + (0 if base else self.num_decoder_layers)
         # FunnelModel adds two hidden layers: input embeddings and the sum of the upsampled encoder hidden state with
         # the last hidden state of the first block (which is the first hidden state of the decoder).
         if not base:
           __snake_case:		Any					    =		self.num_hidden_layers + 2




       def 				UpperCAmelCase__					(    self					:       Any						):
         __snake_case:		Tuple					    =		ids_tensor([self.batch_size, self.seq_length]				,							self.vocab_size						)

         __snake_case:		Optional[int]					    =		None
         if self.use_input_mask:
           __snake_case:		Optional[Any]					    =		ids_tensor([self.batch_size, self.seq_length]				,							vocab_size=2						)

         __snake_case:		Dict					    =		None
         if self.use_token_type_ids:
           __snake_case:		List[str]					    =		ids_tensor([self.batch_size, self.seq_length]				,							self.type_vocab_size						)

         __snake_case:		Optional[Any]					    =		None
         __snake_case:		Optional[int]					    =		None
         __snake_case:		Any					    =		None
         if self.use_labels:
           __snake_case:		List[Any]					    =		ids_tensor([self.batch_size]				,							self.type_sequence_label_size						)
           __snake_case:		Any					    =		ids_tensor([self.batch_size, self.seq_length]				,							self.num_labels						)
           __snake_case:		Union[str, Any]					    =		ids_tensor([self.batch_size]				,							self.num_choices						)
           __snake_case:		Tuple					    =		ids_tensor([self.batch_size, self.seq_length]				,							1						)

         __snake_case:		Dict					    =		self.get_config()

         return (
             config,
             input_ids,
             token_type_ids,
             input_mask,
             sequence_labels,
             token_labels,
             choice_labels,
             fake_token_labels,
         )




       def 				UpperCAmelCase__					(    self					:       Dict						):
         return FunnelConfig(
             vocab_size=self.vocab_size				,							block_sizes=self.block_sizes				,							num_decoder_layers=self.num_decoder_layers				,							d_model=self.d_model				,							n_head=self.n_head				,							d_head=self.d_head				,							d_inner=self.d_inner				,							hidden_act=self.hidden_act				,							hidden_dropout=self.hidden_dropout				,							attention_dropout=self.attention_dropout				,							activation_dropout=self.activation_dropout				,							max_position_embeddings=self.max_position_embeddings				,							type_vocab_size=self.type_vocab_size				,							initializer_std=self.initializer_std				,							)




       def 				UpperCAmelCase__					(    self					:       List[Any]				,							A					:       Tuple				,							A					:       Any				,							A					:       int				,							A					:       Optional[Any]				,							A					:       List[str]				,							A					:       Optional[int]				,							A					:       Tuple				,							A					:       Optional[Any]				,							):
         __snake_case:		int					    =		FunnelModel(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		Union[str, Any]					    =		model(A				,							attention_mask=A				,							token_type_ids=A						)
         __snake_case:		Dict					    =		model(A				,							token_type_ids=A						)
         __snake_case:		Optional[int]					    =		model(A						)
         self.parent.assertEqual(result.last_hidden_state.shape				,							(self.batch_size, self.seq_length, self.d_model)						)

         __snake_case:		Any					    =		False
         __snake_case:		List[Any]					    =		model(A						)
         self.parent.assertEqual(result.last_hidden_state.shape				,							(self.batch_size, self.seq_length, self.d_model)						)

         __snake_case:		List[str]					    =		False
         __snake_case:		Optional[int]					    =		model(A						)
         self.parent.assertEqual(result.last_hidden_state.shape				,							(self.batch_size, self.seq_length, self.d_model)						)




       def 				UpperCAmelCase__					(    self					:       Optional[Any]				,							A					:       Union[str, Any]				,							A					:       str				,							A					:       Any				,							A					:       str				,							A					:       Any				,							A					:       str				,							A					:       List[str]				,							A					:       Dict				,							):
         __snake_case:		Dict					    =		FunnelBaseModel(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		List[str]					    =		model(A				,							attention_mask=A				,							token_type_ids=A						)
         __snake_case:		Union[str, Any]					    =		model(A				,							token_type_ids=A						)
         __snake_case:		List[Any]					    =		model(A						)
         self.parent.assertEqual(result.last_hidden_state.shape				,							(self.batch_size, 2, self.d_model)						)

         __snake_case:		Optional[Any]					    =		False
         __snake_case:		Tuple					    =		model(A						)
         self.parent.assertEqual(result.last_hidden_state.shape				,							(self.batch_size, 3, self.d_model)						)

         __snake_case:		Union[str, Any]					    =		False
         __snake_case:		List[Any]					    =		model(A						)
         self.parent.assertEqual(result.last_hidden_state.shape				,							(self.batch_size, 2, self.d_model)						)




       def 				UpperCAmelCase__					(    self					:       Tuple				,							A					:       Union[str, Any]				,							A					:       Optional[int]				,							A					:       Optional[int]				,							A					:       Any				,							A					:       str				,							A					:       Optional[int]				,							A					:       Dict				,							A					:       str				,							):
         __snake_case:		Optional[int]					    =		self.num_labels
         __snake_case:		Optional[Any]					    =		FunnelForPreTraining(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		Dict					    =		model(A				,							attention_mask=A				,							token_type_ids=A				,							labels=A						)
         self.parent.assertEqual(result.logits.shape				,							(self.batch_size, self.seq_length)						)




       def 				UpperCAmelCase__					(    self					:       List[str]				,							A					:       Optional[int]				,							A					:       List[str]				,							A					:       Tuple				,							A					:       Any				,							A					:       str				,							A					:       Optional[Any]				,							A					:       Optional[Any]				,							A					:       Union[str, Any]				,							):
         __snake_case:		List[str]					    =		FunnelForMaskedLM(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		Any					    =		model(A				,							attention_mask=A				,							token_type_ids=A				,							labels=A						)
         self.parent.assertEqual(result.logits.shape				,							(self.batch_size, self.seq_length, self.vocab_size)						)




       def 				UpperCAmelCase__					(    self					:       Dict				,							A					:       str				,							A					:       Dict				,							A					:       List[str]				,							A					:       Any				,							A					:       Optional[Any]				,							A					:       Optional[Any]				,							A					:       Optional[Any]				,							A					:       Any				,							):
         __snake_case:		Dict					    =		self.num_labels
         __snake_case:		Dict					    =		FunnelForSequenceClassification(A						)
         model.to(A						)
         model.eval()
         __snake_case:		str					    =		model(A				,							attention_mask=A				,							token_type_ids=A				,							labels=A						)
         self.parent.assertEqual(result.logits.shape				,							(self.batch_size, self.num_labels)						)




       def 				UpperCAmelCase__					(    self					:       List[str]				,							A					:       Optional[int]				,							A					:       str				,							A					:       Optional[int]				,							A					:       str				,							A					:       Tuple				,							A					:       int				,							A					:       int				,							A					:       Dict				,							):
         __snake_case:		List[Any]					    =		self.num_choices
         __snake_case:		Optional[Any]					    =		FunnelForMultipleChoice(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		int					    =		input_ids.unsqueeze(1						).expand(-1				,							self.num_choices				,							-1						).contiguous()
         __snake_case:		Tuple					    =		token_type_ids.unsqueeze(1						).expand(-1				,							self.num_choices				,							-1						).contiguous()
         __snake_case:		List[str]					    =		input_mask.unsqueeze(1						).expand(-1				,							self.num_choices				,							-1						).contiguous()
         __snake_case:		Tuple					    =		model(
             A				,							attention_mask=A				,							token_type_ids=A				,							labels=A				,							)
         self.parent.assertEqual(result.logits.shape				,							(self.batch_size, self.num_choices)						)




       def 				UpperCAmelCase__					(    self					:       List[Any]				,							A					:       Optional[int]				,							A					:       str				,							A					:       Union[str, Any]				,							A					:       Optional[int]				,							A					:       Union[str, Any]				,							A					:       Optional[int]				,							A					:       Union[str, Any]				,							A					:       Union[str, Any]				,							):
         __snake_case:		Dict					    =		self.num_labels
         __snake_case:		List[Any]					    =		FunnelForTokenClassification(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		Tuple					    =		model(A				,							attention_mask=A				,							token_type_ids=A				,							labels=A						)
         self.parent.assertEqual(result.logits.shape				,							(self.batch_size, self.seq_length, self.num_labels)						)




       def 				UpperCAmelCase__					(    self					:       Dict				,							A					:       Optional[Any]				,							A					:       int				,							A					:       Dict				,							A					:       Optional[int]				,							A					:       Tuple				,							A					:       str				,							A					:       List[Any]				,							A					:       Optional[Any]				,							):
         __snake_case:		List[str]					    =		FunnelForQuestionAnswering(config=A						)
         model.to(A						)
         model.eval()
         __snake_case:		str					    =		model(
             A				,							attention_mask=A				,							token_type_ids=A				,							start_positions=A				,							end_positions=A				,							)
         self.parent.assertEqual(result.start_logits.shape				,							(self.batch_size, self.seq_length)						)
         self.parent.assertEqual(result.end_logits.shape				,							(self.batch_size, self.seq_length)						)






       def 				UpperCAmelCase__					(    self					:       str						):
         __snake_case:		Optional[int]					    =		self.prepare_config_and_inputs()
         (
             (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  (
             __snake_case
         )    ,  
         ):		Dict					    =		config_and_inputs
         __snake_case:		Union[str, Any]					    =		{"""input_ids""": input_ids, """token_type_ids""": token_type_ids, """attention_mask""": input_mask}
         return config, inputs_dict







@require_torch
class        __snake_case			(				__lowerCamelCase  ,     __lowerCamelCase  ,     unittest.TestCase     ):





       '''simple docstring'''







       lowerCAmelCase__ 						=	False
       lowerCAmelCase__ 						=	False
       lowerCAmelCase__ 						=	(
           (
               FunnelModel,
               FunnelForMaskedLM,
               FunnelForPreTraining,
               FunnelForQuestionAnswering,
               FunnelForTokenClassification,
           )
           if is_torch_available()
           else ()
       )
       lowerCAmelCase__ 						=	(
           {
               """feature-extraction""": (FunnelBaseModel, FunnelModel),
               """fill-mask""": FunnelForMaskedLM,
               """question-answering""": FunnelForQuestionAnswering,
               """text-classification""": FunnelForSequenceClassification,
               """token-classification""": FunnelForTokenClassification,
               """zero-shot""": FunnelForSequenceClassification,
           }
           if is_torch_available()
           else {}
       )




       def 				UpperCAmelCase__					(    self					:       int				,							A					:       Optional[Any]				,							A					:       List[str]				,							A					:       int=False						):
         __snake_case:		str					    =		super()._prepare_for_class(A				,							A				,							return_labels=A						)

         if return_labels:
           if model_class in get_values(A						):
             __snake_case:		str					    =		torch.zeros(
                 (self.model_tester.batch_size, self.model_tester.seq_length)				,							dtype=torch.long				,							device=A						)
         return inputs_dict




       def 				UpperCAmelCase__					(    self					:       List[str]						):
         __snake_case:		Union[str, Any]					    =		FunnelModelTester(self						)
         __snake_case:		List[str]					    =		ConfigTester(self				,							config_class=A						)




       def 				UpperCAmelCase__					(    self					:       Dict						):
         self.config_tester.run_common_tests()




       def 				UpperCAmelCase__					(    self					:       Optional[Any]						):
         __snake_case:		int					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*A						)




       def 				UpperCAmelCase__					(    self					:       Optional[int]						):
         __snake_case:		Optional[int]					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_pretraining(*A						)




       def 				UpperCAmelCase__					(    self					:       str						):
         __snake_case:		Optional[int]					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_masked_lm(*A						)




       def 				UpperCAmelCase__					(    self					:       Optional[int]						):
         __snake_case:		Any					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_token_classification(*A						)




       def 				UpperCAmelCase__					(    self					:       str						):
         __snake_case:		str					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_question_answering(*A						)






       def 				UpperCAmelCase__					(    self					:       Union[str, Any]				,							A					:       List[Any]						):
         if hasattr(A				,							"""weight"""						) and module.weight is not None:
           module.weight.data.fill_(3						)
         if hasattr(A				,							"""bias"""						) and module.bias is not None:
           module.bias.data.fill_(3						)

         for param in ["r_w_bias", "r_r_bias", "r_kernel", "r_s_bias", "seg_embed"]:
           if hasattr(A				,							A						) and getattr(A				,							A						) is not None:
             __snake_case:		str					    =		getattr(A				,							A						)
             weight.data.fill_(3						)







@require_torch
class        __snake_case			(				__lowerCamelCase  ,     unittest.TestCase     ):





       '''simple docstring'''







       lowerCAmelCase__ 						=	False
       lowerCAmelCase__ 						=	False
       lowerCAmelCase__ 						=	(
           (FunnelBaseModel, FunnelForMultipleChoice, FunnelForSequenceClassification) if is_torch_available() else ()
       )




       def 				UpperCAmelCase__					(    self					:       Any						):
         __snake_case:		Optional[int]					    =		FunnelModelTester(self				,							base=A						)
         __snake_case:		int					    =		ConfigTester(self				,							config_class=A						)




       def 				UpperCAmelCase__					(    self					:       Dict						):
         self.config_tester.run_common_tests()




       def 				UpperCAmelCase__					(    self					:       Optional[int]						):
         __snake_case:		List[str]					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_base_model(*A						)




       def 				UpperCAmelCase__					(    self					:       str						):
         __snake_case:		Dict					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_sequence_classification(*A						)




       def 				UpperCAmelCase__					(    self					:       Optional[Any]						):
         __snake_case:		List[str]					    =		self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_multiple_choice(*A						)




       def 				UpperCAmelCase__					(    self					:       Optional[Any]						):
         __snake_case    ,  __snake_case:		Optional[Any]					    =		self.model_tester.prepare_config_and_inputs_for_common()
         __snake_case:		List[str]					    =		True

         for model_class in self.all_model_classes:
           if model_class.__name__ == "FunnelBaseModel":
             continue
           __snake_case:		List[Any]					    =		model_class(A						)
           model.to(A						)
           model.train()
           __snake_case:		Optional[int]					    =		self._prepare_for_class(A				,							A				,							return_labels=A						)
           __snake_case:		Optional[int]					    =		model(**A						).loss
           loss.backward()






       def 				UpperCAmelCase__					(    self					:       Union[str, Any]				,							A					:       Optional[Any]						):
         if hasattr(A				,							"""weight"""						) and module.weight is not None:
           module.weight.data.fill_(3						)
         if hasattr(A				,							"""bias"""						) and module.bias is not None:
           module.bias.data.fill_(3						)

         for param in ["r_w_bias", "r_r_bias", "r_kernel", "r_s_bias", "seg_embed"]:
           if hasattr(A				,							A						) and getattr(A				,							A						) is not None:
             __snake_case:		int					    =		getattr(A				,							A						)
             weight.data.fill_(3						)







@require_torch
@require_sentencepiece
@require_tokenizers
class        __snake_case			(				unittest.TestCase     ):





       '''simple docstring'''







       def 				UpperCAmelCase__					(    self					:       str						):
         __snake_case:		List[str]					    =		13
         __snake_case:		List[Any]					    =		7
         __snake_case:		List[Any]					    =		torch.arange(0				,							batch_size * sequence_length						).long().reshape(A				,							A						)
         __snake_case:		Tuple					    =		[0, 1, 2, 3, 4, 5, 6, 4, 1, 3, 5, 0, 1]
         __snake_case:		List[str]					    =		torch.tensor([[2] + [0] * a + [1] * (sequence_length - a - 1) for a in lengths]						)

         __snake_case:		Optional[int]					    =		FunnelModel.from_pretrained("""sgugger/funnel-random-tiny"""						)
         __snake_case:		List[Any]					    =		model(A				,							token_type_ids=A						)[0].abs()

         __snake_case:		str					    =		torch.tensor(2344.8352						)
         __snake_case:		Union[str, Any]					    =		torch.tensor(0.8052						)
         self.assertTrue(torch.allclose(output.sum()				,							A				,							atol=1E-4						)						)
         self.assertTrue(torch.allclose(output.mean()				,							A				,							atol=1E-4						)						)

         __snake_case:		str					    =		torch.tensor([[1] * 7, [1] * 4 + [0] * 3] * 6 + [[0, 1, 1, 0, 0, 1, 1]]						)
         __snake_case:		Tuple					    =		model(A				,							attention_mask=A				,							token_type_ids=A						)[0].abs()

         __snake_case:		List[str]					    =		torch.tensor(2343.8425						)
         __snake_case:		List[Any]					    =		torch.tensor(0.8049						)
         self.assertTrue(torch.allclose(output.sum()				,							A				,							atol=1E-4						)						)
         self.assertTrue(torch.allclose(output.mean()				,							A				,							atol=1E-4						)						)






       @slow
       def 				UpperCAmelCase__					(    self					:       Optional[Any]						):
         __snake_case:		Union[str, Any]					    =		FunnelTokenizer.from_pretrained("""huggingface/funnel-small"""						)
         __snake_case:		Tuple					    =		FunnelModel.from_pretrained("""huggingface/funnel-small"""						)
         __snake_case:		Tuple					    =		tokenizer("""Hello! I am the Funnel Transformer model."""				,							return_tensors="""pt"""						)
         __snake_case:		Union[str, Any]					    =		model(**A						)[0]

         __snake_case:		int					    =		torch.tensor(235.7246						)
         __snake_case:		Dict					    =		torch.tensor(0.0256						)
         self.assertTrue(torch.allclose(output.sum()				,							A				,							atol=1E-4						)						)
         self.assertTrue(torch.allclose(output.mean()				,							A				,							atol=1E-4						)						)






