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








'''simple docstring'''



from ..utils import deprecate
from .attention_processor import (  # noqa: F401
    Attention,
    AttentionProcessor,
    AttnAddedKVProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRALinearLayer,
    LoRAXFormersAttnProcessor,
    SlicedAttnAddedKVProcessor,
    SlicedAttnProcessor,
    XFormersAttnProcessor,
)
from .attention_processor import AttnProcessor as AttnProcessorRename  # noqa: F401


deprecate(
    "cross_attention",
    "0.20.0",
    "Importing from cross_attention is deprecated. Please import from diffusers.models.attention_processor instead.",
    standard_warn=False,
)


UpperCamelCase__					:						str											=     AttentionProcessor

class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	int:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	Any:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	List[str]:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	Tuple:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	str:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	Union[str, Any]:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	str:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






class       _a (_lowerCamelCase):





       """simple docstring"""




       def __init__(							self				,							*A__				,							**A__					)			->	Any:
          _SCREAMING_SNAKE_CASE          = F"{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'					)					)} instead."
          deprecate("""cross_attention"""				,							"""0.20.0"""				,							A__				,							standard_warn=A__					)
          super().__init__(*A__				,							**A__					)






