

'''simple docstring'''





import re
import time
from typing import Optional

import IPython.display as disp

from ..trainer_callback import TrainerCallback
from ..trainer_utils import IntervalStrategy, has_length

def      A (       A_							:					str       ):
						snake_case       :     List[str] =    int(A_       )
						snake_case       :     Any =    t // 3600, (t // 60) % 60, t % 60
						return F"""{h}:{m:02d}:{s:02d}""" if h != 0 else F"""{m:02d}:{s:02d}"""

def      A (       A_							:					Optional[int]    ,							A_							:					int    ,							A_							:					List[str]    ,							A_							:					Optional[int]    ,							A_							:					Optional[int]=300       ):
						# docstyle-ignore
						return F"""
    <div>
      {prefix}
      <progress value='{value}' max='{total}' style='width:{width}px; height:20px; vertical-align: middle;'></progress>
      {label}
    </div>
    """

def      A (       A_							:					Dict       ):
						snake_case       :     Tuple =    '''<table border="1" class="dataframe">\n'''
						html_code += """  <thead>\n <tr style="text-align: left;">\n"""
						for i in items[0]:
												html_code += F"""      <th>{i}</th>\n"""
						html_code += "    </tr>\n  </thead>\n  <tbody>\n"
						for line in items[1:]:
												html_code += "    <tr>\n"
												for elt in line:
																		snake_case       :     List[str] =    F"""{elt:.6f}""" if isinstance(A_    ,							A_       ) else str(A_       )
																		html_code += F"""      <td>{elt}</td>\n"""
												html_code += "    </tr>\n"
						html_code += "  </tbody>\n</table><p>"
						return html_code

class 						a   :
				_snake_case		    = 5
				_snake_case		    = 0.2




				def __init__(							self	:				Union[str, Any],    SCREAMING_SNAKE_CASE_	:				int,    SCREAMING_SNAKE_CASE_	:				Optional[str] = None,    SCREAMING_SNAKE_CASE_	:				bool = True,    SCREAMING_SNAKE_CASE_	:				Optional["NotebookTrainingTracker"] = None,    SCREAMING_SNAKE_CASE_	:				int = 3_00,    ):
										snake_case       :     List[Any] =    total
										snake_case       :     Union[str, Any] =    '''''' if prefix is None else prefix
										snake_case       :     List[Any] =    leave
										snake_case       :     int =    parent
										snake_case       :     List[str] =    width
										snake_case       :     Optional[int] =    None
										snake_case       :     Optional[Any] =    None
										snake_case       :     Tuple =    None




				def 	__snake_case							(							self	:				Any,    SCREAMING_SNAKE_CASE_	:				int,    SCREAMING_SNAKE_CASE_	:				bool = False,    SCREAMING_SNAKE_CASE_	:				str = None      ):
										snake_case       :     List[Any] =    value
										if comment is not None:
																snake_case       :     Tuple =    comment
										if self.last_value is None:
																snake_case       :     str =    time.time()
																snake_case       :     List[str] =    value
																snake_case       :     str =    None
																snake_case       :     Dict =    self.warmup
																snake_case       :     Tuple =    1
																self.update_bar(SCREAMING_SNAKE_CASE_      )
										elif value <= self.last_value and not force_update:
																return
										elif force_update or self.first_calls > 0 or value >= min(self.last_value + self.wait_for,    self.total      ):
																if self.first_calls > 0:
																						self.first_calls -= 1
																snake_case       :     Tuple =    time.time()
																snake_case       :     str =    current_time - self.start_time
																# We could have value = self.start_value if the update is called twixe with the same start value.
																if value > self.start_value:
																						snake_case       :     Any =    self.elapsed_time / (value - self.start_value)
																else:
																						snake_case       :     List[str] =    None
																if value >= self.total:
																						snake_case       :     List[Any] =    self.total
																						snake_case       :     Tuple =    None
																						if not self.leave:
																												self.close()
																elif self.average_time_per_item is not None:
																						snake_case       :     Optional[int] =    self.average_time_per_item * (self.total - value)
																self.update_bar(SCREAMING_SNAKE_CASE_      )
																snake_case       :     List[str] =    value
																snake_case       :     List[Any] =    current_time
																if self.average_time_per_item is None:
																						snake_case       :     List[str] =    1
																else:
																						snake_case       :     Optional[int] =    max(int(self.update_every / self.average_time_per_item      ),    1      )




				def 	__snake_case							(							self	:				Any,    SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				Optional[Any]=None      ):
										snake_case       :     Optional[Any] =    ''' ''' * (len(str(self.total      )      ) - len(str(SCREAMING_SNAKE_CASE_      )      )) + str(SCREAMING_SNAKE_CASE_      )
										if self.elapsed_time is None:
																snake_case       :     int =    F"""[{spaced_value}/{self.total} : < :"""
										elif self.predicted_remaining is None:
																snake_case       :     List[str] =    F"""[{spaced_value}/{self.total} {format_time(self.elapsed_time      )}"""
										else:
																snake_case       :     Optional[int] =    (
																    F"""[{spaced_value}/{self.total} {format_time(self.elapsed_time      )} <"""
																    F""" {format_time(self.predicted_remaining      )}"""
																)
																self.label += F""", {1/self.average_time_per_item:.2f} it/s"""
										self.label += "]" if self.comment is None or len(self.comment      ) == 0 else F""", {self.comment}]"""
										self.display()




				def 	__snake_case							(							self	:				Optional[int]      ):
										snake_case       :     str =    html_progress_bar(self.value,    self.total,    self.prefix,    self.label,    self.width      )
										if self.parent is not None:
																# If this is a child bar, the parent will take care of the display.
																self.parent.display()
																return
										if self.output is None:
																snake_case       :     Any =    disp.display(disp.HTML(self.html_code      ),    display_id=SCREAMING_SNAKE_CASE_      )
										else:
																self.output.update(disp.HTML(self.html_code      )      )




				def 	__snake_case							(							self	:				Optional[int]      ):
										if self.parent is None and self.output is not None:
																self.output.update(disp.HTML(''''''      )      )




class 						a   (				__magic_name__    ):




				def __init__(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				Optional[int]=None      ):
										super().__init__(SCREAMING_SNAKE_CASE_      )
										snake_case       :     Tuple =    None if column_names is None else [column_names]
										snake_case       :     str =    None




				def 	__snake_case							(							self	:				Dict      ):
										snake_case       :     Tuple =    html_progress_bar(self.value,    self.total,    self.prefix,    self.label,    self.width      )
										if self.inner_table is not None:
																self.html_code += text_to_html_table(self.inner_table      )
										if self.child_bar is not None:
																self.html_code += self.child_bar.html_code
										if self.output is None:
																snake_case       :     List[Any] =    disp.display(disp.HTML(self.html_code      ),    display_id=SCREAMING_SNAKE_CASE_      )
										else:
																self.output.update(disp.HTML(self.html_code      )      )




				def 	__snake_case							(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				Any      ):
										if self.inner_table is None:
																snake_case       :     Optional[Any] =    [list(values.keys()      ), list(values.values()      )]
										else:
																snake_case       :     Tuple =    self.inner_table[0]
																if len(self.inner_table      ) == 1:
																						# We give a chance to update the column names at the first iteration
																						for key in values.keys():
																												if key not in columns:
																																		columns.append(SCREAMING_SNAKE_CASE_      )
																						snake_case       :     int =    columns
																self.inner_table.append([values[c] for c in columns]      )




				def 	__snake_case							(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				Optional[Any],    SCREAMING_SNAKE_CASE_	:				int=None,    SCREAMING_SNAKE_CASE_	:				str=3_00      ):
										snake_case       :     int =    NotebookProgressBar(SCREAMING_SNAKE_CASE_,    prefix=SCREAMING_SNAKE_CASE_,    parent=self,    width=SCREAMING_SNAKE_CASE_      )
										return self.child_bar




				def 	__snake_case							(							self	:				Union[str, Any]      ):
										snake_case       :     int =    None
										self.display()




class 						a   (				__magic_name__    ):




				def __init__(							self	:				Optional[Any]      ):
										snake_case       :     Dict =    None
										snake_case       :     Union[str, Any] =    None
										snake_case       :     Dict =    False




				def 	__snake_case							(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				Any,    SCREAMING_SNAKE_CASE_	:				int,    SCREAMING_SNAKE_CASE_	:				str,    **SCREAMING_SNAKE_CASE_	:				List[str]      ):
										snake_case       :     str =    '''Epoch''' if args.evaluation_strategy == IntervalStrategy.EPOCH else '''Step'''
										snake_case       :     List[str] =    0
										snake_case       :     Dict =    0
										snake_case       :     List[Any] =    [self.first_column] + ['''Training Loss''']
										if args.evaluation_strategy != IntervalStrategy.NO:
																column_names.append('''Validation Loss'''      )
										snake_case       :     Tuple =    NotebookTrainingTracker(state.max_steps,    SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				List[Any],    SCREAMING_SNAKE_CASE_	:				Dict,    SCREAMING_SNAKE_CASE_	:				str,    **SCREAMING_SNAKE_CASE_	:				Union[str, Any]      ):
										snake_case       :     Tuple =    int(state.epoch      ) if int(state.epoch      ) == state.epoch else F"""{state.epoch:.2f}"""
										self.training_tracker.update(
										    state.global_step + 1,    comment=F"""Epoch {epoch}/{state.num_train_epochs}""",    force_update=self._force_next_update,    )
										snake_case       :     str =    False




				def 	__snake_case							(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				int,    SCREAMING_SNAKE_CASE_	:				Union[str, Any],    SCREAMING_SNAKE_CASE_	:				int,    SCREAMING_SNAKE_CASE_	:				Tuple=None,    **SCREAMING_SNAKE_CASE_	:				Tuple      ):
										if not has_length(SCREAMING_SNAKE_CASE_      ):
																return
										if self.prediction_bar is None:
																if self.training_tracker is not None:
																						snake_case       :     Dict =    self.training_tracker.add_child(len(SCREAMING_SNAKE_CASE_      )      )
																else:
																						snake_case       :     Optional[Any] =    NotebookProgressBar(len(SCREAMING_SNAKE_CASE_      )      )
																self.prediction_bar.update(1      )
										else:
																self.prediction_bar.update(self.prediction_bar.value + 1      )




				def 	__snake_case							(							self	:				Optional[int],    SCREAMING_SNAKE_CASE_	:				Any,    SCREAMING_SNAKE_CASE_	:				Dict,    SCREAMING_SNAKE_CASE_	:				Union[str, Any],    **SCREAMING_SNAKE_CASE_	:				List[Any]      ):
										if self.prediction_bar is not None:
																self.prediction_bar.close()
										snake_case       :     Optional[int] =    None




				def 	__snake_case							(							self	:				Optional[Any],    SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				Any,    SCREAMING_SNAKE_CASE_	:				Optional[int],    SCREAMING_SNAKE_CASE_	:				Optional[int]=None,    **SCREAMING_SNAKE_CASE_	:				Union[str, Any]      ):
										# Only for when there is no evaluation
										if args.evaluation_strategy == IntervalStrategy.NO and "loss" in logs:
																snake_case       :     List[str] =    {'''Training Loss''': logs['''loss''']}
																# First column is necessarily Step sine we're not in epoch eval strategy
																snake_case       :     List[Any] =    state.global_step
																self.training_tracker.write_line(SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Any,    SCREAMING_SNAKE_CASE_	:				Optional[int],    SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				Optional[Any]=None,    **SCREAMING_SNAKE_CASE_	:				Any      ):
										if self.training_tracker is not None:
																snake_case       :     List[Any] =    {'''Training Loss''': '''No log''', '''Validation Loss''': '''No log'''}
																for log in reversed(state.log_history      ):
																						if "loss" in log:
																												snake_case       :     Union[str, Any] =    log['''loss''']
																												break

																if self.first_column == "Epoch":
																						snake_case       :     Optional[int] =    int(state.epoch      )
																else:
																						snake_case       :     Optional[int] =    state.global_step
																snake_case       :     Optional[int] =    '''eval'''
																for k in metrics:
																						if k.endswith('''_loss'''      ):
																												snake_case       :     str =    re.sub(R'''\_loss$''',    '''''',    SCREAMING_SNAKE_CASE_      )
																snake_case       :     str =    metrics.pop('''total_flos''',    SCREAMING_SNAKE_CASE_      )
																snake_case       :     int =    metrics.pop('''epoch''',    SCREAMING_SNAKE_CASE_      )
																snake_case       :     List[str] =    metrics.pop(F"""{metric_key_prefix}_runtime""",    SCREAMING_SNAKE_CASE_      )
																snake_case       :     Union[str, Any] =    metrics.pop(F"""{metric_key_prefix}_samples_per_second""",    SCREAMING_SNAKE_CASE_      )
																snake_case       :     Any =    metrics.pop(F"""{metric_key_prefix}_steps_per_second""",    SCREAMING_SNAKE_CASE_      )
																snake_case       :     Dict =    metrics.pop(F"""{metric_key_prefix}_jit_compilation_time""",    SCREAMING_SNAKE_CASE_      )
																for k, v in metrics.items():
																						if k == F"""{metric_key_prefix}_loss""":
																												snake_case       :     Optional[int] =    v
																						else:
																												snake_case       :     int =    k.split('''_'''      )
																												snake_case       :     Any =    ''' '''.join([part.capitalize() for part in splits[1:]]      )
																												snake_case       :     Tuple =    v
																self.training_tracker.write_line(SCREAMING_SNAKE_CASE_      )
																self.training_tracker.remove_child()
																snake_case       :     Any =    None
																# Evaluation takes a long time so we should force the next update.
																snake_case       :     Optional[int] =    True




				def 	__snake_case							(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				List[Any],    SCREAMING_SNAKE_CASE_	:				List[str],    SCREAMING_SNAKE_CASE_	:				Dict,    **SCREAMING_SNAKE_CASE_	:				str      ):
										self.training_tracker.update(
										    state.global_step,    comment=F"""Epoch {int(state.epoch      )}/{state.num_train_epochs}""",    force_update=SCREAMING_SNAKE_CASE_      )
										snake_case       :     Tuple =    None



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

UpperCAmelCase    					=   logging.getLogger()

def      A (       ):
						snake_case       :     Dict =    argparse.ArgumentParser()
						parser.add_argument('''-f'''       )
						snake_case       :     Any =    parser.parse_args()
						return args.f

def      A (       A_							:					str       ):
						snake_case       :     str =    {}
						snake_case       :     Union[str, Any] =    os.path.join(A_    ,							'''all_results.json'''       )
						if os.path.exists(A_       ):
												with open(A_    ,							'''r'''       ) as f:
																		snake_case       :     List[str] =    json.load(A_       )
						else:
												raise ValueError(F"""can't find {path}"""       )
						return results

def      A (       ):
						snake_case       :     Any =    torch.cuda.is_available() and torch_device == '''cuda'''
						return is_using_cuda and is_apex_available()


UpperCAmelCase    					=   logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)





class 						a   (				__magic_name__    ):




				@classmethod
				def 	__snake_case							(							cls	:				Tuple      ):
										# Write Accelerate config, will pick up on CPU, GPU, and multi-GPU
										snake_case       :     int =    tempfile.mkdtemp()
										snake_case       :     Optional[Any] =    os.path.join(cls.tmpdir,    '''default_config.yml'''      )
										write_basic_config(save_location=cls.configPath      )
										snake_case       :     Tuple =    ['''accelerate''', '''launch''', '''--config_file''', cls.configPath]




				@classmethod
				def 	__snake_case							(							cls	:				Optional[int]      ):
										shutil.rmtree(cls.tmpdir      )




				@mock.patch.dict(os.environ,    {'''WANDB_MODE''': '''offline'''}      )
				def 	__snake_case							(							self	:				Any      ):
										snake_case       :     Any =    self.get_auto_remove_tmp_dir()
										snake_case       :     Optional[Any] =    F"""
            {self.examples_dir}/pytorch/text-classification/run_glue_no_trainer.py
            --model_name_or_path distilbert-base-uncased
            --output_dir {tmp_dir}
            --train_file ./tests/fixtures/tests_samples/MRPC/train.csv
            --validation_file ./tests/fixtures/tests_samples/MRPC/dev.csv
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --learning_rate=1e-4
            --seed=42
            --checkpointing_steps epoch
            --with_tracking
        """.split()

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

										run_command(self._launch_args + testargs      )
										snake_case       :     int =    get_results(SCREAMING_SNAKE_CASE_      )
										self.assertGreaterEqual(result['''eval_accuracy'''],    0.75      )
										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 	__snake_case							(							self	:				Tuple      ):
										snake_case       :     int =    self.get_auto_remove_tmp_dir()
										snake_case       :     str =    F"""
            {self.examples_dir}/pytorch/language-modeling/run_clm_no_trainer.py
            --model_name_or_path distilgpt2
            --train_file ./tests/fixtures/sample_text.txt
            --validation_file ./tests/fixtures/sample_text.txt
            --block_size 128
            --per_device_train_batch_size 5
            --per_device_eval_batch_size 5
            --num_train_epochs 2
            --output_dir {tmp_dir}
            --checkpointing_steps epoch
            --with_tracking
        """.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      )
										snake_case       :     List[str] =    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 	__snake_case							(							self	:				List[Any]      ):
										snake_case       :     Union[str, Any] =    self.get_auto_remove_tmp_dir()
										snake_case       :     int =    F"""
            {self.examples_dir}/pytorch/language-modeling/run_mlm_no_trainer.py
            --model_name_or_path distilroberta-base
            --train_file ./tests/fixtures/sample_text.txt
            --validation_file ./tests/fixtures/sample_text.txt
            --output_dir {tmp_dir}
            --num_train_epochs=1
            --checkpointing_steps epoch
            --with_tracking
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     List[Any] =    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 	__snake_case							(							self	:				Dict      ):
										# with so little data distributed training needs more epochs to get the score on par with 0/1 gpu
										snake_case       :     Any =    7 if get_gpu_count() > 1 else 2

										snake_case       :     Optional[int] =    self.get_auto_remove_tmp_dir()
										snake_case       :     str =    F"""
            {self.examples_dir}/pytorch/token-classification/run_ner_no_trainer.py
            --model_name_or_path bert-base-uncased
            --train_file tests/fixtures/tests_samples/conll/sample.json
            --validation_file tests/fixtures/tests_samples/conll/sample.json
            --output_dir {tmp_dir}
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=2
            --num_train_epochs={epochs}
            --seed 7
            --checkpointing_steps epoch
            --with_tracking
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     Optional[int] =    get_results(SCREAMING_SNAKE_CASE_      )
										self.assertGreaterEqual(result['''eval_accuracy'''],    0.75      )
										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 	__snake_case							(							self	:				Tuple      ):
										snake_case       :     Tuple =    self.get_auto_remove_tmp_dir()
										snake_case       :     Optional[int] =    F"""
            {self.examples_dir}/pytorch/question-answering/run_qa_no_trainer.py
            --model_name_or_path bert-base-uncased
            --version_2_with_negative
            --train_file tests/fixtures/tests_samples/SQUAD/sample.json
            --validation_file tests/fixtures/tests_samples/SQUAD/sample.json
            --output_dir {tmp_dir}
            --seed=42
            --max_train_steps=10
            --num_warmup_steps=2
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --checkpointing_steps epoch
            --with_tracking
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     str =    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 	__snake_case							(							self	:				Optional[int]      ):
										snake_case       :     Any =    self.get_auto_remove_tmp_dir()
										snake_case       :     List[Any] =    F"""
            {self.examples_dir}/pytorch/multiple-choice/run_swag_no_trainer.py
            --model_name_or_path bert-base-uncased
            --train_file tests/fixtures/tests_samples/swag/sample.json
            --validation_file tests/fixtures/tests_samples/swag/sample.json
            --output_dir {tmp_dir}
            --max_train_steps=20
            --num_warmup_steps=2
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --with_tracking
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     Union[str, Any] =    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 	__snake_case							(							self	:				str      ):
										snake_case       :     str =    self.get_auto_remove_tmp_dir()
										snake_case       :     Any =    F"""
            {self.examples_dir}/pytorch/summarization/run_summarization_no_trainer.py
            --model_name_or_path t5-small
            --train_file tests/fixtures/tests_samples/xsum/sample.json
            --validation_file tests/fixtures/tests_samples/xsum/sample.json
            --output_dir {tmp_dir}
            --max_train_steps=50
            --num_warmup_steps=8
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --checkpointing_steps epoch
            --with_tracking
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     str =    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 	__snake_case							(							self	:				int      ):
										snake_case       :     Any =    self.get_auto_remove_tmp_dir()
										snake_case       :     Optional[int] =    F"""
            {self.examples_dir}/pytorch/translation/run_translation_no_trainer.py
            --model_name_or_path sshleifer/student_marian_en_ro_6_1
            --source_lang en
            --target_lang ro
            --train_file tests/fixtures/tests_samples/wmt16/sample.json
            --validation_file tests/fixtures/tests_samples/wmt16/sample.json
            --output_dir {tmp_dir}
            --max_train_steps=50
            --num_warmup_steps=8
            --num_beams=6
            --learning_rate=3e-3
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --source_lang en_XX
            --target_lang ro_RO
            --checkpointing_steps epoch
            --with_tracking
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     Optional[int] =    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 	__snake_case							(							self	:				Union[str, Any]      ):
										snake_case       :     Any =    logging.StreamHandler(sys.stdout      )
										logger.addHandler(SCREAMING_SNAKE_CASE_      )

										snake_case       :     Union[str, Any] =    self.get_auto_remove_tmp_dir()
										snake_case       :     Dict =    F"""
            {self.examples_dir}/pytorch/semantic-segmentation/run_semantic_segmentation_no_trainer.py
            --dataset_name huggingface/semantic-segmentation-test-sample
            --output_dir {tmp_dir}
            --max_train_steps=10
            --num_warmup_steps=2
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --checkpointing_steps epoch
        """.split()

										run_command(self._launch_args + testargs      )
										snake_case       :     List[Any] =    get_results(SCREAMING_SNAKE_CASE_      )
										self.assertGreaterEqual(result['''eval_overall_accuracy'''],    0.10      )




				@mock.patch.dict(os.environ,    {'''WANDB_MODE''': '''offline'''}      )
				def 	__snake_case							(							self	:				Union[str, Any]      ):
										snake_case       :     str =    self.get_auto_remove_tmp_dir()
										snake_case       :     Optional[Any] =    F"""
            {self.examples_dir}/pytorch/image-classification/run_image_classification_no_trainer.py
            --model_name_or_path google/vit-base-patch16-224-in21k
            --dataset_name hf-internal-testing/cats_vs_dogs_sample
            --learning_rate 1e-4
            --per_device_train_batch_size 2
            --per_device_eval_batch_size 1
            --max_train_steps 2
            --train_val_split 0.1
            --seed 42
            --output_dir {tmp_dir}
            --with_tracking
            --checkpointing_steps 1
        """.split()

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

										run_command(self._launch_args + testargs      )
										snake_case       :     Tuple =    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'''      )      )      )

