

'''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     (     _SCREAMING_SNAKE_CASE       ):
 def __init__( self      ,				*__magic_name__      ,				__magic_name__=None      ,				__magic_name__=None      ,				**__magic_name__						)					->							Tuple:
      super().__init__(*__magic_name__      ,				**__magic_name__						)
      _a			      =							eval_examples
      _a			      =							post_process_function
 def 							__UpperCAmelCase		( self      ,				__magic_name__ = None      ,				__magic_name__=None      ,				__magic_name__ = None      ,				__magic_name__ = "eval"      ,				**__magic_name__      ,				)					->							Dict[str, float]:
      _a			      =							gen_kwargs.copy()
      _a			      =							(
          gen_kwargs['max_length'] if gen_kwargs.get('max_length'						) is not None else 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

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

      # Temporarily disable metric computation, we will do it in the loop here.
      _a			      =							self.compute_metrics
      _a			      =							None
      _a			      =							time.time()
      _a			      =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
      try:
           _a			      =							eval_loop(
               __magic_name__      ,				description='Evaluation'      ,				prediction_loss_only=True if compute_metrics is None else None      ,				ignore_keys=__magic_name__      ,				metric_key_prefix=__magic_name__      ,				)
      finally:
           _a			      =							compute_metrics
      _a			      =							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(
              __magic_name__      ,				__magic_name__      ,				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
           _a			      =							self.post_process_function(__magic_name__      ,				__magic_name__      ,				__magic_name__						)
           _a			      =							self.compute_metrics(__magic_name__						)

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

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

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

      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()						)

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




 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				__magic_name__      ,				__magic_name__=None      ,				__magic_name__ = "test"      ,				**__magic_name__						)					->							Tuple:
      _a			      =							gen_kwargs.copy()

      _a			      =							self.get_test_dataloader(__magic_name__						)

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

      _a			      =							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(
              __magic_name__      ,				__magic_name__      ,				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

      _a			      =							self.post_process_function(__magic_name__      ,				__magic_name__      ,				__magic_name__      ,				'predict'						)
      _a			      =							self.compute_metrics(__magic_name__						)

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





'''simple docstring'''






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

import torch

from ..models.auto import AutoModelForVisualQuestionAnswering, AutoProcessor
from ..utils import requires_backends
from .base import PipelineTool


if TYPE_CHECKING:
 from PIL import Image




class  a     (     _SCREAMING_SNAKE_CASE       ):
 _lowerCAmelCase 					=	"""dandelin/vilt-b32-finetuned-vqa"""
 _lowerCAmelCase 					=	(
     """This is a tool that answers a question about an image. It takes an input named `image` which should be the """
     """image containing the information, as well as a `question` which should be the question in English. It """
     """returns a text that is the answer to the question."""
 )
 _lowerCAmelCase 					=	"""image_qa"""
 _lowerCAmelCase 					=	AutoProcessor
 _lowerCAmelCase 					=	AutoModelForVisualQuestionAnswering

 _lowerCAmelCase 					=	["""image""", """text"""]
 _lowerCAmelCase 					=	["""text"""]
 def __init__( self      ,				*__magic_name__      ,				**__magic_name__						)					->							Tuple:
      requires_backends(self      ,				['vision']						)
      super().__init__(*__magic_name__      ,				**__magic_name__						)
 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				__magic_name__						)					->							Tuple:
      return self.pre_processor(__magic_name__      ,				__magic_name__      ,				return_tensors='pt'						)
 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							Any:
      with torch.no_grad():
           return self.model(**__magic_name__						).logits




 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							Optional[int]:
      _a			      =							outputs.argmax(-1						).item()
      return self.model.config.idalabel[idx]

