






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   UpperCamelCase			(					snake_case__			):







		"""simple docstring"""



		def __init__(	self		:							Any    ,_SCREAMING_SNAKE_CASE		:							Union["PreTrainedModel", nn.Module] = None    ,_SCREAMING_SNAKE_CASE		:							"TrainingArguments" = None    ,_SCREAMING_SNAKE_CASE		:							Optional["DataCollator"] = None    ,_SCREAMING_SNAKE_CASE		:							Optional[Dataset] = None    ,_SCREAMING_SNAKE_CASE		:							Optional[Union[Dataset, Dict[str, Dataset]]] = None    ,_SCREAMING_SNAKE_CASE		:							Optional["PreTrainedTokenizerBase"] = None    ,_SCREAMING_SNAKE_CASE		:							Optional[Callable[[], "PreTrainedModel"]] = None    ,_SCREAMING_SNAKE_CASE		:							Optional[Callable[["EvalPrediction"], Dict]] = None    ,_SCREAMING_SNAKE_CASE		:							Optional[List["TrainerCallback"]] = None    ,_SCREAMING_SNAKE_CASE		:							Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None)    ,_SCREAMING_SNAKE_CASE		:							Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None    ,)     -> Optional[Any]:


						'''simple docstring'''
						super().__init__(
						    model=_SCREAMING_SNAKE_CASE    ,args=_SCREAMING_SNAKE_CASE    ,data_collator=_SCREAMING_SNAKE_CASE    ,train_dataset=_SCREAMING_SNAKE_CASE    ,eval_dataset=_SCREAMING_SNAKE_CASE    ,tokenizer=_SCREAMING_SNAKE_CASE    ,model_init=_SCREAMING_SNAKE_CASE    ,compute_metrics=_SCREAMING_SNAKE_CASE    ,callbacks=_SCREAMING_SNAKE_CASE    ,optimizers=_SCREAMING_SNAKE_CASE    ,preprocess_logits_for_metrics=_SCREAMING_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:
										A									=   self.load_generation_config(self.args.generation_config		)
										A									=   gen_config
		@staticmethod
		def    A(	_SCREAMING_SNAKE_CASE		:							Union[str, GenerationConfig]		)     -> GenerationConfig:


						'''simple docstring'''

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

						# str or Path
						A									=   Path(_SCREAMING_SNAKE_CASE		) if isinstance(_SCREAMING_SNAKE_CASE    ,_SCREAMING_SNAKE_CASE		) else gen_config_arg
						A									=   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():
										A									=   pretrained_model_name.name
										A									=   pretrained_model_name.parent
						# dir path
						elif pretrained_model_name.is_dir():
										pass
						# model id or URL
						else:
										A									=   gen_config_arg

						A									=   GenerationConfig.from_pretrained(_SCREAMING_SNAKE_CASE    ,_SCREAMING_SNAKE_CASE		)
						return gen_config
		def    A(	self		:							Optional[Any]    ,_SCREAMING_SNAKE_CASE		:							Optional[Dataset] = None    ,_SCREAMING_SNAKE_CASE		:							Optional[List[str]] = None    ,_SCREAMING_SNAKE_CASE		:							str = "eval"    ,**_SCREAMING_SNAKE_CASE		:							str    ,)     -> Dict[str, float]:


						'''simple docstring'''
						A									=   gen_kwargs.copy()
						if gen_kwargs.get('max_length'		) is None and gen_kwargs.get('max_new_tokens'		) is None:
										A									=   self.args.generation_max_length
						A									=   (
						    gen_kwargs['num_beams'] if gen_kwargs.get('num_beams'		) is not None else self.args.generation_num_beams
						)
						A									=   gen_kwargs

						return super().evaluate(_SCREAMING_SNAKE_CASE    ,ignore_keys=_SCREAMING_SNAKE_CASE    ,metric_key_prefix=_SCREAMING_SNAKE_CASE		)
		def    A(	self		:							Dict    ,_SCREAMING_SNAKE_CASE		:							Dataset    ,_SCREAMING_SNAKE_CASE		:							Optional[List[str]] = None    ,_SCREAMING_SNAKE_CASE		:							str = "test"    ,**_SCREAMING_SNAKE_CASE		:							int    ,)     -> "PredictionOutput":


						'''simple docstring'''
						A									=   gen_kwargs.copy()
						if gen_kwargs.get('max_length'		) is None and gen_kwargs.get('max_new_tokens'		) is None:
										A									=   self.args.generation_max_length
						A									=   (
						    gen_kwargs['num_beams'] if gen_kwargs.get('num_beams'		) is not None else self.args.generation_num_beams
						)
						A									=   gen_kwargs

						return super().predict(_SCREAMING_SNAKE_CASE    ,ignore_keys=_SCREAMING_SNAKE_CASE    ,metric_key_prefix=_SCREAMING_SNAKE_CASE		)
		def    A(	self		:							Optional[int]    ,_SCREAMING_SNAKE_CASE		:							nn.Module    ,_SCREAMING_SNAKE_CASE		:							Dict[str, Union[torch.Tensor, Any]]    ,_SCREAMING_SNAKE_CASE		:							bool    ,_SCREAMING_SNAKE_CASE		:							Optional[List[str]] = None    ,**_SCREAMING_SNAKE_CASE		:							int    ,)     -> Tuple[Optional[float], Optional[torch.Tensor], Optional[torch.Tensor]]:


						'''simple docstring'''

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

						A									=   'labels' in inputs
						A									=   self._prepare_inputs(_SCREAMING_SNAKE_CASE		)

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

						if len(_SCREAMING_SNAKE_CASE		) == 0 and hasattr(self    ,'_gen_kwargs'		):
										A									=   self._gen_kwargs.copy()

						if gen_kwargs.get('max_length'		) is None and gen_kwargs.get('max_new_tokens'		) is None:
										A									=   self.model.config.max_length
						A									=   (
						    gen_kwargs['num_beams'] if gen_kwargs.get('num_beams'		) is not None else self.model.config.num_beams
						)
						A									=   True if is_deepspeed_zeroa_enabled() else False
						A									=   (
						    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
						):
										A									=   {k: v for k, v in inputs.items() if k != 'decoder_input_ids'}
						A									=   self.model.generate(**_SCREAMING_SNAKE_CASE    ,**_SCREAMING_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:
										A									=   False

						# Retrieves GenerationConfig from model.generation_config
						A									=   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:
										A									=   self._pad_tensors_to_max_len(_SCREAMING_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:
										A									=   self._pad_tensors_to_max_len(_SCREAMING_SNAKE_CASE    ,gen_config.max_new_tokens + 1		)

						with torch.no_grad():
										if has_labels:
														with self.compute_loss_context_manager():
																		A									=   model(**_SCREAMING_SNAKE_CASE		)
														if self.label_smoother is not None:
																		A									=   self.label_smoother(_SCREAMING_SNAKE_CASE    ,inputs['labels']		).mean().detach()
														else:
																		A									=   (outputs['loss'] if isinstance(_SCREAMING_SNAKE_CASE    ,_SCREAMING_SNAKE_CASE		) else outputs[0]).mean().detach()
										else:
														A									=   None

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

						if has_labels:
										A									=   inputs['labels']
										if labels.shape[-1] < gen_config.max_length:
														A									=   self._pad_tensors_to_max_len(_SCREAMING_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:
														A									=   self._pad_tensors_to_max_len(_SCREAMING_SNAKE_CASE    ,gen_config.max_new_tokens + 1		)
						else:
										A									=   None

						return loss, generated_tokens, labels







		def    A(	self		:							List[str]    ,_SCREAMING_SNAKE_CASE		:							str    ,_SCREAMING_SNAKE_CASE		:							Any		)     -> Optional[int]:


						'''simple docstring'''
						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
										A									=   (
										    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:
														A									=   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'		)

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













import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
					import torch

if is_vision_available():
					from PIL import Image

					from transformers import CLIPImageProcessor
class   UpperCamelCase			(					unittest.TestCase			):







		"""simple docstring"""



		def __init__(	self		:							int    ,_SCREAMING_SNAKE_CASE		:							Any    ,_SCREAMING_SNAKE_CASE		:							Optional[int]=7    ,_SCREAMING_SNAKE_CASE		:							List[Any]=3    ,_SCREAMING_SNAKE_CASE		:							int=1_8    ,_SCREAMING_SNAKE_CASE		:							Tuple=3_0    ,_SCREAMING_SNAKE_CASE		:							Optional[Any]=4_0_0    ,_SCREAMING_SNAKE_CASE		:							Dict=True    ,_SCREAMING_SNAKE_CASE		:							int=None    ,_SCREAMING_SNAKE_CASE		:							Tuple=True    ,_SCREAMING_SNAKE_CASE		:							Any=None    ,_SCREAMING_SNAKE_CASE		:							Dict=True    ,_SCREAMING_SNAKE_CASE		:							List[str]=[0.48145466, 0.4578275, 0.40821073]    ,_SCREAMING_SNAKE_CASE		:							int=[0.26862954, 0.26130258, 0.27577711]    ,_SCREAMING_SNAKE_CASE		:							Union[str, Any]=True    ,)     -> List[str]:


						'''simple docstring'''
						A									=   size if size is not None else {'shortest_edge': 2_0}
						A									=   crop_size if crop_size is not None else {'height': 1_8, 'width': 1_8}
						A									=   parent
						A									=   batch_size
						A									=   num_channels
						A									=   image_size
						A									=   min_resolution
						A									=   max_resolution
						A									=   do_resize
						A									=   size
						A									=   do_center_crop
						A									=   crop_size
						A									=   do_normalize
						A									=   image_mean
						A									=   image_std
						A									=   do_convert_rgb
		def    A(	self		:							List[str]		)     -> Optional[int]:


						'''simple docstring'''
						return {
						    "do_resize": self.do_resize,
						    "size": self.size,
						    "do_center_crop": self.do_center_crop,
						    "crop_size": self.crop_size,
						    "do_normalize": self.do_normalize,
						    "image_mean": self.image_mean,
						    "image_std": self.image_std,
						    "do_convert_rgb": self.do_convert_rgb,
						}







		def    A(	self		:							Optional[int]    ,_SCREAMING_SNAKE_CASE		:							str=False    ,_SCREAMING_SNAKE_CASE		:							List[str]=False    ,_SCREAMING_SNAKE_CASE		:							Optional[Any]=False		)     -> str:


						'''simple docstring'''
						assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

						if equal_resolution:
										A									=   []
										for i in range(self.batch_size		):
														image_inputs.append(
														    np.random.randint(
														        2_5_5    ,size=(self.num_channels, self.max_resolution, self.max_resolution)    ,dtype=np.uinta		)		)
						else:
										A									=   []
										for i in range(self.batch_size		):
														A			,		A									=   np.random.choice(np.arange(self.min_resolution    ,self.max_resolution		)    ,2		)
														image_inputs.append(np.random.randint(2_5_5    ,size=(self.num_channels, width, height)    ,dtype=np.uinta		)		)

						if not numpify and not torchify:
										# PIL expects the channel dimension as last dimension
										A									=   [Image.fromarray(np.moveaxis(_SCREAMING_SNAKE_CASE    ,0    ,-1		)		) for x in image_inputs]

						if torchify:
										A									=   [torch.from_numpy(_SCREAMING_SNAKE_CASE		) for x in image_inputs]

						return image_inputs


@require_torch
@require_vision
class   UpperCamelCase			(					snake_case__      ,							unittest.TestCase			):







		"""simple docstring"""



		snake_case			 = CLIPImageProcessor if is_vision_available() else None
		def    A(	self		:							Optional[Any]		)     -> Union[str, Any]:


						'''simple docstring'''
						A									=   CLIPImageProcessingTester(self		)
		@property
		def    A(	self		:							Optional[int]		)     -> Any:


						'''simple docstring'''
						return self.image_processor_tester.prepare_image_processor_dict()
		def    A(	self		:							Optional[int]		)     -> Tuple:


						'''simple docstring'''
						A									=   self.image_processing_class(**self.image_processor_dict		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_resize'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'size'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_center_crop'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'center_crop'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_normalize'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'image_mean'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'image_std'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_convert_rgb'		)		)
		def    A(	self		:							Optional[Any]		)     -> int:


						'''simple docstring'''
						A									=   self.image_processing_class.from_dict(self.image_processor_dict		)
						self.assertEqual(image_processor.size    ,{'shortest_edge': 2_0}		)
						self.assertEqual(image_processor.crop_size    ,{'height': 1_8, 'width': 1_8}		)

						A									=   self.image_processing_class.from_dict(self.image_processor_dict    ,size=4_2    ,crop_size=8_4		)
						self.assertEqual(image_processor.size    ,{'shortest_edge': 4_2}		)
						self.assertEqual(image_processor.crop_size    ,{'height': 8_4, 'width': 8_4}		)
		def    A(	self		:							Any		)     -> Union[str, Any]:


						'''simple docstring'''
						pass
		def    A(	self		:							Tuple		)     -> Optional[int]:


						'''simple docstring'''
						# Initialize image_processing
						A									=   self.image_processing_class(**self.image_processor_dict		)
						# create random PIL images
						A									=   self.image_processor_tester.prepare_inputs(equal_resolution=_SCREAMING_SNAKE_CASE		)
						for image in image_inputs:
										self.assertIsInstance(_SCREAMING_SNAKE_CASE    ,Image.Image		)

						# Test not batched input
						A									=   image_processing(image_inputs[0]    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        1,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)

						# Test batched
						A									=   image_processing(_SCREAMING_SNAKE_CASE    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        self.image_processor_tester.batch_size,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)
		def    A(	self		:							int		)     -> Tuple:


						'''simple docstring'''
						# Initialize image_processing
						A									=   self.image_processing_class(**self.image_processor_dict		)
						# create random numpy tensors
						A									=   self.image_processor_tester.prepare_inputs(equal_resolution=_SCREAMING_SNAKE_CASE    ,numpify=_SCREAMING_SNAKE_CASE		)
						for image in image_inputs:
										self.assertIsInstance(_SCREAMING_SNAKE_CASE    ,np.ndarray		)

						# Test not batched input
						A									=   image_processing(image_inputs[0]    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        1,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)

						# Test batched
						A									=   image_processing(_SCREAMING_SNAKE_CASE    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        self.image_processor_tester.batch_size,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)







		def    A(	self		:							str		)     -> str:


						'''simple docstring'''
						# Initialize image_processing
						A									=   self.image_processing_class(**self.image_processor_dict		)
						# create random PyTorch tensors
						A									=   self.image_processor_tester.prepare_inputs(equal_resolution=_SCREAMING_SNAKE_CASE    ,torchify=_SCREAMING_SNAKE_CASE		)
						for image in image_inputs:
										self.assertIsInstance(_SCREAMING_SNAKE_CASE    ,torch.Tensor		)

						# Test not batched input
						A									=   image_processing(image_inputs[0]    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        1,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)

						# Test batched
						A									=   image_processing(_SCREAMING_SNAKE_CASE    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        self.image_processor_tester.batch_size,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)


@require_torch
@require_vision
class   UpperCamelCase			(					snake_case__      ,							unittest.TestCase			):







		"""simple docstring"""



		snake_case			 = CLIPImageProcessor if is_vision_available() else None
		def    A(	self		:							Tuple		)     -> Union[str, Any]:


						'''simple docstring'''
						A									=   CLIPImageProcessingTester(self    ,num_channels=4		)
						A									=   3
		@property
		def    A(	self		:							str		)     -> Dict:


						'''simple docstring'''
						return self.image_processor_tester.prepare_image_processor_dict()
		def    A(	self		:							Any		)     -> int:


						'''simple docstring'''
						A									=   self.image_processing_class(**self.image_processor_dict		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_resize'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'size'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_center_crop'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'center_crop'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_normalize'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'image_mean'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'image_std'		)		)
						self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE    ,'do_convert_rgb'		)		)
		def    A(	self		:							Any		)     -> str:


						'''simple docstring'''
						pass







		def    A(	self		:							Union[str, Any]		)     -> Dict:


						'''simple docstring'''
						# Initialize image_processing
						A									=   self.image_processing_class(**self.image_processor_dict		)
						# create random PIL images
						A									=   self.image_processor_tester.prepare_inputs(equal_resolution=_SCREAMING_SNAKE_CASE		)
						for image in image_inputs:
										self.assertIsInstance(_SCREAMING_SNAKE_CASE    ,Image.Image		)

						# Test not batched input
						A									=   image_processing(image_inputs[0]    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        1,
						        self.expected_encoded_image_num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)

						# Test batched
						A									=   image_processing(_SCREAMING_SNAKE_CASE    ,return_tensors='pt'		).pixel_values
						self.assertEqual(
						    encoded_images.shape    ,(
						        self.image_processor_tester.batch_size,
						        self.expected_encoded_image_num_channels,
						        self.image_processor_tester.crop_size['height'],
						        self.image_processor_tester.crop_size['width'],
						    )    ,)





