






"""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


_SCREAMING_SNAKE_CASE :    Any    =						logging.get_logger(__name__)






class 					a						( __snake_case		):






				def __init__(	self					:							Optional[int]	,						__SCREAMING_SNAKE_CASE					:							Union["PreTrainedModel", nn.Module] = None	,						__SCREAMING_SNAKE_CASE					:							"TrainingArguments" = None	,						__SCREAMING_SNAKE_CASE					:							Optional["DataCollator"] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[Dataset] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[Union[Dataset, Dict[str, Dataset]]] = None	,						__SCREAMING_SNAKE_CASE					:							Optional["PreTrainedTokenizerBase"] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[Callable[[], "PreTrainedModel"]] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[Callable[["EvalPrediction"], Dict]] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[List["TrainerCallback"]] = None	,						__SCREAMING_SNAKE_CASE					:							Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None)	,						__SCREAMING_SNAKE_CASE					:							Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None	,						)					->		Optional[Any]:
						super().__init__(
						    model=__SCREAMING_SNAKE_CASE	,						args=__SCREAMING_SNAKE_CASE	,						data_collator=__SCREAMING_SNAKE_CASE	,						train_dataset=__SCREAMING_SNAKE_CASE	,						eval_dataset=__SCREAMING_SNAKE_CASE	,						tokenizer=__SCREAMING_SNAKE_CASE	,						model_init=__SCREAMING_SNAKE_CASE	,						compute_metrics=__SCREAMING_SNAKE_CASE	,						callbacks=__SCREAMING_SNAKE_CASE	,						optimizers=__SCREAMING_SNAKE_CASE	,						preprocess_logits_for_metrics=__SCREAMING_SNAKE_CASE	,						)

						# 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:
								lowerCamelCase_   					=						self.load_generation_config(self.args.generation_config )
								lowerCamelCase_   					=						gen_config






				@staticmethod
				def        UpperCamelCase	(	__SCREAMING_SNAKE_CASE					:							Union[str, GenerationConfig] )					->		GenerationConfig:

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

						# str or Path
						lowerCamelCase_   					=						Path(__SCREAMING_SNAKE_CASE ) if isinstance(__SCREAMING_SNAKE_CASE	,						__SCREAMING_SNAKE_CASE ) else gen_config_arg
						lowerCamelCase_   					=						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():
								lowerCamelCase_   					=						pretrained_model_name.name
								lowerCamelCase_   					=						pretrained_model_name.parent
						# dir path
						elif pretrained_model_name.is_dir():
								pass
						# model id or URL
						else:
								lowerCamelCase_   					=						gen_config_arg

						lowerCamelCase_   					=						GenerationConfig.from_pretrained(__SCREAMING_SNAKE_CASE	,						__SCREAMING_SNAKE_CASE )
						return gen_config






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

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






				def        UpperCamelCase	(	self					:							Dict	,						__SCREAMING_SNAKE_CASE					:							Dataset	,						__SCREAMING_SNAKE_CASE					:							Optional[List[str]] = None	,						__SCREAMING_SNAKE_CASE					:							str = "test"	,						**__SCREAMING_SNAKE_CASE					:							Union[str, Any]	,						)					->		"PredictionOutput":
						lowerCamelCase_   					=						gen_kwargs.copy()
						if gen_kwargs.get('max_length' ) is None and gen_kwargs.get('max_new_tokens' ) is None:
								lowerCamelCase_   					=						self.args.generation_max_length
						lowerCamelCase_   					=						(
						    gen_kwargs['num_beams'] if gen_kwargs.get('num_beams' ) is not None else self.args.generation_num_beams
						)
						lowerCamelCase_   					=						gen_kwargs

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






				def        UpperCamelCase	(	self					:							Tuple	,						__SCREAMING_SNAKE_CASE					:							nn.Module	,						__SCREAMING_SNAKE_CASE					:							Dict[str, Union[torch.Tensor, Any]]	,						__SCREAMING_SNAKE_CASE					:							bool	,						__SCREAMING_SNAKE_CASE					:							Optional[List[str]] = None	,						**__SCREAMING_SNAKE_CASE					:							List[str]	,						)					->		Tuple[Optional[float], Optional[torch.Tensor], Optional[torch.Tensor]]:

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

						lowerCamelCase_   					=						'labels' in inputs
						lowerCamelCase_   					=						self._prepare_inputs(__SCREAMING_SNAKE_CASE )

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

						if len(__SCREAMING_SNAKE_CASE ) == 0 and hasattr(self	,						'_gen_kwargs' ):
								lowerCamelCase_   					=						self._gen_kwargs.copy()

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

						# 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:
								lowerCamelCase_   					=						False

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

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

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

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

						return loss, generated_tokens, labels






				def        UpperCamelCase	(	self					:							Any	,						__SCREAMING_SNAKE_CASE					:							str	,						__SCREAMING_SNAKE_CASE					:							Optional[int] )					->		Tuple:
						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
								lowerCamelCase_   					=						(
								    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:
										lowerCamelCase_   					=						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' )

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












"""simple docstring"""





from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


_SCREAMING_SNAKE_CASE :    Union[str, Any]    =						{
    '''configuration_decision_transformer''': [
        '''DECISION_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''DecisionTransformerConfig''',
    ],
}

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _SCREAMING_SNAKE_CASE :    Optional[Any]    =						[
      '''DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST''',
      '''DecisionTransformerGPT2Model''',
      '''DecisionTransformerGPT2PreTrainedModel''',
      '''DecisionTransformerModel''',
      '''DecisionTransformerPreTrainedModel''',
  ]


if TYPE_CHECKING:
  from .configuration_decision_transformer import (
      DECISION_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP,
      DecisionTransformerConfig,
  )

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_decision_transformer import (
        DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
        DecisionTransformerGPTaModel,
        DecisionTransformerGPTaPreTrainedModel,
        DecisionTransformerModel,
        DecisionTransformerPreTrainedModel,
    )


else:
  import sys

  _SCREAMING_SNAKE_CASE :    List[str]    =						_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




