




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






class       A      (		_a		):
     """simple docstring"""

     def __init__(     self	: List[Any]						,	A__	: Union["PreTrainedModel", nn.Module] = None						,	A__	: "TrainingArguments" = None						,	A__	: Optional["DataCollator"] = None						,	A__	: Optional[Dataset] = None						,	A__	: Optional[Union[Dataset, Dict[str, Dataset]]] = None						,	A__	: Optional["PreTrainedTokenizerBase"] = None						,	A__	: Optional[Callable[[], "PreTrainedModel"]] = None						,	A__	: Optional[Callable[["EvalPrediction"], Dict]] = None						,	A__	: Optional[List["TrainerCallback"]] = None						,	A__	: Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None)						,	A__	: Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None						,	)					->							str:
           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:
                 lowerCamelCase__   	=					self.load_generation_config(self.args.generation_config      )
                 lowerCamelCase__   	=					gen_config




     @staticmethod
     def 			__snake_case (     A__	: Union[str, GenerationConfig]      )					->							int:





           if isinstance(A__						,	A__      ):
                 return deepcopy(A__      )

           # str or Path
           lowerCamelCase__   	=					Path(A__      ) if isinstance(A__						,	A__      ) 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(A__						,	A__      )
           return gen_config




     def 			__snake_case (     self	: List[Any]						,	A__	: Optional[Dataset] = None						,	A__	: Optional[List[str]] = None						,	A__	: str = "eval"						,	**A__	: List[str]						,	)					->							Tuple:
           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(A__						,	ignore_keys=A__						,	metric_key_prefix=A__      )




     def 			__snake_case (     self	: Dict						,	A__	: Dataset						,	A__	: Optional[List[str]] = None						,	A__	: str = "test"						,	**A__	: List[str]						,	)					->							Union[str, Any]:
           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(A__						,	ignore_keys=A__						,	metric_key_prefix=A__      )




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





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

           lowerCamelCase__   	=					'labels' in inputs
           lowerCamelCase__   	=					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'      ):
                 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(**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:
                 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(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:
                 lowerCamelCase__   	=					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():
                             lowerCamelCase__   	=					model(**A__      )
                       if self.label_smoother is not None:
                             lowerCamelCase__   	=					self.label_smoother(A__						,	inputs['labels']      ).mean().detach()
                       else:
                             lowerCamelCase__   	=					(outputs['loss'] if isinstance(A__						,	A__      ) 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(A__						,	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(A__						,	gen_config.max_new_tokens + 1      )
           else:
                 lowerCamelCase__   	=					None

           return loss, generated_tokens, labels




     def 			__snake_case (     self	: int						,	A__	: Union[str, Any]						,	A__	: List[Any]      )					->							int:





           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






# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

from .config_args import default_config_file, load_config_from_file
from .config_utils import SubcommandHelpFormatter


__lowercase				    =      """Update an existing config file with the latest defaults while maintaining the old configuration."""


def        __SCREAMING_SNAKE_CASE	(				__lowercase:		Optional[Any]):



  '''simple docstring'''






  __lowerCAmelCase								=					args.config_file
  if config_file is None and Path(__lowercase).exists():
    __lowerCAmelCase								=					default_config_file
  elif not Path(__lowercase).exists():
    raise ValueError(F'''The passed config file located at {config_file} doesn\'t exist.''')
  __lowerCAmelCase								=					load_config_from_file(__lowercase)

  if config_file.endswith('''.json'''):
    config.to_json_file(__lowercase)
  else:
    config.to_yaml_file(__lowercase)
  return config_file


def        __SCREAMING_SNAKE_CASE	(				__lowercase:		Tuple  ,				__lowercase:		int):



  '''simple docstring'''






  __lowerCAmelCase								=					parser.add_parser('''update'''  ,				parents=__lowercase  ,				help=__lowercase  ,				formatter_class=__lowercase)
  parser.add_argument(
      '''--config_file'''  ,				default=__lowercase  ,				help=(
          '''The path to the config file to update. Will default to a file named default_config.yaml in the cache '''
          '''location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '''
          '''such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '''
          '''with \'huggingface\'.'''
      )  ,				)

  parser.set_defaults(func=__lowercase)
  return parser




def        __SCREAMING_SNAKE_CASE	(				__lowercase:		Any):



  '''simple docstring'''






  __lowerCAmelCase								=					update_config(__lowercase)
  print(F'''Sucessfully updated the configuration file at {config_file}.''')
