






import math
from typing import Dict, Iterable, List, Optional, Tuple, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    get_image_size,
    is_torch_available,
    is_torch_tensor,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_torch_available():
						import torch

if is_vision_available():
						import PIL


a_		:						Optional[int]             =    logging.get_logger(__name__)






def 				lowerCamelCase__				(_UpperCAmelCase							,       _UpperCAmelCase							,       _UpperCAmelCase							,       _UpperCAmelCase):


			def constraint_to_multiple_of(_UpperCAmelCase							,       _UpperCAmelCase							,       _UpperCAmelCase=0							,       _UpperCAmelCase=None):
						SCREAMING_SNAKE_CASE						= round(val / multiple) * multiple

						if max_val is not None and x > max_val:
									SCREAMING_SNAKE_CASE						= math.floor(val / multiple) * multiple

						if x < min_val:
									SCREAMING_SNAKE_CASE						= math.ceil(val / multiple) * multiple

						return x

			SCREAMING_SNAKE_CASE						= (output_size, output_size) if isinstance(_UpperCamelCase							,       _UpperCamelCase) else output_size

			SCREAMING_SNAKE_CASE				,						SCREAMING_SNAKE_CASE						= get_image_size(_UpperCamelCase)
			SCREAMING_SNAKE_CASE				,						SCREAMING_SNAKE_CASE						= output_size

			# determine new height and width
			SCREAMING_SNAKE_CASE						= output_height / input_height
			SCREAMING_SNAKE_CASE						= output_width / input_width

			if keep_aspect_ratio:
						# scale as little as possible
						if abs(1 - scale_width) < abs(1 - scale_height):
									# fit width
									SCREAMING_SNAKE_CASE						= scale_width
						else:
									# fit height
									SCREAMING_SNAKE_CASE						= scale_height

			SCREAMING_SNAKE_CASE						= constraint_to_multiple_of(scale_height * input_height							,       multiple=_UpperCamelCase)
			SCREAMING_SNAKE_CASE						= constraint_to_multiple_of(scale_width * input_width							,       multiple=_UpperCamelCase)

			return (new_height, new_width)







class 		_snake_case  (  A__					):
			_lowercase		:    List[str]			     =							['''pixel_values''']






			def __init__(       self		,					a = True		,					a = None		,					a = PILImageResampling.BILINEAR		,					a = False		,					a = 1		,					a = True		,					a = 1 / 255		,					a = True		,					a = None		,					a = None		,					**a		,					)      ->       None:
						super().__init__(**UpperCAmelCase__)
						SCREAMING_SNAKE_CASE						= size if size is not None else {'height': 384, 'width': 384}
						SCREAMING_SNAKE_CASE						= get_size_dict(UpperCAmelCase__)
						SCREAMING_SNAKE_CASE						= do_resize
						SCREAMING_SNAKE_CASE						= size
						SCREAMING_SNAKE_CASE						= keep_aspect_ratio
						SCREAMING_SNAKE_CASE						= ensure_multiple_of
						SCREAMING_SNAKE_CASE						= resample
						SCREAMING_SNAKE_CASE						= do_rescale
						SCREAMING_SNAKE_CASE						= rescale_factor
						SCREAMING_SNAKE_CASE						= do_normalize
						SCREAMING_SNAKE_CASE						= image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
						SCREAMING_SNAKE_CASE						= image_std if image_std is not None else IMAGENET_STANDARD_STD






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a = False		,					a = 1		,					a = PILImageResampling.BICUBIC		,					a = None		,					**a		,					)      ->       np.ndarray:
						SCREAMING_SNAKE_CASE						= get_size_dict(UpperCAmelCase__)
						if "height" not in size or "width" not in size:
									raise ValueError(f'''The size dictionary must contain the keys \'height\' and \'width\'. Got {size.keys()}''')
						SCREAMING_SNAKE_CASE						= get_resize_output_image_size(
						    UpperCAmelCase__		,					output_size=(size['height'], size['width'])		,					keep_aspect_ratio=UpperCAmelCase__		,					multiple=UpperCAmelCase__		,					)
						return resize(UpperCAmelCase__		,					size=UpperCAmelCase__		,					resample=UpperCAmelCase__		,					data_format=UpperCAmelCase__		,					**UpperCAmelCase__)






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a = None		,					**a		,					)      ->       Any:
						return rescale(UpperCAmelCase__		,					scale=UpperCAmelCase__		,					data_format=UpperCAmelCase__		,					**UpperCAmelCase__)






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a		,					a = None		,					**a		,					)      ->       np.ndarray:
						return normalize(UpperCAmelCase__		,					mean=UpperCAmelCase__		,					std=UpperCAmelCase__		,					data_format=UpperCAmelCase__		,					**UpperCAmelCase__)






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = None		,					a = ChannelDimension.FIRST		,					**a		,					)      ->       PIL.Image.Image:
						SCREAMING_SNAKE_CASE						= do_resize if do_resize is not None else self.do_resize
						SCREAMING_SNAKE_CASE						= size if size is not None else self.size
						SCREAMING_SNAKE_CASE						= get_size_dict(UpperCAmelCase__)
						SCREAMING_SNAKE_CASE						= keep_aspect_ratio if keep_aspect_ratio is not None else self.keep_aspect_ratio
						SCREAMING_SNAKE_CASE						= ensure_multiple_of if ensure_multiple_of is not None else self.ensure_multiple_of
						SCREAMING_SNAKE_CASE						= resample if resample is not None else self.resample
						SCREAMING_SNAKE_CASE						= do_rescale if do_rescale is not None else self.do_rescale
						SCREAMING_SNAKE_CASE						= rescale_factor if rescale_factor is not None else self.rescale_factor
						SCREAMING_SNAKE_CASE						= do_normalize if do_normalize is not None else self.do_normalize
						SCREAMING_SNAKE_CASE						= image_mean if image_mean is not None else self.image_mean
						SCREAMING_SNAKE_CASE						= image_std if image_std is not None else self.image_std

						SCREAMING_SNAKE_CASE						= make_list_of_images(UpperCAmelCase__)

						if not valid_images(UpperCAmelCase__):
									raise ValueError(
									    'Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '
									    'torch.Tensor, tf.Tensor or jax.ndarray.')

						if do_resize and size is None or resample is None:
									raise ValueError('Size and resample must be specified if do_resize is True.')

						if do_rescale and rescale_factor is None:
									raise ValueError('Rescale factor must be specified if do_rescale is True.')

						if do_normalize and (image_mean is None or image_std is None):
									raise ValueError('Image mean and std must be specified if do_normalize is True.')

						# All transformations expect numpy arrays.
						SCREAMING_SNAKE_CASE						= [to_numpy_array(UpperCAmelCase__) for image in images]

						if do_resize:
									SCREAMING_SNAKE_CASE						= [self.resize(image=UpperCAmelCase__		,					size=UpperCAmelCase__		,					resample=UpperCAmelCase__) for image in images]

						if do_rescale:
									SCREAMING_SNAKE_CASE						= [self.rescale(image=UpperCAmelCase__		,					scale=UpperCAmelCase__) for image in images]

						if do_normalize:
									SCREAMING_SNAKE_CASE						= [self.normalize(image=UpperCAmelCase__		,					mean=UpperCAmelCase__		,					std=UpperCAmelCase__) for image in images]

						SCREAMING_SNAKE_CASE						= [to_channel_dimension_format(UpperCAmelCase__		,					UpperCAmelCase__) for image in images]

						SCREAMING_SNAKE_CASE						= {'pixel_values': images}
						return BatchFeature(data=UpperCAmelCase__		,					tensor_type=UpperCAmelCase__)







			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a = None)      ->       Optional[Any]:
						SCREAMING_SNAKE_CASE						= outputs.logits

						# Resize logits and compute semantic segmentation maps
						if target_sizes is not None:
									if len(UpperCAmelCase__) != len(UpperCAmelCase__):
												raise ValueError(
												    'Make sure that you pass in as many target sizes as the batch dimension of the logits')

									if is_torch_tensor(UpperCAmelCase__):
												SCREAMING_SNAKE_CASE						= target_sizes.numpy()

									SCREAMING_SNAKE_CASE						= []

									for idx in range(len(UpperCAmelCase__)):
												SCREAMING_SNAKE_CASE						= torch.nn.functional.interpolate(
												    logits[idx].unsqueeze(dim=0)		,					size=target_sizes[idx]		,					mode='bilinear'		,					align_corners=UpperCAmelCase__)
												SCREAMING_SNAKE_CASE						= resized_logits[0].argmax(dim=0)
												semantic_segmentation.append(UpperCAmelCase__)
						else:
									SCREAMING_SNAKE_CASE						= logits.argmax(dim=1)
									SCREAMING_SNAKE_CASE						= [semantic_segmentation[i] for i in range(semantic_segmentation.shape[0])]

						return semantic_segmentation












# 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


a_		:						Dict             =    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)

a_		:						List[str]             =    {'base': 'patrickvonplaten/wav2vec2_tiny_random', 'robust': 'patrickvonplaten/wav2vec2_tiny_random_robust'}

a_		:						Union[str, Any]             =    'zero2'
a_		:						List[Any]             =    'zero3'
a_		:						List[str]             =    [ZEROa, ZEROa]






def 				lowerCamelCase__				(_UpperCAmelCase							,       _UpperCAmelCase							,       _UpperCAmelCase):
			# customize the test name generator function as we want both params to appear in the sub-test
			# name, as by default it shows only the first param
			SCREAMING_SNAKE_CASE						= parameterized.to_safe_name('_'.join(str(_UpperCAmelCase) for x in param.args))
			return F'''{func.__name__}_{param_based_name}'''


# Cartesian-product of zero stages with models to test
a_		:						List[Any]             =    list(itertools.product(stages, models.keys()))







@slow
@require_deepspeed
@require_torch_gpu
class 		_snake_case  (  A__					):






			@parameterized.expand(a		,					name_func=a)
			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a)      ->       List[str]:
						self.run_and_check(
						    stage=a		,					model=a		,					distributed=a		,					fpaa=a		,					)






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






			@parameterized.expand(a		,					name_func=a)
			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a)      ->       int:
						self.run_and_check(
						    stage=a		,					model=a		,					distributed=a		,					fpaa=a		,					)






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






			def    SCREAMING_SNAKE_CASE__    (       self		,					a)      ->       Dict:
						# XXX: run_asr is premature and doesn't save any results
						# so all we check for now is that the process didn't fail
						pass






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a = 10		,					a = True		,					a = True		,					a = True		,					)      ->       Dict:
						SCREAMING_SNAKE_CASE						= models[model]

						SCREAMING_SNAKE_CASE						= self.run_trainer(
						    stage=a		,					model_name=a		,					eval_steps=a		,					num_train_epochs=1		,					distributed=a		,					fpaa=a		,					)

						self.do_checks(a)

						return output_dir






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

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

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

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

						return output_dir







			def    SCREAMING_SNAKE_CASE__    (       self		,					a=False)      ->       Optional[int]:
						# 1. explicitly set --num_nodes=1 just in case these tests end up run on a multi-node setup
						# - it won't be able to handle that
						# 2. for now testing with just 2 gpus max (since some quality tests may give different
						# results with mode gpus because we use very little data)
						SCREAMING_SNAKE_CASE						= min(2		,					get_gpu_count()) if distributed else 1
						return f'''deepspeed --num_nodes 1 --num_gpus {num_gpus}'''.split()




