

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


_lowerCamelCase	 =			logging.get_logger(__name__)
class  _snake_case						(__SCREAMING_SNAKE_CASE):






			def __init__(   self       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = None       ,_snake_case = (None, None)       ,_snake_case = None       ,):
				super().__init__(
				    model=_snake_case       ,args=_snake_case       ,data_collator=_snake_case       ,train_dataset=_snake_case       ,eval_dataset=_snake_case       ,tokenizer=_snake_case       ,model_init=_snake_case       ,compute_metrics=_snake_case       ,callbacks=_snake_case       ,optimizers=_snake_case       ,preprocess_logits_for_metrics=_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:
					UpperCAmelCase_			:	Tuple      =   self.load_generation_config(self.args.generation_config			)
					UpperCAmelCase_			:	Union[str, Any]      =   gen_config






			@staticmethod
			def 						UpperCamelCase__				(   _snake_case			):

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

				# str or Path
				UpperCAmelCase_			:	List[Any]      =   Path(_snake_case			) if isinstance(_snake_case       ,_snake_case			) else gen_config_arg
				UpperCAmelCase_			:	Union[str, Any]      =   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():
					UpperCAmelCase_			:	Optional[Any]      =   pretrained_model_name.name
					UpperCAmelCase_			:	Dict      =   pretrained_model_name.parent
				# dir path
				elif pretrained_model_name.is_dir():
					pass
				# model id or URL
				else:
					UpperCAmelCase_			:	Dict      =   gen_config_arg

				UpperCAmelCase_			:	Tuple      =   GenerationConfig.from_pretrained(_snake_case       ,_snake_case			)
				return gen_config






			def 						UpperCamelCase__				(   self       ,_snake_case = None       ,_snake_case = None       ,_snake_case = "eval"       ,**_snake_case       ,):
				UpperCAmelCase_			:	Dict      =   gen_kwargs.copy()
				if gen_kwargs.get("max_length"			) is None and gen_kwargs.get("max_new_tokens"			) is None:
					UpperCAmelCase_			:	str      =   self.args.generation_max_length
				UpperCAmelCase_			:	List[Any]      =   (
				    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"			) is not None else self.args.generation_num_beams
				)
				UpperCAmelCase_			:	Optional[int]      =   gen_kwargs

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






			def 						UpperCamelCase__				(   self       ,_snake_case       ,_snake_case = None       ,_snake_case = "test"       ,**_snake_case       ,):
				UpperCAmelCase_			:	Tuple      =   gen_kwargs.copy()
				if gen_kwargs.get("max_length"			) is None and gen_kwargs.get("max_new_tokens"			) is None:
					UpperCAmelCase_			:	Any      =   self.args.generation_max_length
				UpperCAmelCase_			:	int      =   (
				    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"			) is not None else self.args.generation_num_beams
				)
				UpperCAmelCase_			:	Tuple      =   gen_kwargs

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






			def 						UpperCamelCase__				(   self       ,_snake_case       ,_snake_case       ,_snake_case       ,_snake_case = None       ,**_snake_case       ,):

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

				UpperCAmelCase_			:	Tuple      =   "labels" in inputs
				UpperCAmelCase_			:	Any      =   self._prepare_inputs(_snake_case			)

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

				if len(_snake_case			) == 0 and hasattr(self       ,"_gen_kwargs"			):
					UpperCAmelCase_			:	int      =   self._gen_kwargs.copy()

				if gen_kwargs.get("max_length"			) is None and gen_kwargs.get("max_new_tokens"			) is None:
					UpperCAmelCase_			:	List[str]      =   self.model.config.max_length
				UpperCAmelCase_			:	Any      =   (
				    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"			) is not None else self.model.config.num_beams
				)
				UpperCAmelCase_			:	Optional[int]      =   True if is_deepspeed_zeroa_enabled() else False
				UpperCAmelCase_			:	List[Any]      =   (
				    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
				):
					UpperCAmelCase_			:	List[str]      =   {k: v for k, v in inputs.items() if k != "decoder_input_ids"}
				UpperCAmelCase_			:	Any      =   self.model.generate(**_snake_case       ,**_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:
					UpperCAmelCase_			:	int      =   False

				# Retrieves GenerationConfig from model.generation_config
				UpperCAmelCase_			:	Tuple      =   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:
					UpperCAmelCase_			:	Any      =   self._pad_tensors_to_max_len(_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:
					UpperCAmelCase_			:	Any      =   self._pad_tensors_to_max_len(_snake_case       ,gen_config.max_new_tokens + 1			)

				with torch.no_grad():
					if has_labels:
						with self.compute_loss_context_manager():
							UpperCAmelCase_			:	Optional[Any]      =   model(**_snake_case			)
						if self.label_smoother is not None:
							UpperCAmelCase_			:	List[Any]      =   self.label_smoother(_snake_case       ,inputs["labels"]			).mean().detach()
						else:
							UpperCAmelCase_			:	str      =   (outputs["loss"] if isinstance(_snake_case       ,_snake_case			) else outputs[0]).mean().detach()
					else:
						UpperCAmelCase_			:	Optional[Any]      =   None

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

				if has_labels:
					UpperCAmelCase_			:	Any      =   inputs["labels"]
					if labels.shape[-1] < gen_config.max_length:
						UpperCAmelCase_			:	str      =   self._pad_tensors_to_max_len(_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:
						UpperCAmelCase_			:	Optional[Any]      =   self._pad_tensors_to_max_len(_snake_case       ,gen_config.max_new_tokens + 1			)
				else:
					UpperCAmelCase_			:	str      =   None

				return loss, generated_tokens, labels






			def 						UpperCamelCase__				(   self       ,_snake_case       ,_snake_case			):
				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
					UpperCAmelCase_			:	Tuple      =   (
					    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:
						UpperCAmelCase_			:	str      =   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"			)

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












"""simple docstring"""



from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


_lowerCAmelCase     :     Any            =	logging.get_logger(__name__)




class        lowerCAmelCase__       (  __magic_name__					):
   SCREAMING_SNAKE_CASE_					      =['''pixel_values''']
   def __init__(						self	:     str				,       snake_case__	:     bool = True				,       snake_case__	:     Optional[Dict[str, int]] = None				,       snake_case__	:     PILImageResampling = PILImageResampling.BILINEAR				,       snake_case__	:     bool = True				,       snake_case__	:     Union[int, float] = 1 / 2_5_5				,       snake_case__	:     bool = True				,       snake_case__	:     Optional[Union[float, List[float]]] = None				,       snake_case__	:     Optional[Union[float, List[float]]] = None				,       **snake_case__	:     Dict				,       ):


        '''simple docstring'''
        super().__init__(**snake_case__				)
        UpperCAmelCase__		:  Dict				=				size if size is not None else {"height": 2_2_4, "width": 2_2_4}
        UpperCAmelCase__		:  Optional[Any]				=				get_size_dict(snake_case__				)
        UpperCAmelCase__		:  Optional[Any]				=				do_resize
        UpperCAmelCase__		:  Any				=				do_rescale
        UpperCAmelCase__		:  Optional[int]				=				do_normalize
        UpperCAmelCase__		:  Dict				=				size
        UpperCAmelCase__		:  Optional[Any]				=				resample
        UpperCAmelCase__		:  List[str]				=				rescale_factor
        UpperCAmelCase__		:  str				=				image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
        UpperCAmelCase__		:  Dict				=				image_std if image_std is not None else IMAGENET_STANDARD_STD
   def 	__a				(						self	:     Optional[Any]				,       snake_case__	:     np.ndarray				,       snake_case__	:     Dict[str, int]				,       snake_case__	:     PILImageResampling = PILImageResampling.BILINEAR				,       snake_case__	:     Optional[Union[str, ChannelDimension]] = None				,       **snake_case__	:     Dict				,       ):


        '''simple docstring'''
        UpperCAmelCase__		:  Optional[int]				=				get_size_dict(snake_case__				)
        if "height" not in size or "width" not in size:
             raise ValueError(f'The `size` dictionary must contain the keys `height` and `width`. Got {size.keys()}'				)
        return resize(
            snake_case__				,       size=(size["height"], size["width"])				,       resample=snake_case__				,       data_format=snake_case__				,       **snake_case__				)
   def 	__a				(						self	:     Any				,       snake_case__	:     np.ndarray				,       snake_case__	:     float				,       snake_case__	:     Optional[Union[str, ChannelDimension]] = None				,       **snake_case__	:     Any				):


        '''simple docstring'''
        return rescale(snake_case__				,       scale=snake_case__				,       data_format=snake_case__				,       **snake_case__				)
   def 	__a				(						self	:     str				,       snake_case__	:     np.ndarray				,       snake_case__	:     Union[float, List[float]]				,       snake_case__	:     Union[float, List[float]]				,       snake_case__	:     Optional[Union[str, ChannelDimension]] = None				,       **snake_case__	:     List[Any]				,       ):


        '''simple docstring'''
        return normalize(snake_case__				,       mean=snake_case__				,       std=snake_case__				,       data_format=snake_case__				,       **snake_case__				)




   def 	__a				(						self	:     Union[str, Any]				,       snake_case__	:     ImageInput				,       snake_case__	:     Optional[bool] = None				,       snake_case__	:     Dict[str, int] = None				,       snake_case__	:     PILImageResampling = None				,       snake_case__	:     Optional[bool] = None				,       snake_case__	:     Optional[float] = None				,       snake_case__	:     Optional[bool] = None				,       snake_case__	:     Optional[Union[float, List[float]]] = None				,       snake_case__	:     Optional[Union[float, List[float]]] = None				,       snake_case__	:     Optional[Union[str, TensorType]] = None				,       snake_case__	:     Union[str, ChannelDimension] = ChannelDimension.FIRST				,       **snake_case__	:     List[Any]				,       ):


        '''simple docstring'''
        UpperCAmelCase__		:  Union[str, Any]				=				do_resize if do_resize is not None else self.do_resize
        UpperCAmelCase__		:  Optional[Any]				=				do_rescale if do_rescale is not None else self.do_rescale
        UpperCAmelCase__		:  str				=				do_normalize if do_normalize is not None else self.do_normalize
        UpperCAmelCase__		:  Dict				=				resample if resample is not None else self.resample
        UpperCAmelCase__		:  List[str]				=				rescale_factor if rescale_factor is not None else self.rescale_factor
        UpperCAmelCase__		:  Optional[int]				=				image_mean if image_mean is not None else self.image_mean
        UpperCAmelCase__		:  Optional[int]				=				image_std if image_std is not None else self.image_std

        UpperCAmelCase__		:  Any				=				size if size is not None else self.size
        UpperCAmelCase__		:  Optional[Any]				=				get_size_dict(snake_case__				)

        UpperCAmelCase__		:  List[Any]				=				make_list_of_images(snake_case__				)

        if not valid_images(snake_case__				):
             raise ValueError(
                 "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
                 "torch.Tensor, tf.Tensor or jax.ndarray."				)

        if do_resize and size is None:
             raise ValueError("Size must be specified if do_resize is True."				)

        if do_rescale and rescale_factor is None:
             raise ValueError("Rescale factor must be specified if do_rescale is True."				)

        # All transformations expect numpy arrays.
        UpperCAmelCase__		:  Optional[int]				=				[to_numpy_array(snake_case__				) for image in images]

        if do_resize:
             UpperCAmelCase__		:  List[Any]				=				[self.resize(image=snake_case__				,       size=snake_case__				,       resample=snake_case__				) for image in images]

        if do_rescale:
             UpperCAmelCase__		:  Tuple				=				[self.rescale(image=snake_case__				,       scale=snake_case__				) for image in images]

        if do_normalize:
             UpperCAmelCase__		:  List[Any]				=				[self.normalize(image=snake_case__				,       mean=snake_case__				,       std=snake_case__				) for image in images]

        UpperCAmelCase__		:  Any				=				[to_channel_dimension_format(snake_case__				,       snake_case__				) for image in images]

        UpperCAmelCase__		:  str				=				{"pixel_values": images}
        return BatchFeature(data=snake_case__				,       tensor_type=snake_case__				)


