
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 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     def __init__(				self    ,    *__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=None    ,    __SCREAMING_SNAKE_CASE=None    ,    **__SCREAMING_SNAKE_CASE	)	-> Any:




           """simple docstring"""







           super().__init__(*__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					eval_examples
           UpperCamelCase__     : Tuple											=					post_process_function






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE = None    ,    __SCREAMING_SNAKE_CASE=None    ,    __SCREAMING_SNAKE_CASE = None    ,    __SCREAMING_SNAKE_CASE = "eval"    ,    **__SCREAMING_SNAKE_CASE    ,    )	-> Dict[str, float]:




           """simple docstring"""







           UpperCamelCase__     : Optional[Any]											=					gen_kwargs.copy()
           UpperCamelCase__     : List[Any]											=					(
               gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length'''	) is not None else self.args.generation_max_length
           )
           UpperCamelCase__     : Union[str, Any]											=					(
               gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''	) is not None else self.args.generation_num_beams
           )
           UpperCamelCase__     : Union[str, Any]											=					gen_kwargs

           UpperCamelCase__     : int											=					self.eval_dataset if eval_dataset is None else eval_dataset
           UpperCamelCase__     : List[str]											=					self.get_eval_dataloader(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Union[str, Any]											=					self.eval_examples if eval_examples is None else eval_examples

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

                 # Prefix all keys with metric_key_prefix + '_'
                 for key in list(metrics.keys()	):
                       if not key.startswith(F'''{metric_key_prefix}_'''	):
                             UpperCamelCase__     : List[Any]											=					metrics.pop(__SCREAMING_SNAKE_CASE	)

                 metrics.update(output.metrics	)
           else:
                 UpperCamelCase__     : str											=					output.metrics

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

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

           UpperCamelCase__     : Any											=					self.callback_handler.on_evaluate(self.args    ,    self.state    ,    self.control    ,    __SCREAMING_SNAKE_CASE	)
           return metrics






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=None    ,    __SCREAMING_SNAKE_CASE = "test"    ,    **__SCREAMING_SNAKE_CASE	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : str											=					gen_kwargs.copy()

           UpperCamelCase__     : Dict											=					self.get_test_dataloader(__SCREAMING_SNAKE_CASE	)

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

           UpperCamelCase__     : Optional[Any]											=					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(
                   __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    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

           UpperCamelCase__     : Optional[Any]											=					self.post_process_function(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    '''predict'''	)
           UpperCamelCase__     : Optional[Any]											=					self.compute_metrics(__SCREAMING_SNAKE_CASE	)

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




import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
    from transformers import TvltImageProcessor

if is_speech_available():
    from transformers import TvltFeatureExtractor

from transformers import TvltProcessor




@require_torch
class 					_lowerCamelCase			(   unittest.TestCase			):





     """simple docstring"""







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




           """simple docstring"""







           UpperCamelCase__     : int											=					'''ZinengTang/tvlt-base'''
           UpperCamelCase__     : Optional[Any]											=					tempfile.mkdtemp()






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> Tuple:




           """simple docstring"""







           return TvltImageProcessor.from_pretrained(self.checkpoint    ,    **__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> str:




           """simple docstring"""







           return TvltFeatureExtractor.from_pretrained(self.checkpoint    ,    **__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           shutil.rmtree(self.tmpdirname	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           UpperCamelCase__     : Optional[Any]											=					self.get_image_processor()
           UpperCamelCase__     : Tuple											=					self.get_feature_extractor()

           UpperCamelCase__     : Dict											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)
           processor.save_pretrained(self.tmpdirname	)
           UpperCamelCase__     : Optional[Any]											=					TvltProcessor.from_pretrained(self.tmpdirname	)

           self.assertIsInstance(processor.feature_extractor    ,    __SCREAMING_SNAKE_CASE	)
           self.assertIsInstance(processor.image_processor    ,    __SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[str]:




           """simple docstring"""







           UpperCamelCase__     : Union[str, Any]											=					self.get_image_processor()
           UpperCamelCase__     : int											=					self.get_feature_extractor()

           UpperCamelCase__     : Union[str, Any]											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : str											=					np.ones([1_2_0_0_0]	)

           UpperCamelCase__     : Dict											=					feature_extractor(__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)
           UpperCamelCase__     : Dict											=					processor(audio=__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)

           for key in audio_dict.keys():
                 self.assertAlmostEqual(audio_dict[key].sum()    ,    input_processor[key].sum()    ,    delta=1e-2	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           UpperCamelCase__     : Tuple											=					self.get_image_processor()
           UpperCamelCase__     : Tuple											=					self.get_feature_extractor()

           UpperCamelCase__     : str											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : List[Any]											=					np.ones([3, 2_2_4, 2_2_4]	)

           UpperCamelCase__     : Optional[int]											=					image_processor(__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)
           UpperCamelCase__     : Tuple											=					processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)

           for key in image_dict.keys():
                 self.assertAlmostEqual(image_dict[key].sum()    ,    input_processor[key].sum()    ,    delta=1e-2	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           UpperCamelCase__     : List[str]											=					self.get_image_processor()
           UpperCamelCase__     : Optional[Any]											=					self.get_feature_extractor()

           UpperCamelCase__     : Union[str, Any]											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Dict											=					np.ones([1_2_0_0_0]	)
           UpperCamelCase__     : str											=					np.ones([3, 2_2_4, 2_2_4]	)

           UpperCamelCase__     : List[Any]											=					processor(audio=__SCREAMING_SNAKE_CASE    ,    images=__SCREAMING_SNAKE_CASE	)

           self.assertListEqual(list(inputs.keys()	)    ,    ['''audio_values''', '''audio_mask''', '''pixel_values''', '''pixel_mask''']	)

           # test if it raises when no input is passed
           with pytest.raises(__SCREAMING_SNAKE_CASE	):
                 processor()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : int											=					self.get_image_processor()
           UpperCamelCase__     : str											=					self.get_feature_extractor()

           UpperCamelCase__     : Dict											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           self.assertListEqual(
               processor.model_input_names    ,    image_processor.model_input_names + feature_extractor.model_input_names    ,    msg='''`processor` and `image_processor`+`feature_extractor` model input names do not match'''    ,    )


