




'''simple docstring'''



from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin







class 						lowercase_       (					SCREAMING_SNAKE_CASE__	,    SCREAMING_SNAKE_CASE__	,    SCREAMING_SNAKE_CASE__  ):


   """simple docstring"""

   lowerCamelCase_							    =		[r'''h\.\d+\.attn\.bias''', r'''h\.\d+\.attn\.masked_bias''']



   @register_to_config
   def __init__(       self   :      int    , __lowerCamelCase   :      List[str]    , __lowerCamelCase   :      Optional[int]    , __lowerCamelCase   :      int = None    , __lowerCamelCase   :      Dict = 5_0_2_5_7    , __lowerCamelCase   :      Tuple = 1_0_2_4    , __lowerCamelCase   :      Optional[int] = 7_6_8    , __lowerCamelCase   :      int = 1_2    , __lowerCamelCase   :      Optional[int] = 1_2    , __lowerCamelCase   :      str = None    , __lowerCamelCase   :      Tuple = "gelu_new"    , __lowerCamelCase   :      Optional[Any] = 0.1    , __lowerCamelCase   :      Optional[Any] = 0.1    , __lowerCamelCase   :      Optional[Any] = 0.1    , __lowerCamelCase   :      List[str] = 1e-5    , __lowerCamelCase   :      Any = 0.0_2    , __lowerCamelCase   :      Dict = True    , __lowerCamelCase   :      Any = True    , __lowerCamelCase   :      str = False    , __lowerCamelCase   :      Any = False    , ):

      """simple docstring"""
      super().__init__()

      _SCREAMING_SNAKE_CASE			= prefix_length

      if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
         raise ValueError(
             F"""`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and"""
             F""" `n_embd`: {n_embd} are not equal."""       )

      _SCREAMING_SNAKE_CASE			= prefix_inner_dim
      _SCREAMING_SNAKE_CASE			= prefix_hidden_dim

      _SCREAMING_SNAKE_CASE			= (
          nn.Linear(self.prefix_inner_dim    , self.prefix_hidden_dim       )
          if self.prefix_hidden_dim is not None
          else nn.Identity()
      )
      _SCREAMING_SNAKE_CASE			= (
          nn.Linear(self.prefix_hidden_dim    , __lowerCamelCase       ) if self.prefix_hidden_dim is not None else nn.Identity()
      )

      _SCREAMING_SNAKE_CASE			= GPTaConfig(
          vocab_size=__lowerCamelCase    , n_positions=__lowerCamelCase    , n_embd=__lowerCamelCase    , n_layer=__lowerCamelCase    , n_head=__lowerCamelCase    , n_inner=__lowerCamelCase    , activation_function=__lowerCamelCase    , resid_pdrop=__lowerCamelCase    , embd_pdrop=__lowerCamelCase    , attn_pdrop=__lowerCamelCase    , layer_norm_epsilon=__lowerCamelCase    , initializer_range=__lowerCamelCase    , scale_attn_weights=__lowerCamelCase    , use_cache=__lowerCamelCase    , scale_attn_by_inverse_layer_idx=__lowerCamelCase    , reorder_and_upcast_attn=__lowerCamelCase    , )
      _SCREAMING_SNAKE_CASE			= GPTaLMHeadModel(__lowerCamelCase       )



   def 				lowerCAmelCase_     (       self   :      Optional[Any]    , __lowerCamelCase   :      Optional[Any]    , __lowerCamelCase   :      int    , __lowerCamelCase   :      Optional[Any] = None    , __lowerCamelCase   :      Dict = None    , ):

      """simple docstring"""
      _SCREAMING_SNAKE_CASE			= self.transformer.transformer.wte(__lowerCamelCase       )
      _SCREAMING_SNAKE_CASE			= self.encode_prefix(__lowerCamelCase       )
      _SCREAMING_SNAKE_CASE			= self.decode_prefix(__lowerCamelCase       )
      _SCREAMING_SNAKE_CASE			= torch.cat((prefix_embeds, embedding_text)    , dim=1       )

      if labels is not None:
         _SCREAMING_SNAKE_CASE			= self.get_dummy_token(input_ids.shape[0]    , input_ids.device       )
         _SCREAMING_SNAKE_CASE			= torch.cat((dummy_token, input_ids)    , dim=1       )
      _SCREAMING_SNAKE_CASE			= self.transformer(inputs_embeds=__lowerCamelCase    , labels=__lowerCamelCase    , attention_mask=__lowerCamelCase       )
      if self.prefix_hidden_dim is not None:
         return out, hidden
      else:
         return out



   def 				lowerCAmelCase_     (       self   :      Union[str, Any]    , __lowerCamelCase   :      str    , __lowerCamelCase   :      Tuple       ):

      """simple docstring"""
      return torch.zeros(__lowerCamelCase    , self.prefix_length    , dtype=torch.intaa    , device=__lowerCamelCase       )



   def 				lowerCAmelCase_     (       self   :      int    , __lowerCamelCase   :      str       ):

      """simple docstring"""
      return self.encode_prefix(__lowerCamelCase       )



   @torch.no_grad()
   def 				lowerCAmelCase_     (       self   :      Tuple    , __lowerCamelCase   :      int    , __lowerCamelCase   :      Union[str, Any]    , __lowerCamelCase   :      Optional[int]       ):

      """simple docstring"""
      _SCREAMING_SNAKE_CASE			= torch.split(__lowerCamelCase    , 1    , dim=0       )
      _SCREAMING_SNAKE_CASE			= []
      _SCREAMING_SNAKE_CASE			= []
      for feature in features:
         _SCREAMING_SNAKE_CASE			= self.decode_prefix(feature.to(__lowerCamelCase       )       )  # back to the clip feature
         # Only support beam search for now
         _SCREAMING_SNAKE_CASE			= self.generate_beam(
             input_embeds=__lowerCamelCase    , device=__lowerCamelCase    , eos_token_id=__lowerCamelCase       )
         generated_tokens.append(output_tokens[0]       )
         generated_seq_lengths.append(seq_lengths[0]       )
      _SCREAMING_SNAKE_CASE			= torch.stack(__lowerCamelCase       )
      _SCREAMING_SNAKE_CASE			= torch.stack(__lowerCamelCase       )
      return generated_tokens, generated_seq_lengths



   @torch.no_grad()
   def 				lowerCAmelCase_     (       self   :      List[str]    , __lowerCamelCase   :      Tuple=None    , __lowerCamelCase   :      Tuple=None    , __lowerCamelCase   :      List[str]=None    , __lowerCamelCase   :      int = 5    , __lowerCamelCase   :      List[Any] = 6_7    , __lowerCamelCase   :      str = 1.0    , __lowerCamelCase   :      List[Any] = None    , ):

      """simple docstring"""
      _SCREAMING_SNAKE_CASE			= eos_token_id
      _SCREAMING_SNAKE_CASE			= None
      _SCREAMING_SNAKE_CASE			= None
      _SCREAMING_SNAKE_CASE			= torch.ones(__lowerCamelCase    , device=__lowerCamelCase    , dtype=torch.int       )
      _SCREAMING_SNAKE_CASE			= torch.zeros(__lowerCamelCase    , device=__lowerCamelCase    , dtype=torch.bool       )

      if input_embeds is not None:
         _SCREAMING_SNAKE_CASE			= input_embeds
      else:
         _SCREAMING_SNAKE_CASE			= self.transformer.transformer.wte(__lowerCamelCase       )

      for i in range(__lowerCamelCase       ):
         _SCREAMING_SNAKE_CASE			= self.transformer(inputs_embeds=__lowerCamelCase       )
         _SCREAMING_SNAKE_CASE			= outputs.logits
         _SCREAMING_SNAKE_CASE			= logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
         _SCREAMING_SNAKE_CASE			= logits.softmax(-1       ).log()

         if scores is None:
            _SCREAMING_SNAKE_CASE			= logits.topk(__lowerCamelCase    , -1       )
            _SCREAMING_SNAKE_CASE			= generated.expand(__lowerCamelCase    , *generated.shape[1:]       )
            _SCREAMING_SNAKE_CASE			= next_tokens.permute(1    , 0       ), scores.squeeze(0       )
            if tokens is None:
               _SCREAMING_SNAKE_CASE			= next_tokens
            else:
               _SCREAMING_SNAKE_CASE			= tokens.expand(__lowerCamelCase    , *tokens.shape[1:]       )
               _SCREAMING_SNAKE_CASE			= torch.cat((tokens, next_tokens)    , dim=1       )
         else:
            _SCREAMING_SNAKE_CASE			= -float(np.inf       )
            _SCREAMING_SNAKE_CASE			= 0
            _SCREAMING_SNAKE_CASE			= scores[:, None] + logits
            seq_lengths[~is_stopped] += 1
            _SCREAMING_SNAKE_CASE			= scores_sum / seq_lengths[:, None]
            _SCREAMING_SNAKE_CASE			= scores_sum_average.view(-1       ).topk(__lowerCamelCase    , -1       )
            _SCREAMING_SNAKE_CASE			= next_tokens // scores_sum.shape[1]
            _SCREAMING_SNAKE_CASE			= seq_lengths[next_tokens_source]
            _SCREAMING_SNAKE_CASE			= next_tokens % scores_sum.shape[1]
            _SCREAMING_SNAKE_CASE			= next_tokens.unsqueeze(1       )
            _SCREAMING_SNAKE_CASE			= tokens[next_tokens_source]
            _SCREAMING_SNAKE_CASE			= torch.cat((tokens, next_tokens)    , dim=1       )
            _SCREAMING_SNAKE_CASE			= generated[next_tokens_source]
            _SCREAMING_SNAKE_CASE			= scores_sum_average * seq_lengths
            _SCREAMING_SNAKE_CASE			= is_stopped[next_tokens_source]

         _SCREAMING_SNAKE_CASE			= self.transformer.transformer.wte(next_tokens.squeeze()       ).view(generated.shape[0]    , 1    , -1       )
         _SCREAMING_SNAKE_CASE			= torch.cat((generated, next_token_embed)    , dim=1       )
         _SCREAMING_SNAKE_CASE			= is_stopped + next_tokens.eq(__lowerCamelCase       ).squeeze()
         if is_stopped.all():
            break

      _SCREAMING_SNAKE_CASE			= scores / seq_lengths
      _SCREAMING_SNAKE_CASE			= scores.argsort(descending=__lowerCamelCase       )
      # tokens tensors are already padded to max_seq_length
      _SCREAMING_SNAKE_CASE			= [tokens[i] for i in order]
      _SCREAMING_SNAKE_CASE			= torch.stack(__lowerCamelCase    , dim=0       )
      _SCREAMING_SNAKE_CASE			= torch.tensor([seq_lengths[i] for i in order]    , dtype=seq_lengths.dtype       )
      return output_texts, seq_lengths








"""simple docstring"""



import os
import textwrap

import pyarrow as pa
import pytest

from datasets import ClassLabel, Features, Image
from datasets.packaged_modules.csv.csv import Csv

from ..utils import require_pil







@pytest.fixture
def 						__lowercase	(  snake_case_    :    Optional[int]		)  ->str:



    '''simple docstring'''







    __A       :		Tuple							=	tmp_path / '''file.csv'''
    __A       :		Union[str, Any]							=	textwrap.dedent(
        '''\
        header1,header2
        1,2
        10,20
        '''		)
    with open(snake_case_							,'''w'''		) as f:
        f.write(snake_case_		)
    return str(snake_case_		)







@pytest.fixture
def 						__lowercase	(  snake_case_    :    str		)  ->Optional[Any]:



    '''simple docstring'''







    __A       :		List[str]							=	tmp_path / '''malformed_file.csv'''
    __A       :		Optional[Any]							=	textwrap.dedent(
        '''\
        header1,header2
        1,2
        10,20,
        '''		)
    with open(snake_case_							,'''w'''		) as f:
        f.write(snake_case_		)
    return str(snake_case_		)







@pytest.fixture
def 						__lowercase	(  snake_case_    :    Any							,snake_case_    :    Optional[int]		)  ->Optional[Any]:



    '''simple docstring'''







    __A       :		List[str]							=	tmp_path / '''csv_with_image.csv'''
    __A       :		List[str]							=	textwrap.dedent(
        F"""\
        image
        {image_file}
        """		)
    with open(snake_case_							,'''w'''		) as f:
        f.write(snake_case_		)
    return str(snake_case_		)







@pytest.fixture
def 						__lowercase	(  snake_case_    :    Optional[int]		)  ->Tuple:



    '''simple docstring'''







    __A       :		Dict							=	tmp_path / '''csv_with_label.csv'''
    __A       :		int							=	textwrap.dedent(
        '''\
        label
        good
        bad
        good
        '''		)
    with open(snake_case_							,'''w'''		) as f:
        f.write(snake_case_		)
    return str(snake_case_		)







@pytest.fixture
def 						__lowercase	(  snake_case_    :    Dict		)  ->Optional[Any]:



    '''simple docstring'''







    __A       :		Optional[int]							=	tmp_path / '''csv_with_int_list.csv'''
    __A       :		Union[str, Any]							=	textwrap.dedent(
        '''\
        int_list
        1 2 3
        4 5 6
        7 8 9
        '''		)
    with open(snake_case_							,'''w'''		) as f:
        f.write(snake_case_		)
    return str(snake_case_		)







def 						__lowercase	(  snake_case_    :    List[str]							,snake_case_    :    Optional[int]							,snake_case_    :    Optional[Any]		)  ->Dict:



    '''simple docstring'''







    __A       :		str							=	Csv()
    __A       :		List[Any]							=	csv._generate_tables([[csv_file, malformed_csv_file]]		)
    with pytest.raises(snake_case_							,match='''Error tokenizing data'''		):
        for _ in generator:
            pass
    assert any(
        record.levelname == '''ERROR'''
        and '''Failed to read file''' in record.message
        and os.path.basename(snake_case_		) in record.message
        for record in caplog.records		)







@require_pil
def 						__lowercase	(  snake_case_    :    Optional[int]		)  ->Union[str, Any]:



    '''simple docstring'''







    with open(snake_case_							,encoding='''utf-8'''		) as f:
        __A       :		int							=	f.read().splitlines()[1]
    __A       :		Optional[Any]							=	Csv(encoding='''utf-8'''							,features=Features({'''image''': Image()}		)		)
    __A       :		Union[str, Any]							=	csv._generate_tables([[csv_file_with_image]]		)
    __A       :		Optional[Any]							=	pa.concat_tables([table for _, table in generator]		)
    assert pa_table.schema.field('''image'''		).type == Image()()
    __A       :		Optional[int]							=	pa_table.to_pydict()['''image''']
    assert generated_content == [{"path": image_file, "bytes": None}]







def 						__lowercase	(  snake_case_    :    Any		)  ->List[Any]:



    '''simple docstring'''







    with open(snake_case_							,encoding='''utf-8'''		) as f:
        __A       :		Optional[int]							=	f.read().splitlines()[1:]
    __A       :		List[Any]							=	Csv(encoding='''utf-8'''							,features=Features({'''label''': ClassLabel(names=['''good''', '''bad''']		)}		)		)
    __A       :		str							=	csv._generate_tables([[csv_file_with_label]]		)
    __A       :		Any							=	pa.concat_tables([table for _, table in generator]		)
    assert pa_table.schema.field('''label'''		).type == ClassLabel(names=['''good''', '''bad''']		)()
    __A       :		Optional[Any]							=	pa_table.to_pydict()['''label''']
    assert generated_content == [ClassLabel(names=['''good''', '''bad''']		).straint(snake_case_		) for label in labels]







def 						__lowercase	(  snake_case_    :    str		)  ->Dict:



    '''simple docstring'''







    __A       :		Optional[int]							=	Csv(encoding='''utf-8'''							,sep=''','''							,converters={'''int_list''': lambda snake_case_		: [int(snake_case_		) for i in x.split()]}		)
    __A       :		Dict							=	csv._generate_tables([[csv_file_with_int_list]]		)
    __A       :		Dict							=	pa.concat_tables([table for _, table in generator]		)
    assert pa.types.is_list(pa_table.schema.field('''int_list'''		).type		)
    __A       :		int							=	pa_table.to_pydict()['''int_list''']
    assert generated_content == [[1, 2, 3], [4, 5, 6], [7, 8, 9]]




