






"""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   :						Any        = logging.get_logger(__name__)






class __a   (   _a				):



  def __init__(self  :		Any				,			lowercase_  :		List[str] = None				,			lowercase_  :		List[Any] = None				,			lowercase_  :		List[Any] = None				,			lowercase_  :		str = None				,			lowercase_  :		List[str] = None				,			lowercase_  :		Dict = None				,			lowercase_  :		Optional[Any] = None				,			lowercase_  :		List[Any] = None				,			lowercase_  :		int = None				,			lowercase_  :		List[Any] = (None, None)				,			lowercase_  :		Optional[int] = None				,			)		->							str:
   super().__init__(
       model=lowerCamelCase__				,			args=lowerCamelCase__				,			data_collator=lowerCamelCase__				,			train_dataset=lowerCamelCase__				,			eval_dataset=lowerCamelCase__				,			tokenizer=lowerCamelCase__				,			model_init=lowerCamelCase__				,			compute_metrics=lowerCamelCase__				,			callbacks=lowerCamelCase__				,			optimizers=lowerCamelCase__				,			preprocess_logits_for_metrics=lowerCamelCase__				,			)

   # 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:
    __A	:       Tuple       =	self.load_generation_config(self.args.generation_config   )
    __A	:       Optional[int]       =	gen_config



  @staticmethod
  def 						A_				(lowercase_  :		Union[str, Any]   )		->							Any:

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

   # str or Path
   __A	:       Optional[int]       =	Path(lowerCamelCase__   ) if isinstance(lowerCamelCase__				,			lowerCamelCase__   ) else gen_config_arg
   __A	:       List[Any]       =	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():
    __A	:       Union[str, Any]       =	pretrained_model_name.name
    __A	:       Tuple       =	pretrained_model_name.parent
   # dir path
   elif pretrained_model_name.is_dir():
    pass
   # model id or URL
   else:
    __A	:       Optional[int]       =	gen_config_arg

   __A	:       str       =	GenerationConfig.from_pretrained(lowerCamelCase__				,			lowerCamelCase__   )
   return gen_config



  def 						A_				(self  :		List[str]				,			lowercase_  :		List[Any] = None				,			lowercase_  :		Any = None				,			lowercase_  :		Dict = "eval"				,			**lowercase_  :		str				,			)		->							List[str]:
   __A	:       Optional[Any]       =	gen_kwargs.copy()
   if gen_kwargs.get('max_length'   ) is None and gen_kwargs.get('max_new_tokens'   ) is None:
    __A	:       Dict       =	self.args.generation_max_length
   __A	:       Any       =	(
       gen_kwargs["num_beams"] if gen_kwargs.get('num_beams'   ) is not None else self.args.generation_num_beams
   )
   __A	:       int       =	gen_kwargs

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



  def 						A_				(self  :		Optional[Any]				,			lowercase_  :		Optional[Any]				,			lowercase_  :		List[Any] = None				,			lowercase_  :		Tuple = "test"				,			**lowercase_  :		Union[str, Any]				,			)		->							Optional[Any]:
   __A	:       Optional[Any]       =	gen_kwargs.copy()
   if gen_kwargs.get('max_length'   ) is None and gen_kwargs.get('max_new_tokens'   ) is None:
    __A	:       List[str]       =	self.args.generation_max_length
   __A	:       List[Any]       =	(
       gen_kwargs["num_beams"] if gen_kwargs.get('num_beams'   ) is not None else self.args.generation_num_beams
   )
   __A	:       Any       =	gen_kwargs

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



  def 						A_				(self  :		Union[str, Any]				,			lowercase_  :		Optional[int]				,			lowercase_  :		Any				,			lowercase_  :		Any				,			lowercase_  :		Tuple = None				,			**lowercase_  :		List[Any]				,			)		->							Dict:

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

   __A	:       Tuple       =	"labels" in inputs
   __A	:       int       =	self._prepare_inputs(lowerCamelCase__   )

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

   if len(lowerCamelCase__   ) == 0 and hasattr(self				,			'_gen_kwargs'   ):
    __A	:       Union[str, Any]       =	self._gen_kwargs.copy()

   if gen_kwargs.get('max_length'   ) is None and gen_kwargs.get('max_new_tokens'   ) is None:
    __A	:       List[str]       =	self.model.config.max_length
   __A	:       Dict       =	(
       gen_kwargs["num_beams"] if gen_kwargs.get('num_beams'   ) is not None else self.model.config.num_beams
   )
   __A	:       Any       =	True if is_deepspeed_zeroa_enabled() else False
   __A	:       List[Any]       =	(
       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
   ):
    __A	:       int       =	{k: v for k, v in inputs.items() if k != "decoder_input_ids"}
   __A	:       List[str]       =	self.model.generate(**lowerCamelCase__				,			**lowerCamelCase__   )

   # 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:
    __A	:       List[str]       =	False

   # Retrieves GenerationConfig from model.generation_config
   __A	:       List[Any]       =	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:
    __A	:       int       =	self._pad_tensors_to_max_len(lowerCamelCase__				,			gen_config.max_length   )
   elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
    __A	:       List[str]       =	self._pad_tensors_to_max_len(lowerCamelCase__				,			gen_config.max_new_tokens + 1   )

   with torch.no_grad():
    if has_labels:
     with self.compute_loss_context_manager():
      __A	:       Tuple       =	model(**lowerCamelCase__   )
     if self.label_smoother is not None:
      __A	:       Tuple       =	self.label_smoother(lowerCamelCase__				,			inputs['labels']   ).mean().detach()
     else:
      __A	:       Dict       =	(outputs["loss"] if isinstance(lowerCamelCase__				,			lowerCamelCase__   ) else outputs[0]).mean().detach()
    else:
     __A	:       int       =	None

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

   if has_labels:
    __A	:       Dict       =	inputs["labels"]
    if labels.shape[-1] < gen_config.max_length:
     __A	:       Union[str, Any]       =	self._pad_tensors_to_max_len(lowerCamelCase__				,			gen_config.max_length   )
    elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
     __A	:       Tuple       =	self._pad_tensors_to_max_len(lowerCamelCase__				,			gen_config.max_new_tokens + 1   )
   else:
    __A	:       Union[str, Any]       =	None

   return loss, generated_tokens, labels



  def 						A_				(self  :		Dict				,			lowercase_  :		List[str]				,			lowercase_  :		Any   )		->							str:
   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
    __A	:       Optional[Any]       =	(
        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:
     __A	:       Dict       =	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'   )

   __A	:       Any       =	pad_token_id * torch.ones(
       (tensor.shape[0], max_length)				,			dtype=tensor.dtype				,			device=tensor.device   )
   __A	:       str       =	tensor
   return padded_tensor













"""simple docstring"""


from __future__ import annotations

from typing import Any






class __a   :



  def __init__(self  :		Optional[Any]				,			lowercase_  :		Any				,			lowercase_  :		Any   )		->							None:  # noqa: A002
   __A	:       Optional[Any]       =	item
   __A	:       Union[str, Any]       =	next






class __a   :



  def __init__(self  :		Tuple   )		->							None:
   __A	:       Node | None       =	None
   __A	:       Dict       =	0



  def 						A_				(self  :		str				,			lowercase_  :		Any   )		->							None:
   __A	:       Dict       =	Node(lowercase_				,			self.head   )
   self.size += 1



  def 						A_				(self  :		Optional[int]   )		->							Any:
   # Switched 'self.is_empty()' to 'self.head is None'
   # because mypy was considering the possibility that 'self.head'
   # can be None in below else part and giving error
   if self.head is None:
    return None
   else:
    __A	:       Dict       =	self.head.item
    __A	:       List[str]       =	self.head.next
    self.size -= 1
    return item



  def 						A_				(self  :		List[str]   )		->							bool:
   return self.head is None



  def __str__(self  :		Any   )		->							str:
   if self.is_empty():
    return ""
   else:
    __A	:       List[str]       =	self.head
    __A	:       Union[str, Any]       =	''
    __A	:       list[str]       =	[]
    while iterate:
     item_list.append(str(iterate.item   )   )
     __A	:       List[str]       =	iterate.next

    __A	:       str       =	' --> '.join(lowercase_   )

    return item_str



  def __len__(self  :		Union[str, Any]   )		->							int:
   return self.size





