





"""simple docstring"""





# XXX: we want transformers master here - in the absense of conftest manipulating sys.path:
# hack it in for now:
import sys
from pathlib import Path


lowerCAmelCase__   	=	Path(__file__).resolve().parents[3] / '''src'''
sys.path.insert(1, str(git_repo_path))

import dataclasses  # noqa
import io  # noqa
import itertools  # noqa
import json  # noqa
import os  # noqa
import unittest  # noqa
from copy import deepcopy  # noqa

from parameterized import parameterized  # noqa
from transformers import TrainingArguments, is_torch_available  # noqa
from transformers.deepspeed import is_deepspeed_available  # noqa
from transformers.file_utils import WEIGHTS_NAME  # noqa
from transformers.testing_utils import (  # noqa
    CaptureLogger,
    ExtendSysPath,
    TestCasePlus,
    execute_subprocess_async,
    get_gpu_count,
    mockenv_context,
    require_deepspeed,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)
from transformers.trainer_utils import set_seed  # noqa


set_seed(42)

lowerCAmelCase__   	=	{'''base''': '''patrickvonplaten/wav2vec2_tiny_random''', '''robust''': '''patrickvonplaten/wav2vec2_tiny_random_robust'''}

lowerCAmelCase__   	=	'''zero2'''
lowerCAmelCase__   	=	'''zero3'''
lowerCAmelCase__   	=	[ZEROa, ZEROa]


def 					a__    (    _SCREAMING_SNAKE_CASE     ,		_SCREAMING_SNAKE_CASE     ,		_SCREAMING_SNAKE_CASE			):




		"""simple docstring"""


		UpperCamelCase      =		parameterized.to_safe_name("_".join(str(_SCREAMING_SNAKE_CASE			) for x in param.args			)			)
		return F"{func.__name__}_{param_based_name}"


# Cartesian-product of zero stages with models to test
lowerCAmelCase__   	=	list(itertools.product(stages, models.keys()))






@slow
@require_deepspeed
@require_torch_gpu
class 				_lowerCamelCase					(				UpperCamelCase__       ):


						@parameterized.expand(UpperCamelCase_      ,							name_func=UpperCamelCase_						)
						def 			snake_case_      (self      ,							__a      ,							__a						)       ->      Optional[Any]:
								self.run_and_check(
								    stage=UpperCamelCase_      ,							model=UpperCamelCase_      ,							distributed=UpperCamelCase_      ,							fpaa=UpperCamelCase_      ,							)


						@require_torch_multi_gpu
						@parameterized.expand(UpperCamelCase_      ,							name_func=UpperCamelCase_						)
						def 			snake_case_      (self      ,							__a      ,							__a						)       ->      List[Any]:
								self.run_and_check(
								    stage=UpperCamelCase_      ,							model=UpperCamelCase_      ,							distributed=UpperCamelCase_      ,							fpaa=UpperCamelCase_      ,							)


						@parameterized.expand(UpperCamelCase_      ,							name_func=UpperCamelCase_						)
						def 			snake_case_      (self      ,							__a      ,							__a						)       ->      Dict:
								self.run_and_check(
								    stage=UpperCamelCase_      ,							model=UpperCamelCase_      ,							distributed=UpperCamelCase_      ,							fpaa=UpperCamelCase_      ,							)


						@require_torch_multi_gpu
						@parameterized.expand(UpperCamelCase_      ,							name_func=UpperCamelCase_						)
						def 			snake_case_      (self      ,							__a      ,							__a						)       ->      str:
								self.run_and_check(
								    stage=UpperCamelCase_      ,							model=UpperCamelCase_      ,							distributed=UpperCamelCase_      ,							fpaa=UpperCamelCase_      ,							)


						def 			snake_case_      (self      ,							__a						)       ->      Tuple:
								pass


						def 			snake_case_      (self      ,							__a      ,							__a      ,							__a = 10      ,							__a = True      ,							__a = True      ,							__a = True      ,							)       ->      List[str]:
								UpperCamelCase      =		models[model]

								UpperCamelCase      =		self.run_trainer(
								    stage=UpperCamelCase_      ,							model_name=UpperCamelCase_      ,							eval_steps=UpperCamelCase_      ,							num_train_epochs=1      ,							distributed=UpperCamelCase_      ,							fpaa=UpperCamelCase_      ,							)

								self.do_checks(UpperCamelCase_						)

								return output_dir


						def 			snake_case_      (self      ,							__a      ,							__a      ,							__a = 10      ,							__a = 1      ,							__a = True      ,							__a = True      ,							)       ->      List[str]:
								UpperCamelCase      =		self.get_auto_remove_tmp_dir("./xxx"      ,							after=UpperCamelCase_						)
								UpperCamelCase      =		F"\n            --model_name_or_path {model_name}\n            --dataset_name hf-internal-testing/librispeech_asr_dummy\n            --dataset_config_name clean\n            --train_split_name validation\n            --validation_split_name validation\n            --output_dir {output_dir}\n            --num_train_epochs {str(UpperCamelCase_						)}\n            --per_device_train_batch_size 2\n            --per_device_eval_batch_size 2\n            --evaluation_strategy steps\n            --learning_rate 5e-4\n            --warmup_steps 8\n            --orthography timit\n            --preprocessing_num_workers 1\n            --group_by_length\n            --freeze_feature_extractor\n            --report_to none\n            --save_steps 0\n            --eval_steps {eval_steps}\n            --report_to none\n        ".split()

								if fpaa:
										args.extend(["--fp16"]						)

								# currently ds_config_wav2vec2_zero.json requires "zero_optimization.find_unused_parameters": true,
								# hence the separate config files
								UpperCamelCase      =		F"--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json".split()
								UpperCamelCase      =		[F"{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py"]
								UpperCamelCase      =		self.get_launcher(UpperCamelCase_						)

								UpperCamelCase      =		launcher + script + args + ds_args
								# keep for quick debug
								# print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] +cmd)); die
								execute_subprocess_async(UpperCamelCase_      ,							env=self.get_env()						)

								return output_dir




						def 			snake_case_      (self      ,							__a=False						)       ->      str:
								UpperCamelCase      =		min(2      ,							get_gpu_count()						) if distributed else 1
								return F"deepspeed --num_nodes 1 --num_gpus {num_gpus}".split()










import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
			import torch

if is_vision_available():
			from PIL import Image

			from transformers import LevitImageProcessor


class        _a							(			unittest.TestCase			):

							def __init__(							self:	Dict    ,						UpperCamelCase_:	Any    ,						UpperCamelCase_:	Dict=7    ,						UpperCamelCase_:	Optional[int]=3    ,						UpperCamelCase_:	List[Any]=18    ,						UpperCamelCase_:	Dict=30    ,						UpperCamelCase_:	Optional[int]=400    ,						UpperCamelCase_:	Optional[int]=True    ,						UpperCamelCase_:	Union[str, Any]=None    ,						UpperCamelCase_:	List[str]=True    ,						UpperCamelCase_:	Dict=None    ,						UpperCamelCase_:	Optional[int]=True    ,						UpperCamelCase_:	Tuple=[0.5, 0.5, 0.5]    ,						UpperCamelCase_:	List[str]=[0.5, 0.5, 0.5]    ,						)			->   int:





									"""simple docstring"""







									lowercase__            =      size if size is not None else {'''shortest_edge''': 18}
									lowercase__            =      crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}
									lowercase__            =      parent
									lowercase__            =      batch_size
									lowercase__            =      num_channels
									lowercase__            =      image_size
									lowercase__            =      min_resolution
									lowercase__            =      max_resolution
									lowercase__            =      do_resize
									lowercase__            =      size
									lowercase__            =      do_center_crop
									lowercase__            =      crop_size
									lowercase__            =      do_normalize
									lowercase__            =      image_mean
									lowercase__            =      image_std


							def       lowerCamelCase_ (							self:	List[str]							)			->   Optional[Any]:





									"""simple docstring"""







									return {
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_normalize": self.do_normalize,
									    "do_resize": self.do_resize,
									    "do_center_crop": self.do_center_crop,
									    "size": self.size,
									    "crop_size": self.crop_size,
									}


@require_torch
@require_vision
class        _a							(			UpperCamelCase__     ,	unittest.TestCase			):
							_lowercase	:			List[str]        = LevitImageProcessor if is_vision_available() else None

							def       lowerCamelCase_ (							self:	Optional[Any]							)			->   List[str]:





									"""simple docstring"""







									lowercase__            =      LevitImageProcessingTester(self							)

							@property
							def       lowerCamelCase_ (							self:	List[Any]							)			->   Any:





									"""simple docstring"""







									return self.image_processor_tester.prepare_image_processor_dict()

							def       lowerCamelCase_ (							self:	Union[str, Any]							)			->   str:





									"""simple docstring"""







									lowercase__            =      self.image_processing_class(**self.image_processor_dict							)
									self.assertTrue(hasattr(UpperCamelCase_    ,						'''image_mean'''							)							)
									self.assertTrue(hasattr(UpperCamelCase_    ,						'''image_std'''							)							)
									self.assertTrue(hasattr(UpperCamelCase_    ,						'''do_normalize'''							)							)
									self.assertTrue(hasattr(UpperCamelCase_    ,						'''do_resize'''							)							)
									self.assertTrue(hasattr(UpperCamelCase_    ,						'''do_center_crop'''							)							)
									self.assertTrue(hasattr(UpperCamelCase_    ,						'''size'''							)							)

							def       lowerCamelCase_ (							self:	str							)			->   Tuple:





									"""simple docstring"""







									lowercase__            =      self.image_processing_class.from_dict(self.image_processor_dict							)
									self.assertEqual(image_processor.size    ,						{'''shortest_edge''': 18}							)
									self.assertEqual(image_processor.crop_size    ,						{'''height''': 18, '''width''': 18}							)

									lowercase__            =      self.image_processing_class.from_dict(self.image_processor_dict    ,						size=42    ,						crop_size=84							)
									self.assertEqual(image_processor.size    ,						{'''shortest_edge''': 42}							)
									self.assertEqual(image_processor.crop_size    ,						{'''height''': 84, '''width''': 84}							)

							def       lowerCamelCase_ (							self:	List[str]							)			->   Dict:





									"""simple docstring"""







									pass

							def       lowerCamelCase_ (							self:	List[str]							)			->   int:





									"""simple docstring"""







									lowercase__            =      self.image_processing_class(**self.image_processor_dict							)
									# create random PIL images
									lowercase__            =      prepare_image_inputs(self.image_processor_tester    ,						equal_resolution=UpperCamelCase_							)
									for image in image_inputs:
											self.assertIsInstance(UpperCamelCase_    ,						Image.Image							)

									# Test not batched input
									lowercase__            =      image_processing(image_inputs[0]    ,						return_tensors='''pt'''							).pixel_values
									self.assertEqual(
									    encoded_images.shape    ,						(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,						)

									# Test batched
									lowercase__            =      image_processing(UpperCamelCase_    ,						return_tensors='''pt'''							).pixel_values
									self.assertEqual(
									    encoded_images.shape    ,						(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,						)

							def       lowerCamelCase_ (							self:	Union[str, Any]							)			->   Optional[int]:





									"""simple docstring"""







									lowercase__            =      self.image_processing_class(**self.image_processor_dict							)
									# create random numpy tensors
									lowercase__            =      prepare_image_inputs(self.image_processor_tester    ,						equal_resolution=UpperCamelCase_    ,						numpify=UpperCamelCase_							)
									for image in image_inputs:
											self.assertIsInstance(UpperCamelCase_    ,						np.ndarray							)

									# Test not batched input
									lowercase__            =      image_processing(image_inputs[0]    ,						return_tensors='''pt'''							).pixel_values
									self.assertEqual(
									    encoded_images.shape    ,						(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,						)

									# Test batched
									lowercase__            =      image_processing(UpperCamelCase_    ,						return_tensors='''pt'''							).pixel_values
									self.assertEqual(
									    encoded_images.shape    ,						(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,						)


							def       lowerCamelCase_ (							self:	Any							)			->   Optional[Any]:





									"""simple docstring"""







									lowercase__            =      self.image_processing_class(**self.image_processor_dict							)
									# create random PyTorch tensors
									lowercase__            =      prepare_image_inputs(self.image_processor_tester    ,						equal_resolution=UpperCamelCase_    ,						torchify=UpperCamelCase_							)
									for image in image_inputs:
											self.assertIsInstance(UpperCamelCase_    ,						torch.Tensor							)

									# Test not batched input
									lowercase__            =      image_processing(image_inputs[0]    ,						return_tensors='''pt'''							).pixel_values
									self.assertEqual(
									    encoded_images.shape    ,						(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,						)

									# Test batched
									lowercase__            =      image_processing(UpperCamelCase_    ,						return_tensors='''pt'''							).pixel_values
									self.assertEqual(
									    encoded_images.shape    ,						(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,						)


