



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


__A             =logging.get_logger(__name__)

class 		UpperCAmelCase__      (      a_       ):




			'''simple docstring'''






			def __init__(       self							:			str	,	a_							:			Union[str, Any] = None	,	a_							:			str = None	,	a_							:			List[str] = None	,	a_							:			Optional[int] = None	,	a_							:			int = None	,	a_							:			Dict = None	,	a_							:			int = None	,	a_							:			Optional[Any] = None	,	a_							:			List[str] = None	,	a_							:			Optional[int] = (None, None)	,	a_							:			Any = None	,	):
					'''simple docstring'''





					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   :					List[Any]													=     self.load_generation_config(self.args.generation_config			)
							__UpperCAmelCase   :					Any													=     gen_config




			@staticmethod
			def 			snake_case__					(       a_							:			List[str]			):
					'''simple docstring'''





					if isinstance(snake_case__	,	snake_case__			):
							return deepcopy(snake_case__			)

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

					__UpperCAmelCase   :					str													=     GenerationConfig.from_pretrained(snake_case__	,	snake_case__			)
					return gen_config




			def 			snake_case__					(       self							:			Any	,	a_							:			List[Any] = None	,	a_							:			str = None	,	a_							:			Optional[int] = "eval"	,	**a_							:			Any	,	):
					'''simple docstring'''





					__UpperCAmelCase   :					List[Any]													=     gen_kwargs.copy()
					if gen_kwargs.get('''max_length'''			) is None and gen_kwargs.get('''max_new_tokens'''			) is None:
							__UpperCAmelCase   :					int													=     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   :					int													=     gen_kwargs

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




			def 			snake_case__					(       self							:			str	,	a_							:			Optional[int]	,	a_							:			Union[str, Any] = None	,	a_							:			Union[str, Any] = "test"	,	**a_							:			List[str]	,	):
					'''simple docstring'''





					__UpperCAmelCase   :					Optional[Any]													=     gen_kwargs.copy()
					if gen_kwargs.get('''max_length'''			) is None and gen_kwargs.get('''max_new_tokens'''			) is None:
							__UpperCAmelCase   :					Optional[Any]													=     self.args.generation_max_length
					__UpperCAmelCase   :					Dict													=     (
					    gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''			) is not None else self.args.generation_num_beams
					)
					__UpperCAmelCase   :					Any													=     gen_kwargs

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




			def 			snake_case__					(       self							:			Any	,	a_							:			Optional[int]	,	a_							:			int	,	a_							:			int	,	a_							:			List[Any] = None	,	**a_							:			Tuple	,	):
					'''simple docstring'''





					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   :					Optional[int]													=     '''labels''' in inputs
					__UpperCAmelCase   :					int													=     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   :					List[str]													=     self._gen_kwargs.copy()

					if gen_kwargs.get('''max_length'''			) is None and gen_kwargs.get('''max_new_tokens'''			) is None:
							__UpperCAmelCase   :					Tuple													=     self.model.config.max_length
					__UpperCAmelCase   :					List[str]													=     (
					    gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''			) is not None else self.model.config.num_beams
					)
					__UpperCAmelCase   :					Dict													=     True if is_deepspeed_zeroa_enabled() else False
					__UpperCAmelCase   :					int													=     (
					    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   :					List[str]													=     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   :					Dict													=     False

					# Retrieves GenerationConfig from model.generation_config
					__UpperCAmelCase   :					List[str]													=     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   :					Optional[int]													=     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   :					Tuple													=     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   :					Union[str, Any]													=     model(**snake_case__			)
									if self.label_smoother is not None:
											__UpperCAmelCase   :					int													=     self.label_smoother(snake_case__	,	inputs['''labels''']			).mean().detach()
									else:
											__UpperCAmelCase   :					int													=     (outputs['''loss'''] if isinstance(snake_case__	,	snake_case__			) else outputs[0]).mean().detach()
							else:
									__UpperCAmelCase   :					int													=     None

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

					if has_labels:
							__UpperCAmelCase   :					Optional[Any]													=     inputs['''labels''']
							if labels.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 labels.shape[-1] < gen_config.max_new_tokens + 1:
									__UpperCAmelCase   :					List[str]													=     self._pad_tensors_to_max_len(snake_case__	,	gen_config.max_new_tokens + 1			)
					else:
							__UpperCAmelCase   :					List[str]													=     None

					return loss, generated_tokens, labels




			def 			snake_case__					(       self							:			str	,	a_							:			Union[str, Any]	,	a_							:			List[str]			):
					'''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
							__UpperCAmelCase   :					Any													=     (
							    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   :					Tuple													=     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   :					int													=     pad_token_id * torch.ones(
					    (tensor.shape[0], max_length)	,	dtype=tensor.dtype	,	device=tensor.device			)
					__UpperCAmelCase   :					Union[str, Any]													=     tensor
					return padded_tensor










import unittest

from transformers import (
    MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    PreTrainedTokenizer,
    is_vision_available,
)
from transformers.pipelines import ImageClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
				from PIL import Image
else:

				class 		UpperCAmelCase__      :




							'''simple docstring'''






							@staticmethod
							def 			snake_case__					(       *a_							:			Union[str, Any]	,	**a_							:			Any			):
									'''simple docstring'''





									pass

@is_pipeline_test
@require_torch_or_tf
@require_vision
class 		UpperCAmelCase__      (      unittest.TestCase       ):




			'''simple docstring'''






			UpperCamelCase      						=	MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
			UpperCamelCase      						=	TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING




			def 			snake_case__					(       self							:			Any	,	a_							:			str	,	a_							:			Dict	,	a_							:			Dict			):
					'''simple docstring'''





					__UpperCAmelCase   :					Any													=     ImageClassificationPipeline(model=a_	,	image_processor=a_	,	top_k=2			)
					__UpperCAmelCase   :					int													=     [
					    Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''			),
					    '''http://images.cocodataset.org/val2017/000000039769.jpg''',
					]
					return image_classifier, examples




			def 			snake_case__					(       self							:			str	,	a_							:			Optional[int]	,	a_							:			List[str]			):
					'''simple docstring'''





					__UpperCAmelCase   :					Tuple													=     image_classifier('''./tests/fixtures/tests_samples/COCO/000000039769.png'''			)

					self.assertEqual(
					    a_	,	[
					        {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					        {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					    ]	,	)

					import datasets

					__UpperCAmelCase   :					Optional[Any]													=     datasets.load_dataset('''hf-internal-testing/fixtures_image_utils'''	,	'''image'''	,	split='''test'''			)

					# Accepts URL + PIL.Image + lists
					__UpperCAmelCase   :					Any													=     image_classifier(
					    [
					        Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''			),
					        '''http://images.cocodataset.org/val2017/000000039769.jpg''',
					        # RGBA
					        dataset[0]['''file'''],
					        # LA
					        dataset[1]['''file'''],
					        # L
					        dataset[2]['''file'''],
					    ]			)
					self.assertEqual(
					    a_	,	[
					        [
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					        ],
					        [
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					        ],
					        [
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					        ],
					        [
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					        ],
					        [
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					            {'''score''': ANY(a_			), '''label''': ANY(a_			)},
					        ],
					    ]	,	)




			@require_torch
			def 			snake_case__					(       self							:			int			):
					'''simple docstring'''





					__UpperCAmelCase   :					Dict													=     '''hf-internal-testing/tiny-random-vit'''
					__UpperCAmelCase   :					Optional[int]													=     pipeline('''image-classification'''	,	model=a_			)

					__UpperCAmelCase   :					Optional[int]													=     image_classifier('''http://images.cocodataset.org/val2017/000000039769.jpg'''			)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[{'''label''': '''LABEL_1''', '''score''': 0.5_7_4}, {'''label''': '''LABEL_0''', '''score''': 0.4_2_6}]	,	)

					__UpperCAmelCase   :					Dict													=     image_classifier(
					    [
					        '''http://images.cocodataset.org/val2017/000000039769.jpg''',
					        '''http://images.cocodataset.org/val2017/000000039769.jpg''',
					    ]	,	top_k=2	,	)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[
					        [{'''label''': '''LABEL_1''', '''score''': 0.5_7_4}, {'''label''': '''LABEL_0''', '''score''': 0.4_2_6}],
					        [{'''label''': '''LABEL_1''', '''score''': 0.5_7_4}, {'''label''': '''LABEL_0''', '''score''': 0.4_2_6}],
					    ]	,	)




			@require_tf
			def 			snake_case__					(       self							:			Optional[int]			):
					'''simple docstring'''





					__UpperCAmelCase   :					Tuple													=     '''hf-internal-testing/tiny-random-vit'''
					__UpperCAmelCase   :					Optional[Any]													=     pipeline('''image-classification'''	,	model=a_	,	framework='''tf'''			)

					__UpperCAmelCase   :					Optional[Any]													=     image_classifier('''http://images.cocodataset.org/val2017/000000039769.jpg'''			)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[{'''label''': '''LABEL_1''', '''score''': 0.5_7_4}, {'''label''': '''LABEL_0''', '''score''': 0.4_2_6}]	,	)

					__UpperCAmelCase   :					Any													=     image_classifier(
					    [
					        '''http://images.cocodataset.org/val2017/000000039769.jpg''',
					        '''http://images.cocodataset.org/val2017/000000039769.jpg''',
					    ]	,	top_k=2	,	)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[
					        [{'''label''': '''LABEL_1''', '''score''': 0.5_7_4}, {'''label''': '''LABEL_0''', '''score''': 0.4_2_6}],
					        [{'''label''': '''LABEL_1''', '''score''': 0.5_7_4}, {'''label''': '''LABEL_0''', '''score''': 0.4_2_6}],
					    ]	,	)




			def 			snake_case__					(       self							:			Optional[int]			):
					'''simple docstring'''





					__UpperCAmelCase   :					Union[str, Any]													=     PreTrainedTokenizer()

					# Assert that the pipeline can be initialized with a feature extractor that is not in any mapping
					__UpperCAmelCase   :					Dict													=     pipeline(
					    '''image-classification'''	,	model='''hf-internal-testing/tiny-random-vit'''	,	tokenizer=a_			)

					self.assertIs(image_classifier.tokenizer	,	a_			)




			@slow
			@require_torch
			def 			snake_case__					(       self							:			Optional[int]			):
					'''simple docstring'''





					__UpperCAmelCase   :					List[str]													=     pipeline('''image-classification'''	,	model='''deepmind/vision-perceiver-conv'''			)
					__UpperCAmelCase   :					List[Any]													=     image_classifier('''http://images.cocodataset.org/val2017/000000039769.jpg'''			)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[
					        {'''score''': 0.4_3_8_5, '''label''': '''tabby, tabby cat'''},
					        {'''score''': 0.3_2_1, '''label''': '''tiger cat'''},
					        {'''score''': 0.0_5_0_2, '''label''': '''Egyptian cat'''},
					        {'''score''': 0.0_1_3_7, '''label''': '''crib, cot'''},
					        {'''score''': 0.0_0_7, '''label''': '''radiator'''},
					    ]	,	)

					__UpperCAmelCase   :					List[str]													=     pipeline('''image-classification'''	,	model='''deepmind/vision-perceiver-fourier'''			)
					__UpperCAmelCase   :					List[Any]													=     image_classifier('''http://images.cocodataset.org/val2017/000000039769.jpg'''			)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[
					        {'''score''': 0.5_6_5_8, '''label''': '''tabby, tabby cat'''},
					        {'''score''': 0.1_3_0_9, '''label''': '''tiger cat'''},
					        {'''score''': 0.0_7_2_2, '''label''': '''Egyptian cat'''},
					        {'''score''': 0.0_7_0_7, '''label''': '''remote control, remote'''},
					        {'''score''': 0.0_0_8_2, '''label''': '''computer keyboard, keypad'''},
					    ]	,	)

					__UpperCAmelCase   :					List[str]													=     pipeline('''image-classification'''	,	model='''deepmind/vision-perceiver-learned'''			)
					__UpperCAmelCase   :					Union[str, Any]													=     image_classifier('''http://images.cocodataset.org/val2017/000000039769.jpg'''			)
					self.assertEqual(
					    nested_simplify(a_	,	decimals=4			)	,	[
					        {'''score''': 0.3_0_2_2, '''label''': '''tabby, tabby cat'''},
					        {'''score''': 0.2_3_6_2, '''label''': '''Egyptian cat'''},
					        {'''score''': 0.1_8_5_6, '''label''': '''tiger cat'''},
					        {'''score''': 0.0_3_2_4, '''label''': '''remote control, remote'''},
					        {'''score''': 0.0_0_9_6, '''label''': '''quilt, comforter, comfort, puff'''},
					    ]	,	)





