

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


lowercase			:				Tuple   =     logging.get_logger(__name__)

lowercase			:				Optional[int]   =     {
    "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 					__lowercase  (						_SCREAMING_SNAKE_CASE				):





							"""simple docstring"""

							def __init__(					self    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=None    ,						*__UpperCAmelCase    ,						**__UpperCAmelCase						)		->       Dict:
														super().__init__(*__UpperCAmelCase    ,						**__UpperCAmelCase						)

														if config is None:
																					assert isinstance(self.model    ,						__UpperCAmelCase						), (
																					    "If no `config` is passed the model to be trained has to be of type `PreTrainedModel`, but is"
																					    f' {self.model.__class__}'
																					)
																					A      :					Tuple      		=					self.model.config
														else:
																					A      :					Union[str, Any]      		=					config

														A      :					Any      		=					data_args
														A      :					List[str]      		=					self.config.tgt_vocab_size if isinstance(self.config    ,						__UpperCAmelCase						) 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:
																					A      :					List[Any]      		=					torch.nn.CrossEntropyLoss(ignore_index=self.config.pad_token_id						)
														else:
																					# dynamically import label_smoothed_nll_loss
																					from utils import label_smoothed_nll_loss

																					A      :					Tuple      		=					label_smoothed_nll_loss




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       Tuple:
														if self.optimizer is None:
																					A      :					str      		=					['''bias''', '''LayerNorm.weight''']
																					A      :					Optional[Any]      		=					[
																					    {
																					        '''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,
																					    },
																					]
																					A      :					List[Any]      		=					Adafactor if self.args.adafactor else AdamW
																					if self.args.adafactor:
																												A      :					Optional[int]      		=					Adafactor
																												A      :					Optional[Any]      		=					{'''scale_parameter''': False, '''relative_step''': False}
																					else:
																												A      :					Optional[Any]      		=					AdamW
																												A      :					List[str]      		=					{
																												    '''betas''': (self.args.adam_betaa, self.args.adam_betaa),
																												    '''eps''': self.args.adam_epsilon,
																												}
																					A      :					Union[str, Any]      		=					self.args.learning_rate
																					if self.sharded_ddp:
																												A      :					Union[str, Any]      		=					OSS(
																												    params=__UpperCAmelCase    ,						optim=__UpperCAmelCase    ,						**__UpperCAmelCase    ,						)
																					else:
																												A      :					Union[str, Any]      		=					optimizer_cls(__UpperCAmelCase    ,						**__UpperCAmelCase						)

														if self.lr_scheduler is None:
																					A      :					Any      		=					self._get_lr_scheduler(__UpperCAmelCase						)
														else:  # ignoring --lr_scheduler
																					logger.warning('''scheduler is passed to `Seq2SeqTrainer`, `--lr_scheduler` arg is ignored.'''						)




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       Dict:
														A      :					Optional[Any]      		=					arg_to_scheduler[self.args.lr_scheduler]
														if self.args.lr_scheduler == "constant":
																					A      :					Optional[Any]      		=					schedule_func(self.optimizer						)
														elif self.args.lr_scheduler == "constant_w_warmup":
																					A      :					Any      		=					schedule_func(self.optimizer    ,						num_warmup_steps=self.args.warmup_steps						)
														else:
																					A      :					Any      		=					schedule_func(
																					    self.optimizer    ,						num_warmup_steps=self.args.warmup_steps    ,						num_training_steps=__UpperCAmelCase						)
														return scheduler




							def       snake_case			(					self						)		->       Optional[torch.utils.data.Sampler]:
														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       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       int:
														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
																												A      :					Any      		=					model(**__UpperCAmelCase    ,						use_cache=__UpperCAmelCase						)[0]
																												A      :					Union[str, Any]      		=					self.loss_fn(logits.view(-1    ,						logits.shape[-1]						)    ,						labels.view(-1						)						)
																					else:
																												# compute usual loss via models
																												A		,		A      :					Dict      		=					model(**__UpperCAmelCase    ,						labels=__UpperCAmelCase    ,						use_cache=__UpperCAmelCase						)[:2]
														else:
																					# compute label smoothed loss
																					A      :					Optional[Any]      		=					model(**__UpperCAmelCase    ,						use_cache=__UpperCAmelCase						)[0]
																					A      :					Tuple      		=					torch.nn.functional.log_softmax(__UpperCAmelCase    ,						dim=-1						)
																					A		,		A      :					str      		=					self.loss_fn(__UpperCAmelCase    ,						__UpperCAmelCase    ,						self.args.label_smoothing    ,						ignore_index=self.config.pad_token_id						)
														return loss, logits




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Optional[Any]:
														A      :					List[Any]      		=					inputs.pop('''labels'''						)
														A		,		A      :					Any      		=					self._compute_loss(__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)
														return loss




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase = None    ,						)		->       Tuple[Optional[float], Optional[torch.Tensor], Optional[torch.Tensor]]:
														A      :					Union[str, Any]      		=					self._prepare_inputs(__UpperCAmelCase						)

														A      :					Tuple      		=					{
														    '''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:
																					A      :					Dict      		=					self.model.generate(
																					    inputs['''input_ids''']    ,						attention_mask=inputs['''attention_mask''']    ,						**__UpperCAmelCase    ,						)
																					# in case the batch is shorter than max length, the output should be padded
																					if generated_tokens.shape[-1] < gen_kwargs["max_length"]:
																												A      :					List[Any]      		=					self._pad_tensors_to_max_len(__UpperCAmelCase    ,						gen_kwargs['''max_length''']						)

														A      :					str      		=					inputs.pop('''labels'''						)
														with torch.no_grad():
																					# compute loss on predict data
																					A		,		A      :					List[str]      		=					self._compute_loss(__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)

														A      :					List[str]      		=					loss.mean().detach()
														if self.args.prediction_loss_only:
																					return (loss, None, None)

														A      :					Any      		=					generated_tokens if self.args.predict_with_generate else logits

														if labels.shape[-1] < gen_kwargs["max_length"]:
																					A      :					Union[str, Any]      		=					self._pad_tensors_to_max_len(__UpperCAmelCase    ,						gen_kwargs['''max_length''']						)

														return (loss, logits, labels)






							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Optional[int]:
														# If PAD token is not defined at least EOS token has to be defined
														A      :					int      		=					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}'						)

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









from typing import Tuple, Union

from ...modeling_outputs import BackboneOutput
from ...modeling_utils import PreTrainedModel
from ...utils import is_timm_available, is_torch_available, requires_backends
from ...utils.backbone_utils import BackboneMixin
from .configuration_timm_backbone import TimmBackboneConfig


if is_timm_available():
		import timm


if is_torch_available():
		from torch import Tensor







class 					__lowercase  (						_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE				):





							"""simple docstring"""

							UpperCAmelCase_						:		int								=	'''pixel_values'''
							UpperCAmelCase_						:		List[str]								=	False
							UpperCAmelCase_						:		Tuple								=	TimmBackboneConfig




							def __init__(					self    ,						__UpperCAmelCase    ,						**__UpperCAmelCase						)		->       int:
														requires_backends(self    ,						'''timm'''						)
														super().__init__(__UpperCAmelCase						)
														A      :					List[str]      		=					config

														if config.backbone is None:
																					raise ValueError('''backbone is not set in the config. Please set it to a timm model name.'''						)

														if config.backbone not in timm.list_models():
																					raise ValueError(f'backbone {config.backbone} is not supported by timm.'						)

														if hasattr(__UpperCAmelCase    ,						'''out_features'''						) and config.out_features is not None:
																					raise ValueError('''out_features is not supported by TimmBackbone. Please use out_indices instead.'''						)

														A      :					str      		=					getattr(__UpperCAmelCase    ,						'''use_pretrained_backbone'''    ,						__UpperCAmelCase						)
														if pretrained is None:
																					raise ValueError('''use_pretrained_backbone is not set in the config. Please set it to True or False.'''						)

														# We just take the final layer by default. This matches the default for the transformers models.
														A      :					Optional[int]      		=					config.out_indices if getattr(__UpperCAmelCase    ,						'''out_indices'''    ,						__UpperCAmelCase						) is not None else (-1,)

														A      :					str      		=					timm.create_model(
														    config.backbone    ,						pretrained=__UpperCAmelCase    ,						features_only=config.features_only    ,						in_chans=config.num_channels    ,						out_indices=__UpperCAmelCase    ,						**__UpperCAmelCase    ,						)
														# These are used to control the output of the model when called. If output_hidden_states is True, then
														# return_layers is modified to include all layers.
														A      :					str      		=					self._backbone.return_layers
														A      :					Any      		=					{layer['''module''']: str(__UpperCAmelCase						) for i, layer in enumerate(self._backbone.feature_info.info						)}
														super()._init_backbone(__UpperCAmelCase						)




							@classmethod
							def       snake_case			(					cls    ,						__UpperCAmelCase    ,						*__UpperCAmelCase    ,						**__UpperCAmelCase						)		->       Any:
														requires_backends(cls    ,						['''vision''', '''timm''']						)
														from ...models.timm_backbone import TimmBackboneConfig

														A      :					Optional[Any]      		=					kwargs.pop('''config'''    ,						TimmBackboneConfig()						)

														A      :					str      		=					kwargs.pop('''use_timm_backbone'''    ,						__UpperCAmelCase						)
														if not use_timm:
																					raise ValueError('''use_timm_backbone must be True for timm backbones'''						)

														A      :					Optional[int]      		=					kwargs.pop('''num_channels'''    ,						config.num_channels						)
														A      :					List[str]      		=					kwargs.pop('''features_only'''    ,						config.features_only						)
														A      :					Optional[int]      		=					kwargs.pop('''use_pretrained_backbone'''    ,						config.use_pretrained_backbone						)
														A      :					Optional[int]      		=					kwargs.pop('''out_indices'''    ,						config.out_indices						)
														A      :					int      		=					TimmBackboneConfig(
														    backbone=__UpperCAmelCase    ,						num_channels=__UpperCAmelCase    ,						features_only=__UpperCAmelCase    ,						use_pretrained_backbone=__UpperCAmelCase    ,						out_indices=__UpperCAmelCase    ,						)
														return super()._from_config(__UpperCAmelCase    ,						**__UpperCAmelCase						)




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       Any:
														pass






							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=None    ,						**__UpperCAmelCase						)		->       Union[BackboneOutput, Tuple[Tensor, ...]]:
														A      :					Optional[int]      		=					return_dict if return_dict is not None else self.config.use_return_dict
														A      :					str      		=					(
														    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
														)
														A      :					Any      		=					output_attentions if output_attentions is not None else self.config.output_attentions

														if output_attentions:
																					raise ValueError('''Cannot output attentions for timm backbones at the moment'''						)

														if output_hidden_states:
																					# We modify the return layers to include all the stages of the backbone
																					A      :					Dict      		=					self._all_layers
																					A      :					int      		=					self._backbone(__UpperCAmelCase    ,						**__UpperCAmelCase						)
																					A      :					Any      		=					self._return_layers
																					A      :					Union[str, Any]      		=					tuple(hidden_states[i] for i in self.out_indices						)
														else:
																					A      :					Optional[int]      		=					self._backbone(__UpperCAmelCase    ,						**__UpperCAmelCase						)
																					A      :					List[Any]      		=					None

														A      :					str      		=					tuple(__UpperCAmelCase						)
														A      :					int      		=					tuple(__UpperCAmelCase						) if hidden_states is not None else None

														if not return_dict:
																					A      :					Optional[Any]      		=					(feature_maps,)
																					if output_hidden_states:
																												A      :					Tuple      		=					output + (hidden_states,)
																					return output

														return BackboneOutput(feature_maps=__UpperCAmelCase    ,						hidden_states=__UpperCAmelCase    ,						attentions=__UpperCAmelCase						)






