



"""simple docstring"""

import unittest
from typing import Dict, List, Optional, Union

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, prepare_image_inputs


if is_torch_available():
     import torch

if is_vision_available():
     from PIL import Image

     from transformers import BridgeTowerImageProcessor




class  _UpperCAmelCase (    unittest.TestCase):

 def __init__(  self	: Any						,  lowercase_	: Optional[Any]						,  lowercase_	: bool = True						,  lowercase_	: Dict[str, int] = None						,  lowercase_	: int = 32						,  lowercase_	: bool = True						,  lowercase_	: Union[int, float] = 1 / 255						,  lowercase_	: bool = True						,  lowercase_	: bool = True						,  lowercase_	: Optional[Union[float, List[float]]] = [0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]						,  lowercase_	: Optional[Union[float, List[float]]] = [0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]						,  lowercase_	: bool = True						,  lowercase_	: Optional[Any]=7						,  lowercase_	: List[str]=30						,  lowercase_	: Any=400						,  lowercase_	: Union[str, Any]=3						,  ):
     snake_case_				:  int       =			parent
     snake_case_				:  List[str]       =			do_resize
     snake_case_				:  str       =			size if size is not None else {'''shortest_edge''': 288}
     snake_case_				:  Any       =			size_divisor
     snake_case_				:  List[Any]       =			do_rescale
     snake_case_				:  Optional[int]       =			rescale_factor
     snake_case_				:  Any       =			do_normalize
     snake_case_				:  str       =			do_center_crop
     snake_case_				:  Any       =			image_mean
     snake_case_				:  Dict       =			image_std
     snake_case_				:  List[str]       =			do_pad
     snake_case_				:  Tuple       =			batch_size
     snake_case_				:  List[str]       =			num_channels
     snake_case_				:  Dict       =			min_resolution
     snake_case_				:  Union[str, Any]       =			max_resolution

 def 		_snake_case      (  self	: Dict       ):
     return {
         "image_mean": self.image_mean,
         "image_std": self.image_std,
         "do_normalize": self.do_normalize,
         "do_resize": self.do_resize,
         "size": self.size,
         "size_divisor": self.size_divisor,
     }






 def 		_snake_case      (  self	: Dict						,  lowercase_	: Dict						,  lowercase_	: List[str]=False       ):
     if not batched:
         snake_case_				:  Dict       =			self.size['''shortest_edge''']
         snake_case_				:  Optional[Any]       =			image_inputs[0]
         if isinstance(lowercase_						,  Image.Image       ):
             snake_case_,			snake_case_				:  Optional[int]       =			image.size
         else:
             snake_case_,			snake_case_				:  List[str]       =			image.shape[1], image.shape[2]
         snake_case_				:  Any       =			size / min(lowercase_						,  lowercase_       )
         if h < w:
             snake_case_,			snake_case_				:  Dict       =			size, scale * w
         else:
             snake_case_,			snake_case_				:  Union[str, Any]       =			scale * h, size

         snake_case_				:  str       =			int((1333 / 800) * size       )
         if max(lowercase_						,  lowercase_       ) > max_size:
             snake_case_				:  List[str]       =			max_size / max(lowercase_						,  lowercase_       )
             snake_case_				:  Tuple       =			newh * scale
             snake_case_				:  Tuple       =			neww * scale

         snake_case_,			snake_case_				:  str       =			int(newh + 0.5       ), int(neww + 0.5       )
         snake_case_,			snake_case_				:  List[str]       =			(
             newh // self.size_divisor * self.size_divisor,
             neww // self.size_divisor * self.size_divisor,
         )

     else:
         snake_case_				:  Union[str, Any]       =			[]
         for image in image_inputs:
             snake_case_,			snake_case_				:  Optional[Any]       =			self.get_expected_values([image]       )
             expected_values.append((expected_height, expected_width)       )
         snake_case_				:  Optional[int]       =			max(lowercase_						,  key=lambda lowercase_       : item[0]       )[0]
         snake_case_				:  Union[str, Any]       =			max(lowercase_						,  key=lambda lowercase_       : item[1]       )[1]

     return expected_height, expected_width





@require_torch
@require_vision
class  _UpperCAmelCase (    lowerCAmelCase__					,  unittest.TestCase):
 _lowerCAmelCase     :    Optional[int]	      =	BridgeTowerImageProcessor if is_vision_available() else None

 def 		_snake_case      (  self	: List[Any]       ):
     snake_case_				:  Tuple       =			BridgeTowerImageProcessingTester(self       )

 @property
 def 		_snake_case      (  self	: Union[str, Any]       ):
     return self.image_processor_tester.prepare_image_processor_dict()

 def 		_snake_case      (  self	: Optional[Any]       ):
     snake_case_				:  str       =			self.image_processing_class(**self.image_processor_dict       )
     self.assertTrue(hasattr(lowercase_						,  '''image_mean'''       )       )
     self.assertTrue(hasattr(lowercase_						,  '''image_std'''       )       )
     self.assertTrue(hasattr(lowercase_						,  '''do_normalize'''       )       )
     self.assertTrue(hasattr(lowercase_						,  '''do_resize'''       )       )
     self.assertTrue(hasattr(lowercase_						,  '''size'''       )       )
     self.assertTrue(hasattr(lowercase_						,  '''size_divisor'''       )       )

 def 		_snake_case      (  self	: Tuple       ):
     pass

 def 		_snake_case      (  self	: Optional[Any]       ):
     # Initialize image processor
     snake_case_				:  List[Any]       =			self.image_processing_class(**self.image_processor_dict       )
     # create random PIL images
     snake_case_				:  str       =			prepare_image_inputs(self.image_processor_tester						,  equal_resolution=lowercase_       )
     for image in image_inputs:
         self.assertIsInstance(lowercase_						,  Image.Image       )

     # Test not batched input
     snake_case_				:  Union[str, Any]       =			image_processing(image_inputs[0]						,  return_tensors='''pt'''       ).pixel_values

     snake_case_,			snake_case_				:  Dict       =			self.image_processor_tester.get_expected_values(lowercase_       )
     self.assertEqual(
         encoded_images.shape						,  (1, self.image_processor_tester.num_channels, expected_height, expected_width)						,  )

     # Test batched
     snake_case_				:  str       =			image_processing(lowercase_						,  return_tensors='''pt'''       ).pixel_values

     snake_case_,			snake_case_				:  int       =			self.image_processor_tester.get_expected_values(lowercase_						,  batched=lowercase_       )
     self.assertEqual(
         encoded_images.shape						,  (
             self.image_processor_tester.batch_size,
             self.image_processor_tester.num_channels,
             expected_height,
             expected_width,
         )						,  )

 def 		_snake_case      (  self	: Union[str, Any]       ):
     # Initialize image processor
     snake_case_				:  Any       =			self.image_processing_class(**self.image_processor_dict       )
     # create random numpy tensors
     snake_case_				:  Any       =			prepare_image_inputs(self.image_processor_tester						,  equal_resolution=lowercase_						,  numpify=lowercase_       )
     for image in image_inputs:
         self.assertIsInstance(lowercase_						,  np.ndarray       )

     # Test not batched input
     snake_case_				:  int       =			image_processing(image_inputs[0]						,  return_tensors='''pt'''       ).pixel_values

     snake_case_,			snake_case_				:  Union[str, Any]       =			self.image_processor_tester.get_expected_values(lowercase_       )
     self.assertEqual(
         encoded_images.shape						,  (1, self.image_processor_tester.num_channels, expected_height, expected_width)						,  )

     # Test batched
     snake_case_				:  List[Any]       =			image_processing(lowercase_						,  return_tensors='''pt'''       ).pixel_values

     snake_case_,			snake_case_				:  int       =			self.image_processor_tester.get_expected_values(lowercase_						,  batched=lowercase_       )
     self.assertEqual(
         encoded_images.shape						,  (
             self.image_processor_tester.batch_size,
             self.image_processor_tester.num_channels,
             expected_height,
             expected_width,
         )						,  )






 def 		_snake_case      (  self	: Optional[Any]       ):
     # Initialize image processor
     snake_case_				:  str       =			self.image_processing_class(**self.image_processor_dict       )
     # create random PyTorch tensors
     snake_case_				:  Union[str, Any]       =			prepare_image_inputs(self.image_processor_tester						,  equal_resolution=lowercase_						,  torchify=lowercase_       )
     for image in image_inputs:
         self.assertIsInstance(lowercase_						,  torch.Tensor       )

     # Test not batched input
     snake_case_				:  int       =			image_processing(image_inputs[0]						,  return_tensors='''pt'''       ).pixel_values

     snake_case_,			snake_case_				:  Dict       =			self.image_processor_tester.get_expected_values(lowercase_       )
     self.assertEqual(
         encoded_images.shape						,  (1, self.image_processor_tester.num_channels, expected_height, expected_width)						,  )

     # Test batched
     snake_case_				:  Dict       =			image_processing(lowercase_						,  return_tensors='''pt'''       ).pixel_values

     snake_case_,			snake_case_				:  Any       =			self.image_processor_tester.get_expected_values(lowercase_						,  batched=lowercase_       )
     self.assertEqual(
         encoded_images.shape						,  (
             self.image_processor_tester.batch_size,
             self.image_processor_tester.num_channels,
             expected_height,
             expected_width,
         )						,  )










"""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  _UpperCAmelCase (    lowerCAmelCase__):

 def __init__(  self	: List[Any]						,  *lowercase_	: List[Any]						,  lowercase_	: Optional[Any]=None						,  lowercase_	: Tuple=None						,  **lowercase_	: Union[str, Any]       ):
     super().__init__(*lowercase_						,  **lowercase_       )
     snake_case_				:  str       =			eval_examples
     snake_case_				:  Optional[Any]       =			post_process_function

 def 		_snake_case      (  self	: Union[str, Any]						,  lowercase_	: Optional[Dataset] = None						,  lowercase_	: str=None						,  lowercase_	: Optional[List[str]] = None						,  lowercase_	: str = "eval"						,  **lowercase_	: Dict						,  ):
     snake_case_				:  int       =			gen_kwargs.copy()
     snake_case_				:  Dict       =			(
         gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length'''       ) is not None else self.args.generation_max_length
     )
     snake_case_				:  Tuple       =			(
         gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''       ) is not None else self.args.generation_num_beams
     )
     snake_case_				:  Optional[Any]       =			gen_kwargs

     snake_case_				:  Optional[int]       =			self.eval_dataset if eval_dataset is None else eval_dataset
     snake_case_				:  Optional[Any]       =			self.get_eval_dataloader(lowercase_       )
     snake_case_				:  Dict       =			self.eval_examples if eval_examples is None else eval_examples

     # Temporarily disable metric computation, we will do it in the loop here.
     snake_case_				:  Optional[int]       =			self.compute_metrics
     snake_case_				:  str       =			None
     snake_case_				:  int       =			time.time()
     snake_case_				:  Any       =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
     try:
         snake_case_				:  str       =			eval_loop(
             lowercase_						,  description='''Evaluation'''						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase_						,  metric_key_prefix=lowercase_						,  )
     finally:
         snake_case_				:  Optional[int]       =			compute_metrics
     snake_case_				:  Dict       =			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
         snake_case_				:  Tuple       =			self.post_process_function(lowercase_						,  lowercase_						,  lowercase_       )
         snake_case_				:  Optional[int]       =			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}_"       ):
                 snake_case_				:  int       =			metrics.pop(lowercase_       )

         metrics.update(output.metrics       )
     else:
         snake_case_				:  Tuple       =			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()       )

     snake_case_				:  str       =			self.callback_handler.on_evaluate(self.args						,  self.state						,  self.control						,  lowercase_       )
     return metrics






 def 		_snake_case      (  self	: Any						,  lowercase_	: int						,  lowercase_	: List[str]						,  lowercase_	: str=None						,  lowercase_	: str = "test"						,  **lowercase_	: Dict       ):
     snake_case_				:  Dict       =			gen_kwargs.copy()

     snake_case_				:  Dict       =			self.get_test_dataloader(lowercase_       )

     # Temporarily disable metric computation, we will do it in the loop here.
     snake_case_				:  Tuple       =			self.compute_metrics
     snake_case_				:  int       =			None
     snake_case_				:  List[Any]       =			time.time()
     snake_case_				:  int       =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
     try:
         snake_case_				:  List[str]       =			eval_loop(
             lowercase_						,  description='''Prediction'''						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase_						,  metric_key_prefix=lowercase_						,  )
     finally:
         snake_case_				:  str       =			compute_metrics

     snake_case_				:  int       =			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

     snake_case_				:  Tuple       =			self.post_process_function(lowercase_						,  lowercase_						,  lowercase_						,  '''predict'''       )
     snake_case_				:  Dict       =			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}_"       ):
             snake_case_				:  Union[str, Any]       =			metrics.pop(lowercase_       )
     metrics.update(output.metrics       )
     return PredictionOutput(predictions=predictions.predictions						,  label_ids=predictions.label_ids						,  metrics=lowercase_       )





