






"""simple docstring"""

import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
			import torch_xla.core.xla_model as xm
			import torch_xla.debug.metrics as met

class 						A_							(      _UpperCAmelCase  ):







				def __init__(							self							:						List[str]						,    *__lowerCamelCase							:						Union[str, Any]						,    __lowerCamelCase							:						Tuple=None						,    __lowerCamelCase							:						Union[str, Any]=None						,    **__lowerCamelCase							:						str	)				->					Union[str, Any]:
										super().__init__(*lowercase__						,    **lowercase__	)
										__magic_name__            =     eval_examples
										__magic_name__            =     post_process_function







				def     _snake_case					(							self							:						List[Any]						,    __lowerCamelCase							:						int = None						,    __lowerCamelCase							:						List[Any]=None						,    __lowerCamelCase							:						List[Any] = None						,    __lowerCamelCase							:						Optional[int] = "eval"						,    **__lowerCamelCase							:						int						,    )				->					Optional[int]:
										__magic_name__            =     gen_kwargs.copy()
										__magic_name__            =     (
										    gen_kwargs["max_length"] if gen_kwargs.get("max_length"	) is not None else self.args.generation_max_length
										)
										__magic_name__            =     (
										    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"	) is not None else self.args.generation_num_beams
										)
										__magic_name__            =     gen_kwargs

										__magic_name__            =     self.eval_dataset if eval_dataset is None else eval_dataset
										__magic_name__            =     self.get_eval_dataloader(lowercase__	)
										__magic_name__            =     self.eval_examples if eval_examples is None else eval_examples

										# Temporarily disable metric computation, we will do it in the loop here.
										__magic_name__            =     self.compute_metrics
										__magic_name__            =     None
										__magic_name__            =     time.time()
										__magic_name__            =     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
																__magic_name__            =     eval_loop(
																    lowercase__						,    description="Evaluation"						,    prediction_loss_only=True if compute_metrics is None else None						,    ignore_keys=lowercase__						,    metric_key_prefix=lowercase__						,    )
										finally:
																__magic_name__            =     compute_metrics
										__magic_name__            =     self.args.eval_batch_size * self.args.world_size
										if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
																start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        lowercase__						,    lowercase__						,    num_samples=output.num_samples						,    num_steps=math.ceil(output.num_samples / total_batch_size	)						,    )	)

										if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
																# Only the main node write the results by default
																__magic_name__            =     self.post_process_function(lowercase__						,    lowercase__						,    lowercase__	)
																__magic_name__            =     self.compute_metrics(lowercase__	)

																# Prefix all keys with metric_key_prefix + '_'
																for key in list(metrics.keys()	):
																						if not key.startswith(f'''{metric_key_prefix}_'''	):
																												__magic_name__            =     metrics.pop(lowercase__	)

																metrics.update(output.metrics	)
										else:
																__magic_name__            =     output.metrics

										if self.args.should_log:
																# Only the main node log the results by default
																self.log(lowercase__	)

										if self.args.tpu_metrics_debug or self.args.debug:
																# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
																xm.master_print(met.metrics_report()	)

										__magic_name__            =     self.callback_handler.on_evaluate(self.args						,    self.state						,    self.control						,    lowercase__	)
										return metrics







				def     _snake_case					(							self							:						Any						,    __lowerCamelCase							:						Dict						,    __lowerCamelCase							:						str						,    __lowerCamelCase							:						Dict=None						,    __lowerCamelCase							:						Union[str, Any] = "test"						,    **__lowerCamelCase							:						Dict	)				->					str:
										__magic_name__            =     gen_kwargs.copy()

										__magic_name__            =     self.get_test_dataloader(lowercase__	)

										# Temporarily disable metric computation, we will do it in the loop here.
										__magic_name__            =     self.compute_metrics
										__magic_name__            =     None
										__magic_name__            =     time.time()
										__magic_name__            =     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
																__magic_name__            =     eval_loop(
																    lowercase__						,    description="Prediction"						,    prediction_loss_only=True if compute_metrics is None else None						,    ignore_keys=lowercase__						,    metric_key_prefix=lowercase__						,    )
										finally:
																__magic_name__            =     compute_metrics

										__magic_name__            =     self.args.eval_batch_size * self.args.world_size
										if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
																start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        lowercase__						,    lowercase__						,    num_samples=output.num_samples						,    num_steps=math.ceil(output.num_samples / total_batch_size	)						,    )	)
										if self.post_process_function is None or self.compute_metrics is None:
																return output

										__magic_name__            =     self.post_process_function(lowercase__						,    lowercase__						,    lowercase__						,    "predict"	)
										__magic_name__            =     self.compute_metrics(lowercase__	)

										# Prefix all keys with metric_key_prefix + '_'
										for key in list(metrics.keys()	):
																if not key.startswith(f'''{metric_key_prefix}_'''	):
																						__magic_name__            =     metrics.pop(lowercase__	)
										metrics.update(output.metrics	)
										return PredictionOutput(predictions=predictions.predictions						,    label_ids=predictions.label_ids						,    metrics=lowercase__	)











"""simple docstring"""

from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType

class 						A_							(      snake_case_  ):
				UpperCAmelCase__        =				['''image_processor''', '''tokenizer''']
				UpperCAmelCase__        =				'''BlipImageProcessor'''
				UpperCAmelCase__        =				'''AutoTokenizer'''







				def __init__(							self							:						Tuple						,    __lowerCamelCase							:						str						,    __lowerCamelCase							:						int	)				->					str:
										__magic_name__            =     False
										super().__init__(__lowerCamelCase						,    __lowerCamelCase	)
										__magic_name__            =     self.image_processor







				def __call__(							self							:						List[Any]						,    __lowerCamelCase							:						ImageInput = None						,    __lowerCamelCase							:						Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None						,    __lowerCamelCase							:						bool = True						,    __lowerCamelCase							:						Union[bool, str, PaddingStrategy] = False						,    __lowerCamelCase							:						Union[bool, str, TruncationStrategy] = None						,    __lowerCamelCase							:						Optional[int] = None						,    __lowerCamelCase							:						int = 0						,    __lowerCamelCase							:						Optional[int] = None						,    __lowerCamelCase							:						Optional[bool] = None						,    __lowerCamelCase							:						bool = False						,    __lowerCamelCase							:						bool = False						,    __lowerCamelCase							:						bool = False						,    __lowerCamelCase							:						bool = False						,    __lowerCamelCase							:						bool = False						,    __lowerCamelCase							:						bool = True						,    __lowerCamelCase							:						Optional[Union[str, TensorType]] = None						,    **__lowerCamelCase							:						Optional[int]						,    )				->					BatchEncoding:
										if images is None and text is None:
																raise ValueError("You have to specify either images or text."	)

										# Get only text
										if images is None:
																__magic_name__            =     self.tokenizer
																__magic_name__            =     self.tokenizer(
																    text=__lowerCamelCase						,    add_special_tokens=__lowerCamelCase						,    padding=__lowerCamelCase						,    truncation=__lowerCamelCase						,    max_length=__lowerCamelCase						,    stride=__lowerCamelCase						,    pad_to_multiple_of=__lowerCamelCase						,    return_attention_mask=__lowerCamelCase						,    return_overflowing_tokens=__lowerCamelCase						,    return_special_tokens_mask=__lowerCamelCase						,    return_offsets_mapping=__lowerCamelCase						,    return_token_type_ids=__lowerCamelCase						,    return_length=__lowerCamelCase						,    verbose=__lowerCamelCase						,    return_tensors=__lowerCamelCase						,    **__lowerCamelCase						,    )
																return text_encoding

										# add pixel_values
										__magic_name__            =     self.image_processor(__lowerCamelCase						,    return_tensors=__lowerCamelCase	)

										if text is not None:
																__magic_name__            =     self.tokenizer(
																    text=__lowerCamelCase						,    add_special_tokens=__lowerCamelCase						,    padding=__lowerCamelCase						,    truncation=__lowerCamelCase						,    max_length=__lowerCamelCase						,    stride=__lowerCamelCase						,    pad_to_multiple_of=__lowerCamelCase						,    return_attention_mask=__lowerCamelCase						,    return_overflowing_tokens=__lowerCamelCase						,    return_special_tokens_mask=__lowerCamelCase						,    return_offsets_mapping=__lowerCamelCase						,    return_token_type_ids=__lowerCamelCase						,    return_length=__lowerCamelCase						,    verbose=__lowerCamelCase						,    return_tensors=__lowerCamelCase						,    **__lowerCamelCase						,    )
										else:
																__magic_name__            =     None

										if text_encoding is not None:
																encoding_image_processor.update(__lowerCamelCase	)

										return encoding_image_processor







				def     _snake_case					(							self							:						Union[str, Any]						,    *__lowerCamelCase							:						Union[str, Any]						,    **__lowerCamelCase							:						Dict	)				->					Dict:
										return self.tokenizer.batch_decode(*__lowerCamelCase						,    **__lowerCamelCase	)







				def     _snake_case					(							self							:						str						,    *__lowerCamelCase							:						Optional[Any]						,    **__lowerCamelCase							:						Tuple	)				->					Optional[Any]:
										return self.tokenizer.decode(*__lowerCamelCase						,    **__lowerCamelCase	)







				@property
				# Copied from transformers.models.blip.processing_blip.BlipProcessor.model_input_names
				def     _snake_case					(							self							:						List[str]	)				->					Optional[Any]:
										__magic_name__            =     self.tokenizer.model_input_names
										__magic_name__            =     self.image_processor.model_input_names
										return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names	)	)



