
import ast
import os
import re
import shutil
import tempfile
import unittest
from unittest import mock

import torch

from accelerate.test_utils.examples import compare_against_test
from accelerate.test_utils.testing import TempDirTestCase, require_trackers, run_command, slow
from accelerate.utils import write_basic_config


# DataLoaders built from `test_samples/MRPC` for quick testing
# Should mock `{script_name}.get_dataloaders` via:
# @mock.patch("{script_name}.get_dataloaders", mocked_dataloaders)

lowercase  : Optional[Any]						   =							[
    """cross_validation.py""",
    """gradient_accumulation.py""",
    """local_sgd.py""",
    """multi_process_metrics.py""",
    """memory.py""",
    """automatic_gradient_accumulation.py""",
    """fsdp_with_peak_mem_tracking.py""",
    """deepspeed_with_config_support.py""",
    """megatron_lm_gpt_pretraining.py""",
]

class __snake_case		(				unittest.TestCase					):
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case		,snake_case = None		,snake_case = None				):



           '''simple docstring'''







           lowercase							:					Any	     =				None
           lowercase							:					List[Any]	     =				os.path.abspath(os.path.join("""examples"""		,"""by_feature"""				)				)
           lowercase							:					Union[str, Any]	     =				os.path.abspath("""examples"""				)
           for item in os.listdir(snake_case				):
                  if item not in EXCLUDE_EXAMPLES:
                         lowercase							:					Dict	     =				os.path.join(snake_case		,snake_case				)
                         if os.path.isfile(snake_case				) and ".py" in item_path:
                                with self.subTest(
                                    tested_script=snake_case		,feature_script=snake_case		,tested_section="""main()""" if parser_only else """training_function()"""		,):
                                       lowercase							:					Optional[int]	     =				compare_against_test(
                                           os.path.join(snake_case		,snake_case				)		,snake_case		,snake_case		,snake_case				)
                                       lowercase							:					str	     =				"""\n""".join(snake_case				)
                                       if special_strings is not None:
                                              for string in special_strings:
                                                     lowercase							:					List[Any]	     =				diff.replace(snake_case		,""""""				)
                                       self.assertEqual(snake_case		,""""""				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           self.one_complete_example("""complete_nlp_example.py"""		,snake_case				)
           self.one_complete_example("""complete_nlp_example.py"""		,snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Dict	     =				os.path.abspath(os.path.join("""examples"""		,"""cv_example.py"""				)				)
           lowercase							:					List[Any]	     =				[
               """ """ * 16 + """{\n\n""",
               """ """ * 20 + """\"accuracy\": eval_metric[\"accuracy\"],\n\n""",
               """ """ * 20 + """\"f1\": eval_metric[\"f1\"],\n\n""",
               """ """ * 20 + """\"train_loss\": total_loss.item() / len(train_dataloader),\n\n""",
               """ """ * 20 + """\"epoch\": epoch,\n\n""",
               """ """ * 16 + """},\n\n""",
               """ """ * 16 + """step=epoch,\n""",
               """ """ * 12,
               """ """ * 8 + """for step, batch in enumerate(active_dataloader):\n""",
           ]
           self.one_complete_example("""complete_cv_example.py"""		,snake_case		,snake_case		,snake_case				)
           self.one_complete_example("""complete_cv_example.py"""		,snake_case		,snake_case		,snake_case				)

@mock.patch.dict(os.environ ,     {"TESTING_MOCKED_DATALOADERS": "1"}					)
class __snake_case		(				lowerCAmelCase					):
    _a		:						Union[str, Any]=    False
    @classmethod
    def 				_SCREAMING_SNAKE_CASE	(  cls				):



           '''simple docstring'''







           super().setUpClass()
           lowercase							:					Tuple	     =				tempfile.mkdtemp()
           lowercase							:					List[Any]	     =				os.path.join(cls._tmpdir		,"""default_config.yml"""				)

           write_basic_config(save_location=cls.configPath				)
           lowercase							:					List[Any]	     =				["""accelerate""", """launch""", """--config_file""", cls.configPath]
    @classmethod
    def 				_SCREAMING_SNAKE_CASE	(  cls				):



           '''simple docstring'''







           super().tearDownClass()
           shutil.rmtree(cls._tmpdir				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Optional[Any]	     =				f"\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps epoch\n        --output_dir {self.tmpdir}\n        ".split()
           run_command(self._launch_args + testargs				)
           self.assertTrue(os.path.exists(os.path.join(self.tmpdir		,"""epoch_0"""				)				)				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					int	     =				f"\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps 1\n        --output_dir {self.tmpdir}\n        ".split()
           lowercase							:					Dict	     =				run_command(self._launch_args + testargs				)
           self.assertTrue(os.path.exists(os.path.join(self.tmpdir		,"""step_2"""				)				)				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Dict	     =				f"\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir		,'epoch_0'				)}\n        ".split()
           lowercase							:					List[Any]	     =				run_command(self._launch_args + testargs		,return_stdout=snake_case				)
           self.assertNotIn("""epoch 0:"""		,snake_case				)
           self.assertIn("""epoch 1:"""		,snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Optional[int]	     =				f"\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir		,'step_2'				)}\n        ".split()
           lowercase							:					int	     =				run_command(self._launch_args + testargs		,return_stdout=snake_case				)
           if torch.cuda.is_available():
                  lowercase							:					Union[str, Any]	     =				torch.cuda.device_count()
           else:
                  lowercase							:					Dict	     =				1
           if num_processes > 1:
                  self.assertNotIn("""epoch 0:"""		,snake_case				)
                  self.assertIn("""epoch 1:"""		,snake_case				)
           else:
                  self.assertIn("""epoch 0:"""		,snake_case				)
                  self.assertIn("""epoch 1:"""		,snake_case				)
    @slow
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Union[str, Any]	     =				"""
        examples/by_feature/cross_validation.py
        --num_folds 2
        """.split()
           with mock.patch.dict(os.environ		,{"""TESTING_MOCKED_DATALOADERS""": """0"""}				):
                  lowercase							:					str	     =				run_command(self._launch_args + testargs		,return_stdout=snake_case				)
                  lowercase							:					Union[str, Any]	     =				re.findall("""({.+})"""		,snake_case				)
                  lowercase							:					List[Any]	     =				[r for r in results if """accuracy""" in r][-1]
                  lowercase							:					Tuple	     =				ast.literal_eval(snake_case				)
                  self.assertGreaterEqual(results["""accuracy"""]		,0.75				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					str	     =				["""examples/by_feature/multi_process_metrics.py"""]
           run_command(self._launch_args + testargs				)
    @require_trackers
    @mock.patch.dict(os.environ		,{"""WANDB_MODE""": """offline"""}				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           with tempfile.TemporaryDirectory() as tmpdir:
                  lowercase							:					Optional[Any]	     =				f"\n            examples/by_feature/tracking.py\n            --with_tracking\n            --project_dir {tmpdir}\n            ".split()
                  run_command(self._launch_args + testargs				)
                  self.assertTrue(os.path.exists(os.path.join(snake_case		,"""tracking"""				)				)				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Tuple	     =				["""examples/by_feature/gradient_accumulation.py"""]
           run_command(self._launch_args + testargs				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           lowercase							:					Optional[Any]	     =				["""examples/by_feature/local_sgd.py"""]
           run_command(self._launch_args + testargs				)


from __future__ import annotations

from itertools import permutations
from random import randint
from timeit import repeat







def 						_snake_case(  )			->   tuple[list[int], int]:
       lowercase							:					List[Any]	     =				[randint(-1_000 ,     1_000    ) for i in range(10    )]
       lowercase							:					Tuple	     =				randint(-5_000 ,     5_000    )
       return (arr, r)


lowercase  : List[Any]						   =							make_dataset()







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )			->   tuple[int, ...]:
       for triplet in permutations(SCREAMING_SNAKE_CASE__ ,     3    ):
              if sum(SCREAMING_SNAKE_CASE__    ) == target:
                     return tuple(sorted(SCREAMING_SNAKE_CASE__    )    )
       return (0, 0, 0)







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )			->   tuple[int, int, int]:
       arr.sort()
       lowercase							:					Optional[int]	     =				len(SCREAMING_SNAKE_CASE__    )
       for i in range(n - 1    ):
              lowercase     ,      lowercase							:					Optional[Any]	     =				i + 1, n - 1
              while left < right:
                     if arr[i] + arr[left] + arr[right] == target:
                            return (arr[i], arr[left], arr[right])
                     elif arr[i] + arr[left] + arr[right] < target:
                            left += 1
                     elif arr[i] + arr[left] + arr[right] > target:
                            right -= 1
       return (0, 0, 0)







def 						_snake_case(  )			->   tuple[float, float]:
       lowercase							:					Dict	     =				"""
from __main__ import dataset, triplet_sum1, triplet_sum2
"""
       lowercase							:					Tuple	     =				"""
triplet_sum1(*dataset)
"""
       lowercase							:					int	     =				"""
triplet_sum2(*dataset)
"""
       lowercase							:					str	     =				repeat(setup=SCREAMING_SNAKE_CASE__ ,     stmt=SCREAMING_SNAKE_CASE__ ,     repeat=5 ,     number=10_000    )
       lowercase							:					Dict	     =				repeat(setup=SCREAMING_SNAKE_CASE__ ,     stmt=SCREAMING_SNAKE_CASE__ ,     repeat=5 ,     number=10_000    )
       return (min(SCREAMING_SNAKE_CASE__    ), min(SCREAMING_SNAKE_CASE__    ))


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

  testmod()
  lowercase  : Union[str, Any]						   =							solution_times()
  print(F'''The time for naive implementation is {times[0]}.''')
  print(F'''The time for optimized implementation is {times[1]}.''')
