


from typing import Any, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import DistributedSampler, RandomSampler

from transformers import PreTrainedModel, Trainer, logging
from transformers.integrations import is_fairscale_available
from transformers.models.fsmt.configuration_fsmt import FSMTConfig
from transformers.optimization import (
    Adafactor,
    AdamW,
    get_constant_schedule,
    get_constant_schedule_with_warmup,
    get_cosine_schedule_with_warmup,
    get_cosine_with_hard_restarts_schedule_with_warmup,
    get_linear_schedule_with_warmup,
    get_polynomial_decay_schedule_with_warmup,
)
from transformers.trainer_pt_utils import get_tpu_sampler
from transformers.training_args import ParallelMode
from transformers.utils import is_torch_tpu_available


if is_fairscale_available():
 from fairscale.optim import OSS


SCREAMING_SNAKE_CASE :   Tuple							      =				logging.get_logger(__name__)

SCREAMING_SNAKE_CASE :   Optional[Any]							      =				{
    "linear": get_linear_schedule_with_warmup,
    "cosine": get_cosine_schedule_with_warmup,
    "cosine_w_restarts": get_cosine_with_hard_restarts_schedule_with_warmup,
    "polynomial": get_polynomial_decay_schedule_with_warmup,
    "constant": get_constant_schedule,
    "constant_w_warmup": get_constant_schedule_with_warmup,
}






class _lowerCamelCase(		_a     ):
       def __init__(      self,   lowerCamelCase=None,   lowerCamelCase=None,   *lowerCamelCase,   **lowerCamelCase) ->     Tuple:




           """simple docstring"""

           super().__init__(*lowerCamelCase,   **lowerCamelCase)

           if config is None:
               assert isinstance(self.model,   lowerCamelCase), (
                   "If no `config` is passed the model to be trained has to be of type `PreTrainedModel`, but is"
                   F''' {self.model.__class__}'''
               )
               _lowercase :							Dict   	=	self.model.config
           else:
               _lowercase :							str   	=	config

           _lowercase :							Union[str, Any]   	=	data_args
           _lowercase :							Union[str, Any]   	=	self.config.tgt_vocab_size if isinstance(self.config,   lowerCamelCase) else self.config.vocab_size

           if self.args.label_smoothing != 0 or (self.data_args is not None and self.data_args.ignore_pad_token_for_loss):
               assert self.config.pad_token_id is not None, (
                   "Make sure that `config.pad_token_id` is correcly defined when ignoring `pad_token` for loss"
                   " calculation or doing label smoothing."
               )

           if self.config.pad_token_id is None and self.config.eos_token_id is not None:
               logger.warning(
                   F'''The `config.pad_token_id` is `None`. Using `config.eos_token_id` = {self.config.eos_token_id} for'''
                   ' padding..')

           if self.args.label_smoothing == 0:
               _lowercase :							int   	=	torch.nn.CrossEntropyLoss(ignore_index=self.config.pad_token_id)
           else:
               # dynamically import label_smoothed_nll_loss
               from utils import label_smoothed_nll_loss

               _lowercase :							Any   	=	label_smoothed_nll_loss
       def     UpperCamelCase  (      self,   lowerCamelCase) ->     Optional[int]:




           """simple docstring"""

           if self.optimizer is None:
               _lowercase :							Optional[int]   	=	['bias', 'LayerNorm.weight']
               _lowercase :							int   	=	[
                   {
                       'params': [p for n, p in self.model.named_parameters() if not any(nd in n for nd in no_decay)],
                       'weight_decay': self.args.weight_decay,
                   },
                   {
                       'params': [p for n, p in self.model.named_parameters() if any(nd in n for nd in no_decay)],
                       'weight_decay': 0.0,
                   },
               ]
               _lowercase :							Optional[Any]   	=	Adafactor if self.args.adafactor else AdamW
               if self.args.adafactor:
                   _lowercase :							Optional[Any]   	=	Adafactor
                   _lowercase :							Union[str, Any]   	=	{'scale_parameter': False, 'relative_step': False}
               else:
                   _lowercase :							Any   	=	AdamW
                   _lowercase :							Tuple   	=	{
                       'betas': (self.args.adam_betaa, self.args.adam_betaa),
                       'eps': self.args.adam_epsilon,
                   }
               _lowercase :							Union[str, Any]   	=	self.args.learning_rate
               if self.sharded_ddp:
                   _lowercase :							Dict   	=	OSS(
                       params=lowerCamelCase,   optim=lowerCamelCase,   **lowerCamelCase,   )
               else:
                   _lowercase :							Optional[int]   	=	optimizer_cls(lowerCamelCase,   **lowerCamelCase)

           if self.lr_scheduler is None:
               _lowercase :							Union[str, Any]   	=	self._get_lr_scheduler(lowerCamelCase)
           else:  # ignoring --lr_scheduler
               logger.warning('scheduler is passed to `Seq2SeqTrainer`, `--lr_scheduler` arg is ignored.')
       def     UpperCamelCase  (      self,   lowerCamelCase) ->     List[str]:




           """simple docstring"""

           _lowercase :							str   	=	arg_to_scheduler[self.args.lr_scheduler]
           if self.args.lr_scheduler == "constant":
               _lowercase :							List[Any]   	=	schedule_func(self.optimizer)
           elif self.args.lr_scheduler == "constant_w_warmup":
               _lowercase :							Tuple   	=	schedule_func(self.optimizer,   num_warmup_steps=self.args.warmup_steps)
           else:
               _lowercase :							Any   	=	schedule_func(
                   self.optimizer,   num_warmup_steps=self.args.warmup_steps,   num_training_steps=lowerCamelCase)
           return scheduler
       def     UpperCamelCase  (      self) ->     Optional[torch.utils.data.Sampler]:




           """simple docstring"""

           if isinstance(self.train_dataset,   torch.utils.data.IterableDataset):
               return None
           elif is_torch_tpu_available():
               return get_tpu_sampler(self.train_dataset)
           else:
               if self.args.sortish_sampler:
                   self.train_dataset.make_sortish_sampler(
                       self.args.per_device_train_batch_size,   distributed=(self.args.parallel_mode == ParallelMode.DISTRIBUTED),   )

               return (
                   RandomSampler(self.train_dataset)
                   if self.args.local_rank == -1
                   else DistributedSampler(self.train_dataset)
               )
       def     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase) ->     List[str]:




           """simple docstring"""

           if self.args.label_smoothing == 0:
               if self.data_args is not None and self.data_args.ignore_pad_token_for_loss:
                   # force training to ignore pad token
                   _lowercase :							Optional[int]   	=	model(**lowerCamelCase,   use_cache=lowerCamelCase)[0]
                   _lowercase :							Optional[int]   	=	self.loss_fn(logits.view(-1,   logits.shape[-1]),   labels.view(-1))
               else:
                   # compute usual loss via models
                   _lowercase :							Dict   	=	model(**lowerCamelCase,   labels=lowerCamelCase,   use_cache=lowerCamelCase)[:2]
           else:
               # compute label smoothed loss
               _lowercase :							Optional[int]   	=	model(**lowerCamelCase,   use_cache=lowerCamelCase)[0]
               _lowercase :							Dict   	=	torch.nn.functional.log_softmax(lowerCamelCase,   dim=-1)
               _lowercase :							Tuple   	=	self.loss_fn(lowerCamelCase,   lowerCamelCase,   self.args.label_smoothing,   ignore_index=self.config.pad_token_id)
           return loss, logits
       def     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase) ->     int:




           """simple docstring"""

           _lowercase :							List[Any]   	=	inputs.pop('labels')
           _lowercase :							Union[str, Any]   	=	self._compute_loss(lowerCamelCase,   lowerCamelCase,   lowerCamelCase)
           return loss
       def     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase = None,   ) ->     Tuple[Optional[float], Optional[torch.Tensor], Optional[torch.Tensor]]:




           """simple docstring"""

           _lowercase :							Any   	=	self._prepare_inputs(lowerCamelCase)

           _lowercase :							List[str]   	=	{
               'max_length': self.data_args.val_max_target_length
               if self.data_args is not None
               else self.config.max_length,
               'num_beams': self.data_args.eval_beams if self.data_args is not None else self.config.num_beams,
           }

           if self.args.predict_with_generate and not self.args.prediction_loss_only:
               _lowercase :							int   	=	self.model.generate(
                   inputs['input_ids'],   attention_mask=inputs['attention_mask'],   **lowerCamelCase,   )
               # in case the batch is shorter than max length, the output should be padded
               if generated_tokens.shape[-1] < gen_kwargs["max_length"]:
                   _lowercase :							Dict   	=	self._pad_tensors_to_max_len(lowerCamelCase,   gen_kwargs['max_length'])

           _lowercase :							Tuple   	=	inputs.pop('labels')
           with torch.no_grad():
               # compute loss on predict data
               _lowercase :							str   	=	self._compute_loss(lowerCamelCase,   lowerCamelCase,   lowerCamelCase)

           _lowercase :							Optional[Any]   	=	loss.mean().detach()
           if self.args.prediction_loss_only:
               return (loss, None, None)

           _lowercase :							Union[str, Any]   	=	generated_tokens if self.args.predict_with_generate else logits

           if labels.shape[-1] < gen_kwargs["max_length"]:
               _lowercase :							List[str]   	=	self._pad_tensors_to_max_len(lowerCamelCase,   gen_kwargs['max_length'])

           return (loss, logits, labels)




       def     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase) ->     List[Any]:




           """simple docstring"""

           _lowercase :							List[str]   	=	self.config.pad_token_id if self.config.pad_token_id is not None else self.config.eos_token_id

           if pad_token_id is None:
               raise ValueError(
                   'Make sure that either `config.pad_token_id` or `config.eos_token_id` is defined if tensor has to be'
                   F''' padded to `max_length`={max_length}''')

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






import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding






class _lowerCamelCase(		_a     ):
       lowercase_				:			int									=    ["""image_processor""", """tokenizer"""]
       lowercase_				:			List[str]									=    """CLIPImageProcessor"""
       lowercase_				:			Union[str, Any]									=    ("""CLIPTokenizer""", """CLIPTokenizerFast""")
       def __init__(      self,   lowerCamelCase=None,   lowerCamelCase=None,   **lowerCamelCase) ->     Dict:




           """simple docstring"""

           _lowercase :							List[Any]   	=	None
           if "feature_extractor" in kwargs:
               warnings.warn(
                   'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
                   ' instead.',   lowerCamelCase,   )
               _lowercase :							Dict   	=	kwargs.pop('feature_extractor')

           _lowercase :							int   	=	image_processor if image_processor is not None else feature_extractor
           if image_processor is None:
               raise ValueError('You need to specify an `image_processor`.')
           if tokenizer is None:
               raise ValueError('You need to specify a `tokenizer`.')

           super().__init__(lowerCamelCase,   lowerCamelCase)
       def __call__(      self,   lowerCamelCase=None,   lowerCamelCase=None,   lowerCamelCase=None,   **lowerCamelCase) ->     Tuple:




           """simple docstring"""

           if text is None and images is None:
               raise ValueError('You have to specify either text or images. Both cannot be none.')

           if text is not None:
               _lowercase :							Optional[Any]   	=	self.tokenizer(lowerCamelCase,   return_tensors=lowerCamelCase,   **lowerCamelCase)

           if images is not None:
               _lowercase :							List[Any]   	=	self.image_processor(lowerCamelCase,   return_tensors=lowerCamelCase,   **lowerCamelCase)

           if text is not None and images is not None:
               _lowercase :							int   	=	image_features.pixel_values
               return encoding
           elif text is not None:
               return encoding
           else:
               return BatchEncoding(data=dict(**lowerCamelCase),   tensor_type=lowerCamelCase)
       def     UpperCamelCase  (      self,   *lowerCamelCase,   **lowerCamelCase) ->     Union[str, Any]:




           """simple docstring"""

           return self.tokenizer.batch_decode(*lowerCamelCase,   **lowerCamelCase)
       def     UpperCamelCase  (      self,   *lowerCamelCase,   **lowerCamelCase) ->     int:




           """simple docstring"""

           return self.tokenizer.decode(*lowerCamelCase,   **lowerCamelCase)
       @property
       def     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

           _lowercase :							int   	=	self.tokenizer.model_input_names
           _lowercase :							Dict   	=	self.image_processor.model_input_names
           return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))
       @property
       def     UpperCamelCase  (      self) ->     List[str]:




           """simple docstring"""

           warnings.warn(
               '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.',   lowerCamelCase,   )
           return self.image_processor_class




       @property
       def     UpperCamelCase  (      self) ->     Optional[Any]:




           """simple docstring"""

           warnings.warn(
               '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.',   lowerCamelCase,   )
           return self.image_processor


