




'''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      SCREAMING_SNAKE_CASE    (						snake_case_				):






      def __init__( self   :  int						,						*lowercase__   :  Union[str, Any]						,						lowercase__   :  str=None						,						lowercase__   :  Optional[int]=None						,						**lowercase__   :  List[Any]       ):


           '''simple docstring'''

           super().__init__(*lowercase__						,						**lowercase__       )
           a_     :							Any     						=  eval_examples
           a_     :							int     						=  post_process_function






      def 							lowercase_						( self   :  Optional[Any]						,						lowercase__   :  Optional[Dataset] = None						,						lowercase__   :  Optional[Any]=None						,						lowercase__   :  Optional[List[str]] = None						,						lowercase__   :  str = "eval"						,						**lowercase__   :  str						,						):


           '''simple docstring'''

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

           a_     :							Optional[int]     						=  self.eval_dataset if eval_dataset is None else eval_dataset
           a_     :							List[Any]     						=  self.get_eval_dataloader(lowercase__       )
           a_     :							Tuple     						=  self.eval_examples if eval_examples is None else eval_examples

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

                metrics.update(output.metrics       )
           else:
                a_     :							Optional[int]     						=  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()       )

           a_     :							List[Any]     						=  self.callback_handler.on_evaluate(self.args						,						self.state						,						self.control						,						lowercase__       )
           return metrics






      def 							lowercase_						( self   :  Dict						,						lowercase__   :  Any						,						lowercase__   :  List[str]						,						lowercase__   :  int=None						,						lowercase__   :  str = "test"						,						**lowercase__   :  str       ):


           '''simple docstring'''

           a_     :							Optional[Any]     						=  gen_kwargs.copy()

           a_     :							List[Any]     						=  self.get_test_dataloader(lowercase__       )

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

           a_     :							Tuple     						=  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

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






'''simple docstring'''


from unittest.mock import Mock, patch

from file_transfer.send_file import send_file



@patch("""socket.socket"""				)
@patch("""builtins.open"""				)
def  _SCREAMING_SNAKE_CASE     (      UpperCamelCase__    :					Dict ,    UpperCamelCase__    :					Tuple				):




     """simple docstring"""


     a_     :							Dict     						=  Mock()
     a_     :							List[str]     						=  conn, Mock()
     a_     :							Optional[int]     						=  iter([1, None]				)
     a_     :							List[str]     						=  lambda UpperCamelCase__				: next(UpperCamelCase__				)

     # ===== invoke =====
     send_file(filename="""mytext.txt""" ,    testing=UpperCamelCase__				)

     # ===== ensurance =====
     sock.assert_called_once()
     sock.return_value.bind.assert_called_once()
     sock.return_value.listen.assert_called_once()
     sock.return_value.accept.assert_called_once()
     conn.recv.assert_called_once()

     file.return_value.__enter__.assert_called_once()
     file.return_value.__enter__.return_value.read.assert_called()

     conn.send.assert_called_once()
     conn.close.assert_called_once()
     sock.return_value.shutdown.assert_called_once()
     sock.return_value.close.assert_called_once()
