


"""simple docstring"""







import math
import os
import re
import sys
import unittest
from pathlib import Path
from typing import Tuple
from unittest.mock import patch

from parameterized import parameterized

from transformers.testing_utils import (
    CaptureStderr,
    ExtendSysPath,
    TestCasePlus,
    execute_subprocess_async,
    get_gpu_count,
    get_torch_dist_unique_port,
    require_apex,
    require_bitsandbytes,
    require_fairscale,
    require_torch,
    require_torch_gpu,
    require_torch_multi_gpu,
    require_torch_non_multi_gpu,
    slow,
)
from transformers.trainer_callback import TrainerState
from transformers.trainer_utils import set_seed


__lowercase    					=      os.path.abspath(os.path.dirname(__file__))
with ExtendSysPath(f'''{bindir}/../../examples/pytorch/translation'''):
 from run_translation import main  # noqa


set_seed(42)
__lowercase    					=      """sshleifer/student_marian_en_ro_6_1"""
__lowercase    					=      """sshleifer/tiny-mbart"""






@require_torch
class 						_A  (  _a  ):







    """simple docstring"""




    def 	__snake_case							(					self				:    Optional[Any]				, __UpperCAmelCase				:    Dict=False				, __UpperCAmelCase				:    str=None				, __UpperCAmelCase				:    Union[str, Any]=True				, __UpperCAmelCase				:    Optional[Any]=True				, __UpperCAmelCase				:    str=True				, __UpperCAmelCase				:    Optional[Any]=True				, ):
      a						:		str      							=			self.run_trainer(
          eval_steps=1				, max_len=12				, model_name=__UpperCAmelCase				, num_train_epochs=1				, distributed=__UpperCAmelCase				, extra_args_str=__UpperCAmelCase				, predict_with_generate=__UpperCAmelCase				, do_train=__UpperCAmelCase				, do_eval=__UpperCAmelCase				, do_predict=__UpperCAmelCase				, )
      a						:		Dict      							=			TrainerState.load_from_json(os.path.join(__UpperCAmelCase				, "trainer_state.json")).log_history

      if not do_eval:
        return

      a						:		Tuple      							=			[log for log in logs if "eval_loss" in log.keys()]

      a						:		Optional[int]      							=			eval_metrics[0]
      if predict_with_generate:
        assert "eval_bleu" in first_step_stats

        a						:		Optional[int]      							=			eval_metrics[-1]
        assert isinstance(last_step_stats["eval_bleu"]				, __UpperCAmelCase)
        assert not math.isnan(float(last_step_stats["eval_loss"])), "eval_loss must not be `nan`"






    @require_torch_non_multi_gpu
    def 	__snake_case							(					self				:    Optional[Any]):
      self.run_seqaseq_quick()






    @require_torch_multi_gpu
    def 	__snake_case							(					self				:    Dict):
      self.run_seqaseq_quick(distributed=__UpperCAmelCase)






    @require_torch_multi_gpu
    def 	__snake_case							(					self				:    Tuple):
      self.run_seqaseq_quick(distributed=__UpperCAmelCase)






    @unittest.skip("Requires an update of the env running those tests")
    @require_torch_multi_gpu
    @require_fairscale
    def 	__snake_case							(					self				:    Tuple):
      self.run_seqaseq_quick(distributed=__UpperCAmelCase				, extra_args_str="--sharded_ddp simple")






    @unittest.skip("Requires an update of the env running those tests")
    @require_torch_multi_gpu
    @require_fairscale
    def 	__snake_case							(					self				:    Optional[Any]):
      self.run_seqaseq_quick(distributed=__UpperCAmelCase				, extra_args_str="--sharded_ddp simple --fp16")






    @unittest.skip("Requires an update of the env running those tests")
    @require_torch_multi_gpu
    @require_fairscale
    def 	__snake_case							(					self				:    Optional[int]):
      self.run_seqaseq_quick(distributed=__UpperCAmelCase				, extra_args_str="--sharded_ddp zero_dp_2"				, predict_with_generate=__UpperCAmelCase)






    @unittest.skip("Requires an update of the env running those tests")
    @require_torch_multi_gpu
    @require_fairscale
    def 	__snake_case							(					self				:    int):
      self.run_seqaseq_quick(
          distributed=__UpperCAmelCase				, extra_args_str="--sharded_ddp zero_dp_2 --fp16"				, predict_with_generate=__UpperCAmelCase)






    @require_apex
    @require_torch_gpu
    def 	__snake_case							(					self				:    Any):
      self.run_seqaseq_quick(distributed=__UpperCAmelCase				, extra_args_str="--fp16 --fp16_backend=apex")
      # test 2nd time - was getting eval_loss': nan'
      # to reproduce the problem set distributed=False
      self.run_seqaseq_quick(distributed=__UpperCAmelCase				, extra_args_str="--fp16 --fp16_backend=apex")






    @parameterized.expand(["base", "low", "high", "mixed"])
    @require_torch_multi_gpu
    def 	__snake_case							(					self				:    List[str]				, __UpperCAmelCase				:    List[str]):
      a						:		Any      							=			{
          # test with the default log_level - should be info and thus log info once
          "base": {"extra_args_str": "", "n_matches": 1},
          # test with low log_level and log_level_replica - should be noisy on all processes
          # now the info string should appear twice on 2 processes
          "low": {"extra_args_str": "--log_level debug --log_level_replica debug", "n_matches": 2},
          # test with high log_level and low log_level_replica
          # now the info string should appear once only on the replica
          "high": {"extra_args_str": "--log_level error --log_level_replica debug", "n_matches": 1},
          # test with high log_level and log_level_replica - should be quiet on all processes
          "mixed": {"extra_args_str": "--log_level error --log_level_replica error", "n_matches": 0},
      }

      a						:		str      							=			experiments[experiment_id]
      a						:		Dict      							=			{"distributed": True, "predict_with_generate": False, "do_eval": False, "do_predict": False}
      a						:		List[str]      							=			"Running training"
      with CaptureStderr() as cl:
        self.run_seqaseq_quick(**__UpperCAmelCase				, extra_args_str=data["extra_args_str"])
      a						:		Tuple      							=			len(re.findall(__UpperCAmelCase				, cl.err))
      self.assertEqual(__UpperCAmelCase				, data["n_matches"])






    @slow
    def 	__snake_case							(					self				:    Tuple):
      a						:		Dict      							=			self.run_trainer(
          eval_steps=2				, max_len=128				, model_name=__UpperCAmelCase				, learning_rate=3e-4				, num_train_epochs=10				, distributed=__UpperCAmelCase				, )

      # Check metrics
      a						:		str      							=			TrainerState.load_from_json(os.path.join(__UpperCAmelCase				, "trainer_state.json")).log_history
      a						:		Optional[Any]      							=			[log for log in logs if "eval_loss" in log.keys()]
      a						:		Dict      							=			eval_metrics[0]
      a						:		Tuple      							=			eval_metrics[-1]

      assert first_step_stats["eval_loss"] > last_step_stats["eval_loss"], "model learned nothing"
      assert isinstance(last_step_stats["eval_bleu"]				, __UpperCAmelCase)

      # test if do_predict saves generations and metrics
      a						:		int      							=			os.listdir(__UpperCAmelCase)
      a						:		Dict      							=			{os.path.basename(__UpperCAmelCase) for p in contents}
      assert "generated_predictions.txt" in contents
      assert "predict_results.json" in contents






    @slow
    @require_bitsandbytes
    def 	__snake_case							(					self				:    List[Any]):
      from transformers.training_args import OptimizerNames

      def train_and_return_metrics(__UpperCAmelCase				:    List[Any]) -> Tuple[int, float]:
        a						:		List[Any]      							=			"--skip_memory_metrics 0"

        a						:		Optional[int]      							=			self.run_trainer(
            max_len=128				, model_name=__UpperCAmelCase				, learning_rate=3e-4				, num_train_epochs=1				, optim=__UpperCAmelCase				, distributed=__UpperCAmelCase				, extra_args_str=__UpperCAmelCase				, do_eval=__UpperCAmelCase				, do_predict=__UpperCAmelCase				, n_gpus_to_use=1				, )

        # Check metrics
        a						:		List[Any]      							=			TrainerState.load_from_json(Path(__UpperCAmelCase				, "trainer_state.json")).log_history
        a						:		Optional[int]      							=			int(logs[0]["train_mem_gpu_peaked_delta"] / 2**20)
        a						:		List[str]      							=			int(logs[0]["train_mem_gpu_alloc_delta"] / 2**20)

        a						:		Any      							=			logs[0]["train_loss"]
        return gpu_peak_mem_mb, gpu_alloc_mem_mb, loss

      a							,		a							,		a						:		Optional[Any]      							=			train_and_return_metrics(OptimizerNames.ADAMW_TORCH.value)
      a							,		a							,		a						:		Optional[Any]      							=			train_and_return_metrics(OptimizerNames.ADAMW_BNB.value)

      a						:		List[str]      							=			gpu_alloc_mem_orig - gpu_alloc_mem_bnb

      a						:		List[str]      							=			gpu_peak_mem_orig + gpu_alloc_mem_orig
      a						:		Dict      							=			gpu_peak_mem_bnb + gpu_alloc_mem_bnb
      a						:		int      							=			gpu_total_mem_orig - gpu_total_mem_bnb

      # sshleifer/student_marian_en_ro_6_1 has 54M parameter, 29M of which is `nn.Embedding` which
      # doesn't get quantized and remains in fp32. Therefore we only have 25M parameters quantized
      # in 2 bytes and the diff in optim memory usage is derived as so:
      #
      # - normal 25*8=~200MB (8 bytes per param)
      # - bnb    25*2= ~50MB (2 bytes per param)
      #
      # Thus we should expect ~150MB total memory saved.
      #
      # Peak memory should be the same - the total should be different by about that same margin
      #
      # After leaving a small margin to accommodate for differences between gpus let's check
      # that we have at least 120MB in savings
      a						:		int      							=			120

      # uncomment the following if this test starts failing - requires py38 for a new print feature
      # gpu_peak_mem_diff = gpu_peak_mem_orig - gpu_peak_mem_bnb
      # print(f"{gpu_alloc_mem_orig=}MB {gpu_peak_mem_orig=}MB {gpu_alloc_mem_orig+gpu_peak_mem_orig=}MB")
      # print(f" {gpu_alloc_mem_bnb=}MB  {gpu_peak_mem_bnb=}MB  {gpu_alloc_mem_bnb+gpu_peak_mem_bnb=}MB")
      # print(f"{gpu_alloc_mem_diff=}MB")
      # print(f"{gpu_peak_mem_diff=}MB")
      # print(f"{gpu_total_mem_orig=}MB, {gpu_total_mem_bnb=}MB")
      # print(f"{gpu_total_mem_diff=}MB, {gpu_total_mem_diff=}MB")

      self.assertGreater(
          __UpperCAmelCase				, __UpperCAmelCase				, "should use ~150MB less alloc gpu memory with BNB, compared to without it for this model but got"
          f''' a difference of {gpu_alloc_mem_diff}MB, with gpu_alloc_mem_orig={gpu_alloc_mem_orig}MB and'''
          f''' gpu_alloc_mem_bnb={gpu_alloc_mem_bnb}MB'''				, )

      self.assertGreater(
          __UpperCAmelCase				, __UpperCAmelCase				, "should use ~150MB less total gpu memory with BNB, compared to without it for this model but got"
          f''' a difference of {gpu_total_mem_diff}MB, with gpu_total_mem_orig={gpu_total_mem_orig}MB and'''
          f''' gpu_total_mem_bnb={gpu_total_mem_bnb}MB'''				, )

      self.assertEqual(
          __UpperCAmelCase				, __UpperCAmelCase				, f'''loss should be the same, but got loss_orig={loss_orig}, loss_bnb={loss_bnb}''')






    def 	__snake_case							(					self				:    str				, __UpperCAmelCase				:    int				, __UpperCAmelCase				:    Optional[int]				, __UpperCAmelCase				:    List[Any]				, __UpperCAmelCase				:    Union[str, Any] = 3e-3				, __UpperCAmelCase				:    str = "adafactor"				, __UpperCAmelCase				:    int = False				, __UpperCAmelCase				:    Union[str, Any] = None				, __UpperCAmelCase				:    Optional[Any] = 0				, __UpperCAmelCase				:    List[str] = True				, __UpperCAmelCase				:    str = True				, __UpperCAmelCase				:    str = True				, __UpperCAmelCase				:    Union[str, Any] = True				, __UpperCAmelCase				:    List[Any] = None				, ):
      a						:		str      							=			self.test_file_dir / "../fixtures/tests_samples/wmt_en_ro"
      a						:		List[str]      							=			self.get_auto_remove_tmp_dir()
      a						:		List[str]      							=			f'''
            --model_name_or_path {model_name}
            --train_file {data_dir}/train.json
            --validation_file {data_dir}/val.json
            --test_file {data_dir}/test.json
            --output_dir {output_dir}
            --overwrite_output_dir
            --max_train_samples 8
            --max_source_length {max_len}
            --max_target_length {max_len}
            --do_train
            --num_train_epochs {str(__UpperCAmelCase)}
            --per_device_train_batch_size 4
            --learning_rate {learning_rate}
            --warmup_steps 8
            --logging_steps 0
            --logging_strategy no
            --save_steps {str(__UpperCAmelCase)}
            --group_by_length
            --label_smoothing_factor 0.1
            --target_lang ro_RO
            --source_lang en_XX
        '''.split()

      a						:		Tuple      							=			f'''
            --do_eval
            --per_device_eval_batch_size 4
            --max_eval_samples 8
            --val_max_target_length {max_len}
            --evaluation_strategy steps
            --eval_steps {str(__UpperCAmelCase)}
        '''.split()

      a						:		List[str]      							=			"\n            --do_predict\n        ".split()

      a						:		List[Any]      							=			[]
      if do_train:
        args += args_train

      if do_eval:
        args += args_eval

      if do_predict:
        args += args_predict

      if predict_with_generate:
        args += "--predict_with_generate".split()

      if do_train:
        if optim == "adafactor":
          args += "--adafactor".split()
        else:
          args += f'''--optim {optim}'''.split()

      if extra_args_str is not None:
        args += extra_args_str.split()

      if distributed:
        if n_gpus_to_use is None:
          a						:		List[str]      							=			get_gpu_count()
        a						:		str      							=			get_torch_dist_unique_port()
        a						:		List[str]      							=			f'''
                -m torch.distributed.run
                --nproc_per_node={n_gpus_to_use}
                --master_port={master_port}
                {self.examples_dir_str}/pytorch/translation/run_translation.py
            '''.split()
        a						:		Tuple      							=			[sys.executable] + distributed_args + args
        # keep for quick debug
        # print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] +cmd)); die
        execute_subprocess_async(__UpperCAmelCase				, env=self.get_env())
      else:
        a						:		Optional[int]      							=			["run_translation.py"] + args
        with patch.object(__UpperCAmelCase				, "argv"				, __UpperCAmelCase):
          main()

      return output_dir










'''simple docstring'''

import math





def __snake_case					(      UpperCAmelCase_						:						float					,				UpperCAmelCase_						:						float		):
 return math.pow(UpperCAmelCase_					,				2		) - a





def __snake_case					(      UpperCAmelCase_						:						float		):
 return 2 * x





def __snake_case					(      UpperCAmelCase_						:						float		):
 lowerCamelCase_      =  2.0

 while start <= a:
  lowerCamelCase_      =  math.pow(UpperCAmelCase_					,				2		)

 return start





def __snake_case					(      UpperCAmelCase_						:						float					,				UpperCAmelCase_						:						int = 9999					,				UpperCAmelCase_						:						float = 0.00_0000_0000_0001		):

 if a < 0:
  raise ValueError("math domain error"		)

 lowerCamelCase_      =  get_initial_point(UpperCAmelCase_		)

 for _ in range(UpperCAmelCase_		):
  lowerCamelCase_      =  value
  lowerCamelCase_      =  value - fx(UpperCAmelCase_					,				UpperCAmelCase_		) / fx_derivative(UpperCAmelCase_		)
  if abs(prev_value - value		) < tolerance:
   return value

 return value


if __name__ == "__main__":
     from doctest import testmod

     testmod()



