






'''simple docstring'''



import torch
from torch import nn
class UpperCAmelCase_					(    nn.Module				):




      """simple docstring"""




      def __init__( self			:		Union[str, Any]     ,      UpperCAmelCase			:		Optional[int]     ,      UpperCAmelCase			:		Tuple     ,      UpperCAmelCase			:		Union[str, Any]     ,      UpperCAmelCase			:		str     ,      UpperCAmelCase			:		Optional[int]=1     ,      UpperCAmelCase			:		List[str]=False		)  ->  int:


       '''simple docstring'''

       super().__init__()

       lowercase					:   Optional[Any]				  =n_token
       lowercase					:   List[Any]				  =d_embed
       lowercase					:   Dict				  =d_proj

       lowercase					:   Dict				  =cutoffs + [n_token]
       lowercase					:   List[str]				  =[0] + self.cutoffs
       lowercase					:   Union[str, Any]				  =div_val

       lowercase					:   int				  =self.cutoffs[0]
       lowercase					:   Dict				  =len(self.cutoffs		) - 1
       lowercase					:   Dict				  =self.shortlist_size + self.n_clusters

       if self.n_clusters > 0:
        lowercase					:   Any				  =nn.Parameter(torch.zeros(self.n_clusters     ,      self.d_embed		)		)
        lowercase					:   Any				  =nn.Parameter(torch.zeros(self.n_clusters		)		)

       lowercase					:   int				  =nn.ModuleList()
       lowercase					:   str				  =nn.ParameterList()

       if div_val == 1:
        for i in range(len(self.cutoffs		)		):
         if d_proj != d_embed:
          self.out_projs.append(nn.Parameter(torch.FloatTensor(UpperCAmelCase     ,      UpperCAmelCase		)		)		)
         else:
          self.out_projs.append(UpperCAmelCase		)

        self.out_layers.append(nn.Linear(UpperCAmelCase     ,      UpperCAmelCase		)		)
       else:
        for i in range(len(self.cutoffs		)		):
         lowercase  ,		lowercase					:   Union[str, Any]				  =self.cutoff_ends[i], self.cutoff_ends[i + 1]
         lowercase					:   Optional[int]				  =d_embed // (div_val**i)

         self.out_projs.append(nn.Parameter(torch.FloatTensor(UpperCAmelCase     ,      UpperCAmelCase		)		)		)

         self.out_layers.append(nn.Linear(UpperCAmelCase     ,      r_idx - l_idx		)		)

       lowercase					:   Optional[Any]				  =keep_order






      def 				A__      ( self			:		Dict     ,      UpperCAmelCase			:		Dict     ,      UpperCAmelCase			:		Dict     ,      UpperCAmelCase			:		Union[str, Any]     ,      UpperCAmelCase			:		Union[str, Any]		)  ->  List[str]:


       '''simple docstring'''

       if proj is None:
        lowercase					:   Any				  =nn.functional.linear(UpperCAmelCase     ,      UpperCAmelCase     ,      bias=UpperCAmelCase		)
       else:
        # if CUDA_MAJOR <= 9 and CUDA_MINOR <= 1:
        lowercase					:   Dict				  =nn.functional.linear(UpperCAmelCase     ,      proj.t().contiguous()		)
        lowercase					:   Dict				  =nn.functional.linear(UpperCAmelCase     ,      UpperCAmelCase     ,      bias=UpperCAmelCase		)
        # else:
        #     logit = torch.einsum('bd,de,ev->bv', (hidden, proj, weight.t()))
        #     if bias is not None:
        #         logit = logit + bias

       return logit






      def 				A__      ( self			:		Tuple     ,      UpperCAmelCase			:		List[str]     ,      UpperCAmelCase			:		Tuple=None     ,      UpperCAmelCase			:		Dict=False		)  ->  str:


       '''simple docstring'''

       if labels is not None:
        # Shift so that tokens < n predict n
        lowercase					:   Dict				  =hidden[..., :-1, :].contiguous()
        lowercase					:   Dict				  =labels[..., 1:].contiguous()
        lowercase					:   int				  =hidden.view(-1     ,      hidden.size(-1		)		)
        lowercase					:   Optional[Any]				  =labels.view(-1		)
        if hidden.size(0		) != labels.size(0		):
         raise RuntimeError('''Input and labels should have the same size in the batch dimension.'''		)
       else:
        lowercase					:   Tuple				  =hidden.view(-1     ,      hidden.size(-1		)		)

       if self.n_clusters == 0:
        lowercase					:   Any				  =self._compute_logit(UpperCAmelCase     ,      self.out_layers[0].weight     ,      self.out_layers[0].bias     ,      self.out_projs[0]		)
        if labels is not None:
         lowercase					:   List[str]				  =labels != -100
         lowercase					:   List[Any]				  =torch.zeros_like(UpperCAmelCase     ,      dtype=hidden.dtype     ,      device=hidden.device		)
         lowercase					:   Optional[int]				  =(
             -nn.functional.log_softmax(UpperCAmelCase     ,      dim=-1		)[mask].gather(1     ,      labels[mask].unsqueeze(1		)		).squeeze(1		)
         )
        else:
         lowercase					:   Tuple				  =nn.functional.log_softmax(UpperCAmelCase     ,      dim=-1		)
       else:
        # construct weights and biases
        lowercase  ,		lowercase					:   List[str]				  =[], []
        for i in range(len(self.cutoffs		)		):
         if self.div_val == 1:
          lowercase  ,		lowercase					:   Any				  =self.cutoff_ends[i], self.cutoff_ends[i + 1]
          lowercase					:   Optional[Any]				  =self.out_layers[0].weight[l_idx:r_idx]
          lowercase					:   str				  =self.out_layers[0].bias[l_idx:r_idx]
         else:
          lowercase					:   Union[str, Any]				  =self.out_layers[i].weight
          lowercase					:   Dict				  =self.out_layers[i].bias

         if i == 0:
          lowercase					:   List[str]				  =torch.cat([weight_i, self.cluster_weight]     ,      dim=0		)
          lowercase					:   List[str]				  =torch.cat([bias_i, self.cluster_bias]     ,      dim=0		)

         weights.append(UpperCAmelCase		)
         biases.append(UpperCAmelCase		)

        lowercase  ,		lowercase  ,		lowercase					:   Any				  =weights[0], biases[0], self.out_projs[0]

        lowercase					:   Optional[int]				  =self._compute_logit(UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase		)
        lowercase					:   str				  =nn.functional.log_softmax(UpperCAmelCase     ,      dim=1		)

        if labels is None:
         lowercase					:   Any				  =hidden.new_empty((head_logit.size(0		), self.n_token)		)
        else:
         lowercase					:   List[Any]				  =torch.zeros_like(UpperCAmelCase     ,      dtype=hidden.dtype     ,      device=hidden.device		)

        lowercase					:   Dict				  =0
        lowercase					:   List[Any]				  =[0] + self.cutoffs
        for i in range(len(UpperCAmelCase		) - 1		):
         lowercase  ,		lowercase					:   List[Any]				  =cutoff_values[i], cutoff_values[i + 1]

         if labels is not None:
          lowercase					:   List[str]				  =(labels >= l_idx) & (labels < r_idx)
          lowercase					:   Optional[Any]				  =mask_i.nonzero().squeeze()

          if indices_i.numel() == 0:
           continue

          lowercase					:   Any				  =labels.index_select(0     ,      UpperCAmelCase		) - l_idx
          lowercase					:   str				  =head_logprob.index_select(0     ,      UpperCAmelCase		)
          lowercase					:   int				  =hidden.index_select(0     ,      UpperCAmelCase		)
         else:
          lowercase					:   Dict				  =hidden

         if i == 0:
          if labels is not None:
           lowercase					:   Optional[int]				  =head_logprob_i.gather(1     ,      target_i[:, None]		).squeeze(1		)
          else:
           lowercase					:   Union[str, Any]				  =head_logprob[:, : self.cutoffs[0]]
         else:
          lowercase  ,		lowercase  ,		lowercase					:   int				  =weights[i], biases[i], self.out_projs[i]

          lowercase					:   Union[str, Any]				  =self._compute_logit(UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase		)
          lowercase					:   Union[str, Any]				  =nn.functional.log_softmax(UpperCAmelCase     ,      dim=1		)
          lowercase					:   Any				  =self.cutoffs[0] + i - 1  # No probability for the head cluster
          if labels is not None:
           lowercase					:   Any				  =head_logprob_i[:, cluster_prob_idx] + tail_logprob_i.gather(
               1     ,      target_i[:, None]		).squeeze(1		)
          else:
           lowercase					:   int				  =head_logprob[:, cluster_prob_idx, None] + tail_logprob_i
           lowercase					:   Union[str, Any]				  =logprob_i

         if labels is not None:
          if (hasattr(self     ,      '''keep_order'''		) and self.keep_order) or keep_order:
           out.index_copy_(0     ,      UpperCAmelCase     ,      -logprob_i		)
          else:
           out[offset : offset + logprob_i.size(0		)].copy_(-logprob_i		)
          offset += logprob_i.size(0		)

       return out






      def 				A__      ( self			:		List[Any]     ,      UpperCAmelCase			:		List[str]		)  ->  Tuple:


       '''simple docstring'''

       if self.n_clusters == 0:
        lowercase					:   str				  =self._compute_logit(UpperCAmelCase     ,      self.out_layers[0].weight     ,      self.out_layers[0].bias     ,      self.out_projs[0]		)
        return nn.functional.log_softmax(UpperCAmelCase     ,      dim=-1		)
       else:
        # construct weights and biases
        lowercase  ,		lowercase					:   Union[str, Any]				  =[], []
        for i in range(len(self.cutoffs		)		):
         if self.div_val == 1:
          lowercase  ,		lowercase					:   Union[str, Any]				  =self.cutoff_ends[i], self.cutoff_ends[i + 1]
          lowercase					:   int				  =self.out_layers[0].weight[l_idx:r_idx]
          lowercase					:   int				  =self.out_layers[0].bias[l_idx:r_idx]
         else:
          lowercase					:   str				  =self.out_layers[i].weight
          lowercase					:   Union[str, Any]				  =self.out_layers[i].bias

         if i == 0:
          lowercase					:   str				  =torch.cat([weight_i, self.cluster_weight]     ,      dim=0		)
          lowercase					:   Dict				  =torch.cat([bias_i, self.cluster_bias]     ,      dim=0		)

         weights.append(UpperCAmelCase		)
         biases.append(UpperCAmelCase		)

        lowercase  ,		lowercase  ,		lowercase					:   Optional[int]				  =weights[0], biases[0], self.out_projs[0]
        lowercase					:   str				  =self._compute_logit(UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase		)

        lowercase					:   int				  =hidden.new_empty((head_logit.size(0		), self.n_token)		)
        lowercase					:   List[str]				  =nn.functional.log_softmax(UpperCAmelCase     ,      dim=1		)

        lowercase					:   str				  =[0] + self.cutoffs
        for i in range(len(UpperCAmelCase		) - 1		):
         lowercase  ,		lowercase					:   Any				  =cutoff_values[i], cutoff_values[i + 1]

         if i == 0:
          lowercase					:   Optional[int]				  =head_logprob[:, : self.cutoffs[0]]
         else:
          lowercase  ,		lowercase  ,		lowercase					:   List[str]				  =weights[i], biases[i], self.out_projs[i]

          lowercase					:   str				  =self._compute_logit(UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase     ,      UpperCAmelCase		)
          lowercase					:   Dict				  =nn.functional.log_softmax(UpperCAmelCase     ,      dim=1		)

          lowercase					:   Optional[int]				  =head_logprob[:, -i] + tail_logprob_i
          lowercase					:   str				  =logprob_i

        return out











'''simple docstring'''



import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class UpperCAmelCase_					(    __A				):




      """simple docstring"""




      @require_torch
      def 				A__      ( self			:		Tuple		)  ->  Optional[int]:


       '''simple docstring'''

       lowercase					:   Any				  ='''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       lowercase					:   Optional[int]				  ='''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

       lowercase					:   Any				  ='''
import socket
def offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")
socket.socket = offline_socket
        '''

       # Force fetching the files so that we can use the cache
       lowercase					:   Tuple				  ='''hf-internal-testing/tiny-random-bert'''
       BertConfig.from_pretrained(UpperCAmelCase		)
       BertModel.from_pretrained(UpperCAmelCase		)
       BertTokenizer.from_pretrained(UpperCAmelCase		)
       pipeline(task='''fill-mask'''     ,      model=UpperCAmelCase		)

       # baseline - just load from_pretrained with normal network
       lowercase					:   List[str]				  =[sys.executable, '''-c''', '''\n'''.join([load, run, mock]		)]

       # should succeed
       lowercase					:   Tuple				  =self.get_env()
       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       lowercase					:   Optional[Any]				  ='''1'''
       lowercase					:   Any				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      0     ,      result.stderr		)
       self.assertIn('''success'''     ,      result.stdout.decode()		)






      @require_torch
      def 				A__      ( self			:		str		)  ->  List[str]:


       '''simple docstring'''

       lowercase					:   str				  ='''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       lowercase					:   Optional[Any]				  ='''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

       lowercase					:   Optional[int]				  ='''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")
socket.socket = offline_socket
        '''

       # Force fetching the files so that we can use the cache
       lowercase					:   Optional[Any]				  ='''hf-internal-testing/tiny-random-bert'''
       BertConfig.from_pretrained(UpperCAmelCase		)
       BertModel.from_pretrained(UpperCAmelCase		)
       BertTokenizer.from_pretrained(UpperCAmelCase		)
       pipeline(task='''fill-mask'''     ,      model=UpperCAmelCase		)

       # baseline - just load from_pretrained with normal network
       lowercase					:   Optional[Any]				  =[sys.executable, '''-c''', '''\n'''.join([load, run, mock]		)]

       # should succeed
       lowercase					:   str				  =self.get_env()
       lowercase					:   Any				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      0     ,      result.stderr		)
       self.assertIn('''success'''     ,      result.stdout.decode()		)






      @require_torch
      def 				A__      ( self			:		Any		)  ->  Optional[Any]:


       '''simple docstring'''

       lowercase					:   Optional[Any]				  ='''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

       lowercase					:   List[Any]				  ='''
mname = "hf-internal-testing/tiny-random-bert-sharded"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print("success")
        '''

       lowercase					:   int				  ='''
import socket
def offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")
socket.socket = offline_socket
        '''

       # baseline - just load from_pretrained with normal network
       lowercase					:   Tuple				  =[sys.executable, '''-c''', '''\n'''.join([load, run]		)]

       # should succeed
       lowercase					:   Optional[Any]				  =self.get_env()
       lowercase					:   List[Any]				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      0     ,      result.stderr		)
       self.assertIn('''success'''     ,      result.stdout.decode()		)

       # next emulate no network
       lowercase					:   Tuple				  =[sys.executable, '''-c''', '''\n'''.join([load, mock, run]		)]

       # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
       # env["TRANSFORMERS_OFFLINE"] = "0"
       # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
       # self.assertEqual(result.returncode, 1, result.stderr)

       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       lowercase					:   Any				  ='''1'''
       lowercase					:   Optional[Any]				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      0     ,      result.stderr		)
       self.assertIn('''success'''     ,      result.stdout.decode()		)






      @require_torch
      def 				A__      ( self			:		Tuple		)  ->  Optional[int]:


       '''simple docstring'''

       lowercase					:   Optional[int]				  ='''
from transformers import pipeline
        '''
       lowercase					:   List[Any]				  ='''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

       lowercase					:   Tuple				  ='''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
       lowercase					:   Tuple				  =self.get_env()
       lowercase					:   Optional[int]				  ='''1'''
       lowercase					:   Union[str, Any]				  =[sys.executable, '''-c''', '''\n'''.join([load, mock, run]		)]
       lowercase					:   Dict				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      1     ,      result.stderr		)
       self.assertIn(
           '''You cannot infer task automatically within `pipeline` when using offline mode'''     ,      result.stderr.decode().replace('''\n'''     ,      ''''''		)     ,      )






      @require_torch
      def 				A__      ( self			:		int		)  ->  Optional[int]:


       '''simple docstring'''

       lowercase					:   List[str]				  ='''
from transformers import AutoModel
        '''
       lowercase					:   Dict				  ='''
mname = "hf-internal-testing/test_dynamic_model"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print("success")
        '''

       # baseline - just load from_pretrained with normal network
       lowercase					:   Dict				  =[sys.executable, '''-c''', '''\n'''.join([load, run]		)]

       # should succeed
       lowercase					:   Optional[Any]				  =self.get_env()
       lowercase					:   int				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      0     ,      result.stderr		)
       self.assertIn('''success'''     ,      result.stdout.decode()		)

       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       lowercase					:   List[str]				  ='''1'''
       lowercase					:   List[Any]				  =subprocess.run(UpperCAmelCase     ,      env=UpperCAmelCase     ,      check=UpperCAmelCase     ,      capture_output=UpperCAmelCase		)
       self.assertEqual(result.returncode     ,      0     ,      result.stderr		)
       self.assertIn('''success'''     ,      result.stdout.decode()		)



