'''simple docstring'''

import argparse
import json
import logging
import os
import shutil
import sys
import tempfile
import unittest
from unittest import mock

import torch
from accelerate.utils import write_basic_config

from transformers.testing_utils import TestCasePlus, get_gpu_count, run_command, slow, torch_device
from transformers.utils import is_apex_available


logging.basicConfig(level=logging.DEBUG)

_A			:      Tuple					      =				logging.getLogger()







def      UpperCamelCase_						(					)					->     Tuple:






 '''simple docstring'''

 __lowerCAmelCase											=						argparse.ArgumentParser()
 parser.add_argument("""-f"""      )
 __lowerCAmelCase											=						parser.parse_args()
 return args.f







def      UpperCamelCase_						(					snake_case_       :       Tuple      )					->     List[Any]:






 '''simple docstring'''

 __lowerCAmelCase											=						{}
 __lowerCAmelCase											=						os.path.join(SCREAMING_SNAKE_CASE__						,		"""all_results.json"""      )
 if os.path.exists(SCREAMING_SNAKE_CASE__      ):
  with open(SCREAMING_SNAKE_CASE__						,		"""r"""      ) as f:
   __lowerCAmelCase											=						json.load(SCREAMING_SNAKE_CASE__      )
 else:
  raise ValueError(f"""can't find {path}"""      )
 return results







def      UpperCamelCase_						(					)					->     Dict:






 '''simple docstring'''

 __lowerCAmelCase											=						torch.cuda.is_available() and torch_device == """cuda"""
 return is_using_cuda and is_apex_available()


_A			:      List[Any]					      =				logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)







class 				_lowercase     (				UpperCAmelCase__			):


 '''simple docstring'''






 @classmethod
 def a					(      cls   :	Union[str, Any]							)  ->    Optional[int]:
  __lowerCAmelCase											=						tempfile.mkdtemp()
  __lowerCAmelCase											=						os.path.join(cls.tmpdir      ,       """default_config.yml"""							)
  write_basic_config(save_location=cls.configPath							)
  __lowerCAmelCase											=						["""accelerate""", """launch""", """--config_file""", cls.configPath]
 @classmethod
 def a					(      cls   :	Dict							)  ->    Optional[int]:
  shutil.rmtree(cls.tmpdir							)
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	str							)  ->    Tuple:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/text-classification/run_glue_no_trainer.py\n            --model_name_or_path distilbert-base-uncased\n            --output_dir {tmp_dir}\n            --train_file ./tests/fixtures/tests_samples/MRPC/train.csv\n            --validation_file ./tests/fixtures/tests_samples/MRPC/dev.csv\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=1\n            --learning_rate=1e-4\n            --seed=42\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  if is_cuda_and_apex_available():
   testargs.append("""--fp16"""							)

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertGreaterEqual(result["""eval_accuracy"""]      ,       0.7_5							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """glue_no_trainer"""							)							)							)
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Union[str, Any]							)  ->    List[Any]:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/language-modeling/run_clm_no_trainer.py\n            --model_name_or_path distilgpt2\n            --train_file ./tests/fixtures/sample_text.txt\n            --validation_file ./tests/fixtures/sample_text.txt\n            --block_size 128\n            --per_device_train_batch_size 5\n            --per_device_eval_batch_size 5\n            --num_train_epochs 2\n            --output_dir {tmp_dir}\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  if torch.cuda.device_count() > 1:
   # Skipping because there are not enough batches to train the model + would need a drop_last to work.
   return

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertLess(result["""perplexity"""]      ,       1_00							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """clm_no_trainer"""							)							)							)
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Union[str, Any]							)  ->    Union[str, Any]:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/language-modeling/run_mlm_no_trainer.py\n            --model_name_or_path distilroberta-base\n            --train_file ./tests/fixtures/sample_text.txt\n            --validation_file ./tests/fixtures/sample_text.txt\n            --output_dir {tmp_dir}\n            --num_train_epochs=1\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertLess(result["""perplexity"""]      ,       42							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """mlm_no_trainer"""							)							)							)
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Optional[int]							)  ->    Optional[int]:
  __lowerCAmelCase											=						7 if get_gpu_count() > 1 else 2

  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/token-classification/run_ner_no_trainer.py\n            --model_name_or_path bert-base-uncased\n            --train_file tests/fixtures/tests_samples/conll/sample.json\n            --validation_file tests/fixtures/tests_samples/conll/sample.json\n            --output_dir {tmp_dir}\n            --learning_rate=2e-4\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=2\n            --num_train_epochs={epochs}\n            --seed 7\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertGreaterEqual(result["""eval_accuracy"""]      ,       0.7_5							)
  self.assertLess(result["""train_loss"""]      ,       0.5							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """ner_no_trainer"""							)							)							)
 @unittest.skip(reason="""Fix me @muellerzr"""							)
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Union[str, Any]							)  ->    str:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/question-answering/run_qa_no_trainer.py\n            --model_name_or_path bert-base-uncased\n            --version_2_with_negative\n            --train_file tests/fixtures/tests_samples/SQUAD/sample.json\n            --validation_file tests/fixtures/tests_samples/SQUAD/sample.json\n            --output_dir {tmp_dir}\n            --seed=42\n            --max_train_steps=10\n            --num_warmup_steps=2\n            --learning_rate=2e-4\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=1\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  # Because we use --version_2_with_negative the testing script uses SQuAD v2 metrics.
  self.assertGreaterEqual(result["""eval_f1"""]      ,       28							)
  self.assertGreaterEqual(result["""eval_exact"""]      ,       28							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """qa_no_trainer"""							)							)							)
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Optional[int]							)  ->    Union[str, Any]:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/multiple-choice/run_swag_no_trainer.py\n            --model_name_or_path bert-base-uncased\n            --train_file tests/fixtures/tests_samples/swag/sample.json\n            --validation_file tests/fixtures/tests_samples/swag/sample.json\n            --output_dir {tmp_dir}\n            --max_train_steps=20\n            --num_warmup_steps=2\n            --learning_rate=2e-4\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=1\n            --with_tracking\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertGreaterEqual(result["""eval_accuracy"""]      ,       0.8							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """swag_no_trainer"""							)							)							)
 @slow
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Union[str, Any]							)  ->    Optional[int]:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/summarization/run_summarization_no_trainer.py\n            --model_name_or_path t5-small\n            --train_file tests/fixtures/tests_samples/xsum/sample.json\n            --validation_file tests/fixtures/tests_samples/xsum/sample.json\n            --output_dir {tmp_dir}\n            --max_train_steps=50\n            --num_warmup_steps=8\n            --learning_rate=2e-4\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=1\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertGreaterEqual(result["""eval_rouge1"""]      ,       10							)
  self.assertGreaterEqual(result["""eval_rouge2"""]      ,       2							)
  self.assertGreaterEqual(result["""eval_rougeL"""]      ,       7							)
  self.assertGreaterEqual(result["""eval_rougeLsum"""]      ,       7							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """summarization_no_trainer"""							)							)							)
 @slow
 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	List[str]							)  ->    Union[str, Any]:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/translation/run_translation_no_trainer.py\n            --model_name_or_path sshleifer/student_marian_en_ro_6_1\n            --source_lang en\n            --target_lang ro\n            --train_file tests/fixtures/tests_samples/wmt16/sample.json\n            --validation_file tests/fixtures/tests_samples/wmt16/sample.json\n            --output_dir {tmp_dir}\n            --max_train_steps=50\n            --num_warmup_steps=8\n            --num_beams=6\n            --learning_rate=3e-3\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=1\n            --source_lang en_XX\n            --target_lang ro_RO\n            --checkpointing_steps epoch\n            --with_tracking\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertGreaterEqual(result["""eval_bleu"""]      ,       30							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """epoch_0"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """translation_no_trainer"""							)							)							)
 @slow
 def a					(      self   :	Union[str, Any]							)  ->    Dict:
  __lowerCAmelCase											=						logging.StreamHandler(sys.stdout							)
  logger.addHandler(SCREAMING_SNAKE_CASE__							)

  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/semantic-segmentation/run_semantic_segmentation_no_trainer.py\n            --dataset_name huggingface/semantic-segmentation-test-sample\n            --output_dir {tmp_dir}\n            --max_train_steps=10\n            --num_warmup_steps=2\n            --learning_rate=2e-4\n            --per_device_train_batch_size=2\n            --per_device_eval_batch_size=1\n            --checkpointing_steps epoch\n        """.split()

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  self.assertGreaterEqual(result["""eval_overall_accuracy"""]      ,       0.1_0							)





 @mock.patch.dict(os.environ      ,       {"""WANDB_MODE""": """offline"""}							)
 def a					(      self   :	Any							)  ->    Tuple:
  __lowerCAmelCase											=						self.get_auto_remove_tmp_dir()
  __lowerCAmelCase											=						f"""\n            {self.examples_dir}/pytorch/image-classification/run_image_classification_no_trainer.py\n            --model_name_or_path google/vit-base-patch16-224-in21k\n            --dataset_name hf-internal-testing/cats_vs_dogs_sample\n            --learning_rate 1e-4\n            --per_device_train_batch_size 2\n            --per_device_eval_batch_size 1\n            --max_train_steps 2\n            --train_val_split 0.1\n            --seed 42\n            --output_dir {tmp_dir}\n            --with_tracking\n            --checkpointing_steps 1\n        """.split()

  if is_cuda_and_apex_available():
   testargs.append("""--fp16"""							)

  run_command(self._launch_args + testargs							)
  __lowerCAmelCase											=						get_results(SCREAMING_SNAKE_CASE__							)
  # The base model scores a 25%
  self.assertGreaterEqual(result["""eval_accuracy"""]      ,       0.6							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """step_1"""							)							)							)
  self.assertTrue(os.path.exists(os.path.join(SCREAMING_SNAKE_CASE__      ,       """image_classification_no_trainer"""							)							)							)








import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY

@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class __snake_case		(      unittest.TestCase     ):
    _a		:						Optional[int]=				MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case      ):



           '''simple docstring'''







           lowercase							:     Any       	=      hf_hub_download(
               repo_id="""nateraw/video-demo"""		,filename="""archery.mp4"""		,repo_type="""dataset"""      )
           lowercase							:     List[str]       	=      VideoClassificationPipeline(model=snake_case		,image_processor=snake_case		,top_k=2      )
           lowercase							:     Dict       	=      [
               example_video_filepath,
               """https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4""",
           ]
           return video_classifier, examples
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case      ):



           '''simple docstring'''







           for example in examples:
                  lowercase							:     int       	=      video_classifier(snake_case      )

                  self.assertEqual(
                      snake_case		,[
                          {"""score""": ANY(snake_case      ), """label""": ANY(snake_case      )},
                          {"""score""": ANY(snake_case      ), """label""": ANY(snake_case      )},
                      ]		,)
    @require_torch
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Any       	=      """hf-internal-testing/tiny-random-VideoMAEForVideoClassification"""
           lowercase							:     str       	=      VideoMAEFeatureExtractor(
               size={"""shortest_edge""": 10}		,crop_size={"""height""": 10, """width""": 10}      )
           lowercase							:     List[Any]       	=      pipeline(
               """video-classification"""		,model=snake_case		,feature_extractor=snake_case		,frame_sampling_rate=4      )

           lowercase							:     Dict       	=      hf_hub_download(repo_id="""nateraw/video-demo"""		,filename="""archery.mp4"""		,repo_type="""dataset"""      )
           lowercase							:     Any       	=      video_classifier(snake_case		,top_k=2      )
           self.assertEqual(
               nested_simplify(snake_case		,decimals=4      )		,[{"""score""": 0.5_199, """label""": """LABEL_0"""}, {"""score""": 0.4_801, """label""": """LABEL_1"""}]		,)

           lowercase							:     str       	=      video_classifier(
               [
                   video_file_path,
                   video_file_path,
               ]		,top_k=2		,)
           self.assertEqual(
               nested_simplify(snake_case		,decimals=4      )		,[
                   [{"""score""": 0.5_199, """label""": """LABEL_0"""}, {"""score""": 0.4_801, """label""": """LABEL_1"""}],
                   [{"""score""": 0.5_199, """label""": """LABEL_0"""}, {"""score""": 0.4_801, """label""": """LABEL_1"""}],
               ]		,)
    @require_tf
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           pass
