
'''simple docstring'''



from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import Dataset

from .deepspeed import is_deepspeed_zeroa_enabled
from .generation.configuration_utils import GenerationConfig
from .trainer import Trainer
from .utils import logging


if TYPE_CHECKING:
     from .data.data_collator import DataCollator
     from .modeling_utils import PreTrainedModel
     from .tokenization_utils_base import PreTrainedTokenizerBase
     from .trainer_callback import TrainerCallback
     from .trainer_utils import EvalPrediction, PredictionOutput
     from .training_args import TrainingArguments


UpperCamelCase__					:						Optional[Any]											=     logging.get_logger(__name__)

class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = None				,							A__ = (None, None)				,							A__ = None				,							)			->	Tuple:
          super().__init__(
              model=A__				,							args=A__				,							data_collator=A__				,							train_dataset=A__				,							eval_dataset=A__				,							tokenizer=A__				,							model_init=A__				,							compute_metrics=A__				,							callbacks=A__				,							optimizers=A__				,							preprocess_logits_for_metrics=A__				,							)

          # Override self.model.generation_config if a GenerationConfig is specified in args.
          # Priority: args.generation_config > model.generation_config > default GenerationConfig.
          if self.args.generation_config is not None:
             _SCREAMING_SNAKE_CASE          = self.load_generation_config(self.args.generation_config					)
             _SCREAMING_SNAKE_CASE          = gen_config






       @staticmethod
       def 				UpperCamelCase (							A__					)			->	GenerationConfig:

          # GenerationConfig provided, nothing to do
          if isinstance(A__				,							A__					):
             return deepcopy(A__					)

          # str or Path
          _SCREAMING_SNAKE_CASE          = Path(A__					) if isinstance(A__				,							A__					) else gen_config_arg
          _SCREAMING_SNAKE_CASE          = None

          # Figuring if it is path pointing to a file, pointing to a directory or else a model id or URL
          # This step is required in order to determine config_file_name
          if pretrained_model_name.is_file():
             _SCREAMING_SNAKE_CASE          = pretrained_model_name.name
             _SCREAMING_SNAKE_CASE          = pretrained_model_name.parent
          # dir path
          elif pretrained_model_name.is_dir():
             pass
          # model id or URL
          else:
             _SCREAMING_SNAKE_CASE          = gen_config_arg

          _SCREAMING_SNAKE_CASE          = GenerationConfig.from_pretrained(A__				,							A__					)
          return gen_config






       def 				UpperCamelCase (							self				,							A__ = None				,							A__ = None				,							A__ = "eval"				,							**A__				,							)			->	Dict[str, float]:
          _SCREAMING_SNAKE_CASE          = gen_kwargs.copy()
          if gen_kwargs.get("""max_length"""					) is None and gen_kwargs.get("""max_new_tokens"""					) is None:
             _SCREAMING_SNAKE_CASE          = self.args.generation_max_length
          _SCREAMING_SNAKE_CASE          = (
              gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""					) is not None else self.args.generation_num_beams
          )
          _SCREAMING_SNAKE_CASE          = gen_kwargs

          return super().evaluate(A__				,							ignore_keys=A__				,							metric_key_prefix=A__					)






       def 				UpperCamelCase (							self				,							A__				,							A__ = None				,							A__ = "test"				,							**A__				,							)			->	"PredictionOutput":
          _SCREAMING_SNAKE_CASE          = gen_kwargs.copy()
          if gen_kwargs.get("""max_length"""					) is None and gen_kwargs.get("""max_new_tokens"""					) is None:
             _SCREAMING_SNAKE_CASE          = self.args.generation_max_length
          _SCREAMING_SNAKE_CASE          = (
              gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""					) is not None else self.args.generation_num_beams
          )
          _SCREAMING_SNAKE_CASE          = gen_kwargs

          return super().predict(A__				,							ignore_keys=A__				,							metric_key_prefix=A__					)






       def 				UpperCamelCase (							self				,							A__				,							A__				,							A__				,							A__ = None				,							**A__				,							)			->	Tuple[Optional[float], Optional[torch.Tensor], Optional[torch.Tensor]]:

          if not self.args.predict_with_generate or prediction_loss_only:
             return super().prediction_step(
                 A__				,							A__				,							prediction_loss_only=A__				,							ignore_keys=A__					)

          _SCREAMING_SNAKE_CASE          = """labels""" in inputs
          _SCREAMING_SNAKE_CASE          = self._prepare_inputs(A__					)

          # XXX: adapt synced_gpus for fairscale as well
          # Priority (handled in generate):
          # gen_kwargs > model.generation_config > default GenerationConfig()

          if len(A__					) == 0 and hasattr(self				,							"""_gen_kwargs"""					):
             _SCREAMING_SNAKE_CASE          = self._gen_kwargs.copy()

          if gen_kwargs.get("""max_length"""					) is None and gen_kwargs.get("""max_new_tokens"""					) is None:
             _SCREAMING_SNAKE_CASE          = self.model.config.max_length
          _SCREAMING_SNAKE_CASE          = (
              gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""					) is not None else self.model.config.num_beams
          )
          _SCREAMING_SNAKE_CASE          = True if is_deepspeed_zeroa_enabled() else False
          _SCREAMING_SNAKE_CASE          = (
              gen_kwargs["""synced_gpus"""] if gen_kwargs.get("""synced_gpus"""					) is not None else default_synced_gpus
          )

          # If the `decoder_input_ids` was created from `labels`, evict the former, so that the model can freely generate
          # (otherwise, it would continue generating from the padded `decoder_input_ids`)
          if (
              "labels" in inputs
              and "decoder_input_ids" in inputs
              and inputs["labels"].shape == inputs["decoder_input_ids"].shape
          ):
             _SCREAMING_SNAKE_CASE          = {k: v for k, v in inputs.items() if k != """decoder_input_ids"""}
          _SCREAMING_SNAKE_CASE          = self.model.generate(**A__				,							**A__					)

          # Temporary hack to ensure the generation config is not initialized for each iteration of the evaluation loop
          # TODO: remove this hack when the legacy code that initializes generation_config from a model config is
          # removed in https://github.com/huggingface/transformers/blob/98d88b23f54e5a23e741833f1e973fdf600cc2c5/src/transformers/generation/utils.py#L1183
          if self.model.generation_config._from_model_config:
             _SCREAMING_SNAKE_CASE          = False

          # Retrieves GenerationConfig from model.generation_config
          _SCREAMING_SNAKE_CASE          = self.model.generation_config
          # in case the batch is shorter than max length, the output should be padded
          if generated_tokens.shape[-1] < gen_config.max_length:
             _SCREAMING_SNAKE_CASE          = self._pad_tensors_to_max_len(A__				,							gen_config.max_length					)
          elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
             _SCREAMING_SNAKE_CASE          = self._pad_tensors_to_max_len(A__				,							gen_config.max_new_tokens + 1					)

          with torch.no_grad():
             if has_labels:
                with self.compute_loss_context_manager():
                   _SCREAMING_SNAKE_CASE          = model(**A__					)
                if self.label_smoother is not None:
                   _SCREAMING_SNAKE_CASE          = self.label_smoother(A__				,							inputs["""labels"""]					).mean().detach()
                else:
                   _SCREAMING_SNAKE_CASE          = (outputs["""loss"""] if isinstance(A__				,							A__					) else outputs[0]).mean().detach()
             else:
                _SCREAMING_SNAKE_CASE          = None

          if self.args.prediction_loss_only:
             return loss, None, None

          if has_labels:
             _SCREAMING_SNAKE_CASE          = inputs["""labels"""]
             if labels.shape[-1] < gen_config.max_length:
                _SCREAMING_SNAKE_CASE          = self._pad_tensors_to_max_len(A__				,							gen_config.max_length					)
             elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
                _SCREAMING_SNAKE_CASE          = self._pad_tensors_to_max_len(A__				,							gen_config.max_new_tokens + 1					)
          else:
             _SCREAMING_SNAKE_CASE          = None

          return loss, generated_tokens, labels






       def 				UpperCamelCase (							self				,							A__				,							A__					)			->	Union[str, Any]:
          if self.tokenizer is not None and hasattr(self.tokenizer				,							"""pad_token_id"""					):
             # If PAD token is not defined at least EOS token has to be defined
             _SCREAMING_SNAKE_CASE          = (
                 self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id
             )
          else:
             if self.model.config.pad_token_id is not None:
                _SCREAMING_SNAKE_CASE          = self.model.config.pad_token_id
             else:
                raise ValueError("""Pad_token_id must be set in the configuration of the model, in order to pad tensors"""					)

          _SCREAMING_SNAKE_CASE          = pad_token_id * torch.ones(
              (tensor.shape[0], max_length)				,							dtype=tensor.dtype				,							device=tensor.device					)
          _SCREAMING_SNAKE_CASE          = tensor
          return padded_tensor









import csv
import os

import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.csv import CsvDatasetReader, CsvDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases





def 	snake_case_      (__A     :						Any					,   __A     :						Union[str, Any]     )			->	int:
   assert isinstance(__A					,   __A     )
   assert dataset.num_rows == 4
   assert dataset.num_columns == 3
   assert dataset.column_names == ["col_1", "col_2", "col_3"]
   for feature, expected_dtype in expected_features.items():
      assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize("""keep_in_memory"""					,   [False, True]     )
def 	snake_case_      (__A     :						int					,   __A     :						int					,   __A     :						List[str]     )			->	Tuple:
   __lowerCAmelCase  :			Dict					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			int					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader(__A					,   cache_dir=__A					,   keep_in_memory=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





@pytest.mark.parametrize(
    """features"""					,   [
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]					,   )
def 	snake_case_      (__A     :						Dict					,   __A     :						List[Any]					,   __A     :						List[str]     )			->	Union[str, Any]:
   __lowerCAmelCase  :			str					    =						tmp_path / """cache"""
   # CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
   __lowerCAmelCase  :			Tuple					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			int					    =						features.copy() if features else default_expected_features
   __lowerCAmelCase  :			List[Any]					    =						(
       Features({feature: Value(__A     ) for feature, dtype in features.items()}     ) if features is not None else None
   )
   __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader(__A					,   features=__A					,   cache_dir=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





@pytest.mark.parametrize("""split"""					,   [None, NamedSplit("""train"""     ), """train""", """test"""]     )
def 	snake_case_      (__A     :						str					,   __A     :						Optional[int]					,   __A     :						int     )			->	Union[str, Any]:
   __lowerCAmelCase  :			str					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Tuple					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			Optional[Any]					    =						CsvDatasetReader(__A					,   cache_dir=__A					,   split=__A     ).read()
   _check_csv_dataset(__A					,   __A     )
   assert dataset.split == split if split else "train"





@pytest.mark.parametrize("""path_type"""					,   [str, list]     )
def 	snake_case_      (__A     :						Union[str, Any]					,   __A     :						Any					,   __A     :						List[str]     )			->	Optional[int]:
   if issubclass(__A					,   __A     ):
      __lowerCAmelCase  :			Dict					    =						csv_path
   elif issubclass(__A					,   __A     ):
      __lowerCAmelCase  :			Tuple					    =						[csv_path]
   __lowerCAmelCase  :			Dict					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Dict					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader(__A					,   cache_dir=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





def 	snake_case_      (__A     :						int					,   __A     :						Optional[int]					,   __A     :						Any=("train",)     )			->	Optional[int]:
   assert isinstance(__A					,   __A     )
   for split in splits:
      __lowerCAmelCase  :			int					    =						dataset_dict[split]
      assert dataset.num_rows == 4
      assert dataset.num_columns == 3
      assert dataset.column_names == ["col_1", "col_2", "col_3"]
      for feature, expected_dtype in expected_features.items():
         assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize("""keep_in_memory"""					,   [False, True]     )
def 	snake_case_      (__A     :						Optional[Any]					,   __A     :						List[str]					,   __A     :						Any     )			->	List[str]:
   __lowerCAmelCase  :			int					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Optional[Any]					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A					,   keep_in_memory=__A     ).read()
   _check_csv_datasetdict(__A					,   __A     )





@pytest.mark.parametrize(
    """features"""					,   [
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]					,   )
def 	snake_case_      (__A     :						List[str]					,   __A     :						List[Any]					,   __A     :						Optional[int]     )			->	Optional[int]:
   __lowerCAmelCase  :			Optional[int]					    =						tmp_path / """cache"""
   # CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
   __lowerCAmelCase  :			Union[str, Any]					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			Optional[int]					    =						features.copy() if features else default_expected_features
   __lowerCAmelCase  :			Optional[Any]					    =						(
       Features({feature: Value(__A     ) for feature, dtype in features.items()}     ) if features is not None else None
   )
   __lowerCAmelCase  :			Union[str, Any]					    =						CsvDatasetReader({"""train""": csv_path}					,   features=__A					,   cache_dir=__A     ).read()
   _check_csv_datasetdict(__A					,   __A     )





@pytest.mark.parametrize("""split"""					,   [None, NamedSplit("""train"""     ), """train""", """test"""]     )
def 	snake_case_      (__A     :						Optional[Any]					,   __A     :						Optional[Any]					,   __A     :						Union[str, Any]     )			->	Optional[int]:
   if split:
      __lowerCAmelCase  :			Any					    =						{split: csv_path}
   else:
      __lowerCAmelCase  :			Dict					    =						{"""train""": csv_path, """test""": csv_path}
   __lowerCAmelCase  :			List[str]					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			int					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			int					    =						CsvDatasetReader(__A					,   cache_dir=__A     ).read()
   _check_csv_datasetdict(__A					,   __A					,   splits=list(path.keys()     )     )
   assert all(dataset[split].split == split for split in path.keys()     )





def 	snake_case_      (__A     :						List[str]     )			->	Optional[Any]:
   with open(__A					,   encoding="""utf-8"""     ) as csvfile:
      yield from csv.reader(__A     )





def 	snake_case_      (__A     :						Optional[int]					,   __A     :						Optional[int]     )			->	Union[str, Any]:
   __lowerCAmelCase  :			Optional[Any]					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Tuple					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=1     ).write()

   __lowerCAmelCase  :			Dict					    =						iter_csv_file(__A     )
   __lowerCAmelCase  :			Union[str, Any]					    =						iter_csv_file(__A     )

   for rowa, rowa in zip(__A					,   __A     ):
      assert rowa == rowa





def 	snake_case_      (__A     :						int					,   __A     :						List[str]     )			->	str:
   __lowerCAmelCase  :			Any					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			List[str]					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			Dict					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=2     ).write()

   __lowerCAmelCase  :			Union[str, Any]					    =						iter_csv_file(__A     )
   __lowerCAmelCase  :			Dict					    =						iter_csv_file(__A     )

   for rowa, rowa in zip(__A					,   __A     ):
      assert rowa == rowa





def 	snake_case_      (__A     :						int					,   __A     :						Optional[Any]     )			->	List[str]:
   __lowerCAmelCase  :			Any					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			List[Any]					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   with pytest.raises(__A     ):
      CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=0     )


