
import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(42)

UpperCamelCase_						:     List[str]	       =							"""bert-base-cased"""
UpperCamelCase_						:     str	       =							"""fp16"""
UpperCamelCase_						:     List[str]	       =							"""bf16"""
UpperCamelCase_						:     Any	       =							[FPaa, BFaa]




@require_fsdp
@require_cuda
class   __lowercase						(						__snake_case       ):






	def       _a(self				: str    ) ->			str:
							super().setUp()

							_lowercase       :	Tuple           =   dict(
							    ACCELERATE_USE_FSDP="true"      ,	MASTER_ADDR="localhost"      ,	MASTER_PORT="10999"      ,	RANK="0"      ,	LOCAL_RANK="0"      ,	WORLD_SIZE="1"      ,	)






	def       _a(self				: Optional[Any]    ) ->			Tuple:
							from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

							for i, strategy in enumerate(snake_case    ):
													_lowercase       :	Optional[Any]           =   self.dist_env.copy()
													_lowercase       :	str           =   F"""{i + 1}"""
													_lowercase       :	str           =   strategy
													with mockenv_context(**snake_case    ):
																			_lowercase       :	int           =   FullyShardedDataParallelPlugin()
																			self.assertEqual(fsdp_plugin.sharding_strategy      ,	ShardingStrategy(i + 1    )    )






	def       _a(self				: Tuple    ) ->			List[str]:
							from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

							for i, prefetch_policy in enumerate(snake_case    ):
													_lowercase       :	Optional[int]           =   self.dist_env.copy()
													_lowercase       :	List[str]           =   prefetch_policy
													with mockenv_context(**snake_case    ):
																			_lowercase       :	List[str]           =   FullyShardedDataParallelPlugin()
																			if prefetch_policy == "NO_PREFETCH":
																									self.assertIsNone(fsdp_plugin.backward_prefetch    )
																			else:
																									self.assertEqual(fsdp_plugin.backward_prefetch      ,	BackwardPrefetch(i + 1    )    )






	def       _a(self				: Optional[Any]    ) ->			Optional[Any]:
							from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

							for i, state_dict_type in enumerate(snake_case    ):
													_lowercase       :	Optional[int]           =   self.dist_env.copy()
													_lowercase       :	Dict           =   state_dict_type
													with mockenv_context(**snake_case    ):
																			_lowercase       :	List[str]           =   FullyShardedDataParallelPlugin()
																			self.assertEqual(fsdp_plugin.state_dict_type      ,	StateDictType(i + 1    )    )
																			if state_dict_type == "FULL_STATE_DICT":
																									self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu    )
																									self.assertTrue(fsdp_plugin.state_dict_config.ranka_only    )






	def       _a(self				: Optional[int]    ) ->			List[str]:
							_lowercase       :	int           =   AutoModel.from_pretrained(snake_case    )
							for policy in FSDP_AUTO_WRAP_POLICY:
													_lowercase       :	Tuple           =   self.dist_env.copy()
													_lowercase       :	List[str]           =   policy
													if policy == "TRANSFORMER_BASED_WRAP":
																			_lowercase       :	Optional[int]           =   "BertLayer"
													elif policy == "SIZE_BASED_WRAP":
																			_lowercase       :	int           =   "2000"
													with mockenv_context(**snake_case    ):
																			_lowercase       :	Dict           =   FullyShardedDataParallelPlugin()
																			fsdp_plugin.set_auto_wrap_policy(snake_case    )
																			if policy == "NO_WRAP":
																									self.assertIsNone(fsdp_plugin.auto_wrap_policy    )
																			else:
																									self.assertIsNotNone(fsdp_plugin.auto_wrap_policy    )

							_lowercase       :	Union[str, Any]           =   self.dist_env.copy()
							_lowercase       :	Union[str, Any]           =   "TRANSFORMER_BASED_WRAP"
							_lowercase       :	str           =   "T5Layer"
							with mockenv_context(**snake_case    ):
													_lowercase       :	List[str]           =   FullyShardedDataParallelPlugin()
													with self.assertRaises(snake_case    ) as cm:
																			fsdp_plugin.set_auto_wrap_policy(snake_case    )
													self.assertTrue("Could not find the transformer layer class to wrap in the model." in str(cm.exception    )    )

							_lowercase       :	Union[str, Any]           =   self.dist_env.copy()
							_lowercase       :	List[Any]           =   "SIZE_BASED_WRAP"
							_lowercase       :	Union[str, Any]           =   "0"
							with mockenv_context(**snake_case    ):
													_lowercase       :	List[Any]           =   FullyShardedDataParallelPlugin()
													fsdp_plugin.set_auto_wrap_policy(snake_case    )
													self.assertIsNone(fsdp_plugin.auto_wrap_policy    )






	def       _a(self				: str    ) ->			int:
							from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
							from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

							for mp_dtype in dtypes:
													_lowercase       :	Tuple           =   self.dist_env.copy()
													_lowercase       :	Optional[int]           =   mp_dtype
													with mockenv_context(**snake_case    ):
																			_lowercase       :	Tuple           =   Accelerator()
																			if mp_dtype == "fp16":
																									_lowercase       :	str           =   torch.floataa
																			elif mp_dtype == "bf16":
																									_lowercase       :	List[str]           =   torch.bfloataa
																			_lowercase       :	Dict           =   MixedPrecision(param_dtype=snake_case      ,	reduce_dtype=snake_case      ,	buffer_dtype=snake_case    )
																			self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy      ,	snake_case    )
																			if mp_dtype == FPaa:
																									self.assertTrue(isinstance(accelerator.scaler      ,	snake_case    )    )
																			elif mp_dtype == BFaa:
																									self.assertIsNone(accelerator.scaler    )
																			AcceleratorState._reset_state(snake_case    )






	def       _a(self				: Tuple    ) ->			Union[str, Any]:
							from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

							for flag in [True, False]:
													_lowercase       :	List[Any]           =   self.dist_env.copy()
													_lowercase       :	Dict           =   str(snake_case    ).lower()
													with mockenv_context(**snake_case    ):
																			_lowercase       :	int           =   FullyShardedDataParallelPlugin()
																			self.assertEqual(fsdp_plugin.cpu_offload      ,	CPUOffload(offload_params=snake_case    )    )




@require_fsdp
@require_multi_gpu
@slow
class   __lowercase						(						__snake_case       ):






	def       _a(self				: str    ) ->			Optional[Any]:
							super().setUp()
							_lowercase       :	Dict           =   0.82
							_lowercase       :	int           =   [
							    "fsdp_shard_grad_op_transformer_based_wrap",
							    "fsdp_full_shard_transformer_based_wrap",
							]
							_lowercase       :	Dict           =   {
							    "multi_gpu_fp16": 3200,
							    "fsdp_shard_grad_op_transformer_based_wrap_fp16": 2000,
							    "fsdp_full_shard_transformer_based_wrap_fp16": 1900,
							    # Disabling below test as it overwhelms the RAM memory usage
							    # on CI self-hosted runner leading to tests getting killed.
							    # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
							}
							_lowercase       :	Optional[Any]           =   160
							_lowercase       :	int           =   160

							_lowercase       :	str           =   inspect.getfile(accelerate.test_utils    )
							_lowercase       :	Tuple           =   os.path.sep.join(mod_file.split(os.path.sep    )[:-1] + ["scripts", "external_deps"]    )






	def       _a(self				: Union[str, Any]    ) ->			Optional[Any]:
							_lowercase       :	Any           =   os.path.join(self.test_scripts_folder      ,	"test_performance.py"    )
							_lowercase       :	Dict           =   ["accelerate", "launch", "--num_processes=2", "--num_machines=1", "--machine_rank=0", "--use_fsdp"]
							for config in self.performance_configs:
													_lowercase       :	Union[str, Any]           =   cmd.copy()
													for i, strategy in enumerate(snake_case    ):
																			if strategy.lower() in config:
																									cmd_config.append(F"""--fsdp_sharding_strategy={i+1}"""    )
																									break

													if "fp32" in config:
																			cmd_config.append("--mixed_precision=no"    )
													else:
																			cmd_config.append("--mixed_precision=fp16"    )

													if "cpu_offload" in config:
																			cmd_config.append("--fsdp_offload_params=True"    )

													for policy in FSDP_AUTO_WRAP_POLICY:
																			if policy.lower() in config:
																									cmd_config.append(F"""--fsdp_auto_wrap_policy={policy}"""    )
																									break

													if policy == "TRANSFORMER_BASED_WRAP":
																			cmd_config.append("--fsdp_transformer_layer_cls_to_wrap=BertLayer"    )
													elif policy == "SIZE_BASED_WRAP":
																			cmd_config.append("--fsdp_min_num_params=2000"    )

													cmd_config.extend(
													    [
													        self.test_file_path,
													        F"""--output_dir={self.tmpdir}""",
													        F"""--performance_lower_bound={self.performance_lower_bound}""",
													    ]    )
													with patch_environment(omp_num_threads=1    ):
																			execute_subprocess_async(snake_case      ,	env=os.environ.copy()    )






	def       _a(self				: str    ) ->			List[Any]:
							_lowercase       :	Tuple           =   os.path.join(self.test_scripts_folder      ,	"test_checkpointing.py"    )
							_lowercase       :	Optional[int]           =   [
							    "accelerate",
							    "launch",
							    "--num_processes=2",
							    "--num_machines=1",
							    "--machine_rank=0",
							    "--use_fsdp",
							    "--mixed_precision=fp16",
							    "--fsdp_transformer_layer_cls_to_wrap=BertLayer",
							]

							for i, strategy in enumerate(snake_case    ):
													_lowercase       :	List[Any]           =   cmd.copy()
													cmd_config.append(F"""--fsdp_sharding_strategy={i+1}"""    )
													if strategy != "FULL_SHARD":
																			continue
													_lowercase       :	str           =   len(snake_case    )
													for state_dict_type in FSDP_STATE_DICT_TYPE:
																			_lowercase       :	List[str]           =   cmd_config[:state_dict_config_index]
																			cmd_config.append(F"""--fsdp_state_dict_type={state_dict_type}"""    )
																			cmd_config.extend(
																			    [
																			        self.test_file_path,
																			        F"""--output_dir={self.tmpdir}""",
																			        "--partial_train_epoch=1",
																			    ]    )
																			with patch_environment(omp_num_threads=1    ):
																									execute_subprocess_async(snake_case      ,	env=os.environ.copy()    )

																			_lowercase       :	int           =   cmd_config[:-1]
																			_lowercase       :	Optional[int]           =   os.path.join(self.tmpdir      ,	"epoch_0"    )
																			cmd_config.extend(
																			    [
																			        F"""--resume_from_checkpoint={resume_from_checkpoint}""",
																			    ]    )
																			with patch_environment(omp_num_threads=1    ):
																									execute_subprocess_async(snake_case      ,	env=os.environ.copy()    )






	def       _a(self				: Dict    ) ->			Any:
							_lowercase       :	Tuple           =   os.path.join(self.test_scripts_folder      ,	"test_peak_memory_usage.py"    )
							_lowercase       :	str           =   [
							    "accelerate",
							    "launch",
							    "--num_processes=2",
							    "--num_machines=1",
							    "--machine_rank=0",
							]
							for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
													_lowercase       :	Optional[Any]           =   cmd.copy()
													if "fp16" in spec:
																			cmd_config.extend(["--mixed_precision=fp16"]    )
													else:
																			cmd_config.extend(["--mixed_precision=no"]    )

													if "multi_gpu" in spec:
																			continue
													else:
																			cmd_config.extend(["--use_fsdp"]    )
																			for i, strategy in enumerate(snake_case    ):
																									if strategy.lower() in spec:
																															cmd_config.append(F"""--fsdp_sharding_strategy={i+1}"""    )
																															break

																			if "cpu_offload" in spec:
																									cmd_config.append("--fsdp_offload_params=True"    )

																			for policy in FSDP_AUTO_WRAP_POLICY:
																									if policy.lower() in spec:
																															cmd_config.append(F"""--fsdp_auto_wrap_policy={policy}"""    )
																															break

																			if policy == "TRANSFORMER_BASED_WRAP":
																									cmd_config.append("--fsdp_transformer_layer_cls_to_wrap=BertLayer"    )
																			elif policy == "SIZE_BASED_WRAP":
																									cmd_config.append("--fsdp_min_num_params=2000"    )

													cmd_config.extend(
													    [
													        self.test_file_path,
													        F"""--output_dir={self.tmpdir}""",
													        F"""--peak_memory_upper_bound={peak_mem_upper_bound}""",
													        F"""--n_train={self.n_train}""",
													        F"""--n_val={self.n_val}""",
													    ]    )
													with patch_environment(omp_num_threads=1    ):
																			execute_subprocess_async(snake_case      ,	env=os.environ.copy()    )






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

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


UpperCamelCase_						:     Any	       =							logging.get_logger(__name__)




class   __lowercase						(						__snake_case       ):
	_A       =   ["pixel_values"]






	def __init__(self				: Dict      ,	snake_case				: bool = True      ,	snake_case				: Dict[str, int] = None      ,	snake_case				: PILImageResampling = PILImageResampling.BICUBIC      ,	snake_case				: bool = True      ,	snake_case				: Dict[str, int] = None      ,	snake_case				: bool = True      ,	snake_case				: Union[int, float] = 1 / 255      ,	snake_case				: bool = True      ,	snake_case				: Optional[Union[float, Iterable[float]]] = IMAGENET_DEFAULT_MEAN      ,	snake_case				: Optional[Union[float, Iterable[float]]] = IMAGENET_DEFAULT_STD      ,	**snake_case				: Optional[int]      ,	) ->			None:
							super().__init__(**snake_case    )
							_lowercase       :	str           =   size if size is not None else {"shortest_edge": 224}
							_lowercase       :	List[str]           =   get_size_dict(snake_case      ,	default_to_square=snake_case    )
							_lowercase       :	Optional[int]           =   crop_size if crop_size is not None else {"height": 224, "width": 224}
							_lowercase       :	List[Any]           =   get_size_dict(snake_case      ,	param_name="crop_size"    )

							_lowercase       :	Optional[int]           =   do_resize
							_lowercase       :	List[str]           =   size
							_lowercase       :	Tuple           =   resample
							_lowercase       :	Union[str, Any]           =   do_center_crop
							_lowercase       :	List[str]           =   crop_size
							_lowercase       :	Tuple           =   do_rescale
							_lowercase       :	Union[str, Any]           =   rescale_factor
							_lowercase       :	str           =   do_normalize
							_lowercase       :	List[Any]           =   image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
							_lowercase       :	List[Any]           =   image_std if image_std is not None else IMAGENET_DEFAULT_STD






	def       _a(self				: str      ,	snake_case				: np.ndarray      ,	snake_case				: Dict[str, int]      ,	snake_case				: PILImageResampling = PILImageResampling.BICUBIC      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: str      ,	) ->			np.ndarray:
							_lowercase       :	Tuple           =   get_size_dict(snake_case      ,	default_to_square=snake_case    )
							# size_dict is a dict with either keys "height" and "width" or "shortest_edge"
							if "shortest_edge" in size:
													_lowercase       :	List[str]           =   int((256 / 224) * size["shortest_edge"]    )
													_lowercase       :	Dict           =   get_resize_output_image_size(snake_case      ,	size=snake_case      ,	default_to_square=snake_case    )
													_lowercase       :	int           =   {"height": output_size[0], "width": output_size[1]}
							if "height" not in size_dict or "width" not in size_dict:
													raise ValueError(
													    F"""Size dict must have keys 'height' and 'width' or 'shortest_edge'. Got {size_dict.keys()}"""    )
							return resize(
							    snake_case      ,	size=(size_dict["height"], size_dict["width"])      ,	resample=snake_case      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: Any      ,	snake_case				: np.ndarray      ,	snake_case				: Dict[str, int]      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: Any      ,	) ->			np.ndarray:
							_lowercase       :	Tuple           =   get_size_dict(snake_case    )
							if "height" not in size or "width" not in size:
													raise ValueError(F"""Size dict must have keys 'height' and 'width'. Got {size.keys()}"""    )
							return center_crop(snake_case      ,	size=(size["height"], size["width"])      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: Dict      ,	snake_case				: np.ndarray      ,	snake_case				: Union[int, float]      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: List[str]      ,	) ->			np.ndarray:
							return rescale(snake_case      ,	scale=snake_case      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: str      ,	snake_case				: np.ndarray      ,	snake_case				: Union[float, List[float]]      ,	snake_case				: Union[float, List[float]]      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: Optional[int]      ,	) ->			np.ndarray:
							return normalize(snake_case      ,	mean=snake_case      ,	std=snake_case      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: Any      ,	snake_case				: ImageInput      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[Dict[str, int]] = None      ,	snake_case				: PILImageResampling = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[Dict[str, int]] = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[float] = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[Union[float, Iterable[float]]] = None      ,	snake_case				: Optional[Union[float, Iterable[float]]] = None      ,	snake_case				: Optional[TensorType] = None      ,	snake_case				: ChannelDimension = ChannelDimension.FIRST      ,	**snake_case				: Tuple      ,	) ->			BatchFeature:
							_lowercase       :	Optional[Any]           =   do_resize if do_resize is not None else self.do_resize
							_lowercase       :	Any           =   resample if resample is not None else self.resample
							_lowercase       :	Optional[Any]           =   do_center_crop if do_center_crop is not None else self.do_center_crop
							_lowercase       :	Union[str, Any]           =   do_rescale if do_rescale is not None else self.do_rescale
							_lowercase       :	Optional[Any]           =   rescale_factor if rescale_factor is not None else self.rescale_factor
							_lowercase       :	str           =   do_normalize if do_normalize is not None else self.do_normalize
							_lowercase       :	Dict           =   image_mean if image_mean is not None else self.image_mean
							_lowercase       :	Dict           =   image_std if image_std is not None else self.image_std

							_lowercase       :	Optional[int]           =   size if size is not None else self.size
							_lowercase       :	Dict           =   get_size_dict(snake_case      ,	default_to_square=snake_case    )
							_lowercase       :	Dict           =   crop_size if crop_size is not None else self.crop_size
							_lowercase       :	Optional[int]           =   get_size_dict(snake_case      ,	param_name="crop_size"    )

							_lowercase       :	Dict           =   make_list_of_images(snake_case    )

							if not valid_images(snake_case    ):
													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:
													raise ValueError("Size must be specified if do_resize is True."    )

							if do_center_crop and crop_size is None:
													raise ValueError("Crop size must be specified if do_center_crop 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.
							_lowercase       :	Dict           =   [to_numpy_array(snake_case    ) for image in images]

							if do_resize:
													_lowercase       :	List[Any]           =   [self.resize(snake_case      ,	snake_case      ,	snake_case    ) for image in images]

							if do_center_crop:
													_lowercase       :	Union[str, Any]           =   [self.center_crop(snake_case      ,	snake_case    ) for image in images]

							if do_rescale:
													_lowercase       :	List[str]           =   [self.rescale(snake_case      ,	snake_case    ) for image in images]

							if do_normalize:
													_lowercase       :	str           =   [self.normalize(snake_case      ,	snake_case      ,	snake_case    ) for image in images]

							_lowercase       :	Any           =   [to_channel_dimension_format(snake_case      ,	snake_case    ) for image in images]

							_lowercase       :	str           =   {"pixel_values": images}
							return BatchFeature(data=snake_case      ,	tensor_type=snake_case    )




