



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)

__lowerCAmelCase       :					List[str]      					=					'''bert-base-cased'''
__lowerCAmelCase       :					Optional[Any]      					=					'''fp16'''
__lowerCAmelCase       :					Optional[Any]      					=					'''bf16'''
__lowerCAmelCase       :					Any      					=					[FPaa, BFaa]



@require_fsdp
@require_cuda
class    _lowerCAmelCase    (				__lowerCAmelCase		):





		"""simple docstring"""






		def    UpperCAmelCase__							(					self   ) ->						List[str]:







								'''simple docstring'''

								super().setUp()

								snake_case_   : Union[str, Any]						=			dict(
								    ACCELERATE_USE_FSDP="""true"""							,    MASTER_ADDR="""localhost"""							,    MASTER_PORT="""10999"""							,    RANK="""0"""							,    LOCAL_RANK="""0"""							,    WORLD_SIZE="""1"""							,    )





		def    UpperCAmelCase__							(					self   ) ->						int:







								'''simple docstring'''

								from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

								for i, strategy in enumerate(lowerCAmelCase_   ):
														snake_case_   : Tuple						=			self.dist_env.copy()
														snake_case_   : Any						=			f'{i + 1}'
														snake_case_   : str						=			strategy
														with mockenv_context(**lowerCAmelCase_   ):
																				snake_case_   : Any						=			FullyShardedDataParallelPlugin()
																				self.assertEqual(fsdp_plugin.sharding_strategy							,    ShardingStrategy(i + 1   )   )





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







								'''simple docstring'''

								from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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





		def    UpperCAmelCase__							(					self   ) ->						Dict:







								'''simple docstring'''

								from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

								for i, state_dict_type in enumerate(lowerCAmelCase_   ):
														snake_case_   : Optional[Any]						=			self.dist_env.copy()
														snake_case_   : Union[str, Any]						=			state_dict_type
														with mockenv_context(**lowerCAmelCase_   ):
																				snake_case_   : 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    UpperCAmelCase__							(					self   ) ->						Union[str, Any]:







								'''simple docstring'''

								snake_case_   : Optional[int]						=			AutoModel.from_pretrained(lowerCAmelCase_   )
								for policy in FSDP_AUTO_WRAP_POLICY:
														snake_case_   : int						=			self.dist_env.copy()
														snake_case_   : Optional[Any]						=			policy
														if policy == "TRANSFORMER_BASED_WRAP":
																				snake_case_   : Dict						=			"""BertLayer"""
														elif policy == "SIZE_BASED_WRAP":
																				snake_case_   : Tuple						=			"""2000"""
														with mockenv_context(**lowerCAmelCase_   ):
																				snake_case_   : Tuple						=			FullyShardedDataParallelPlugin()
																				fsdp_plugin.set_auto_wrap_policy(lowerCAmelCase_   )
																				if policy == "NO_WRAP":
																										self.assertIsNone(fsdp_plugin.auto_wrap_policy   )
																				else:
																										self.assertIsNotNone(fsdp_plugin.auto_wrap_policy   )

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

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





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







								'''simple docstring'''

								from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
								from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

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





		def    UpperCAmelCase__							(					self   ) ->						Dict:







								'''simple docstring'''

								from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

								for flag in [True, False]:
														snake_case_   : Tuple						=			self.dist_env.copy()
														snake_case_   : List[str]						=			str(lowerCAmelCase_   ).lower()
														with mockenv_context(**lowerCAmelCase_   ):
																				snake_case_   : Optional[Any]						=			FullyShardedDataParallelPlugin()
																				self.assertEqual(fsdp_plugin.cpu_offload							,    CPUOffload(offload_params=lowerCAmelCase_   )   )




@require_fsdp
@require_multi_gpu
@slow
class    _lowerCAmelCase    (				__lowerCAmelCase		):





		"""simple docstring"""






		def    UpperCAmelCase__							(					self   ) ->						List[Any]:







								'''simple docstring'''

								super().setUp()
								snake_case_   : Any						=			0.82
								snake_case_   : List[Any]						=			[
								    """fsdp_shard_grad_op_transformer_based_wrap""",
								    """fsdp_full_shard_transformer_based_wrap""",
								]
								snake_case_   : List[Any]						=			{
								    """multi_gpu_fp16""": 3_2_0_0,
								    """fsdp_shard_grad_op_transformer_based_wrap_fp16""": 2_0_0_0,
								    """fsdp_full_shard_transformer_based_wrap_fp16""": 1_9_0_0,
								    # 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
								}
								snake_case_   : int						=			1_6_0
								snake_case_   : int						=			1_6_0

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





		def    UpperCAmelCase__							(					self   ) ->						List[str]:







								'''simple docstring'''

								snake_case_   : List[Any]						=			os.path.join(self.test_scripts_folder							,    """test_performance.py"""   )
								snake_case_   : List[str]						=			["""accelerate""", """launch""", """--num_processes=2""", """--num_machines=1""", """--machine_rank=0""", """--use_fsdp"""]
								for config in self.performance_configs:
														snake_case_   : Optional[int]						=			cmd.copy()
														for i, strategy in enumerate(lowerCAmelCase_   ):
																				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(lowerCAmelCase_							,    env=os.environ.copy()   )





		def    UpperCAmelCase__							(					self   ) ->						List[str]:







								'''simple docstring'''

								snake_case_   : Dict						=			os.path.join(self.test_scripts_folder							,    """test_checkpointing.py"""   )
								snake_case_   : 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(lowerCAmelCase_   ):
														snake_case_   : List[Any]						=			cmd.copy()
														cmd_config.append(f'--fsdp_sharding_strategy={i+1}'   )
														if strategy != "FULL_SHARD":
																				continue
														snake_case_   : int						=			len(lowerCAmelCase_   )
														for state_dict_type in FSDP_STATE_DICT_TYPE:
																				snake_case_   : Any						=			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(lowerCAmelCase_							,    env=os.environ.copy()   )

																				snake_case_   : List[str]						=			cmd_config[:-1]
																				snake_case_   : Any						=			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(lowerCAmelCase_							,    env=os.environ.copy()   )





		def    UpperCAmelCase__							(					self   ) ->						Any:







								'''simple docstring'''

								snake_case_   : Optional[int]						=			os.path.join(self.test_scripts_folder							,    """test_peak_memory_usage.py"""   )
								snake_case_   : List[Any]						=			[
								    """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():
														snake_case_   : Tuple						=			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(lowerCAmelCase_   ):
																										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(lowerCAmelCase_							,    env=os.environ.copy()   )






"""simple docstring"""





import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


__lowerCAmelCase       :					Optional[int]      					=					logging.get_logger(__name__)
def 	__lowerCAmelCase						(		__UpperCamelCase      :			Tuple					, __UpperCamelCase      :			Optional[Any]    ):


						'''simple docstring'''






						snake_case_   : Union[str, Any]						=			nn.functional.normalize(__UpperCamelCase    )
						snake_case_   : Tuple						=			nn.functional.normalize(__UpperCamelCase    )
						return torch.mm(__UpperCamelCase					, normalized_text_embeds.t()    )






class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       CLIPConfig

		_lowerCamelCase			  =       ['''CLIPEncoderLayer''']





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







								'''simple docstring'''

								super().__init__(_lowercase   )

								snake_case_   : Tuple						=			CLIPVisionModel(config.vision_config   )
								snake_case_   : int						=			nn.Linear(config.vision_config.hidden_size							,    config.projection_dim							,    bias=_lowercase   )

								snake_case_   : Optional[Any]						=			nn.Parameter(torch.ones(1_7							,    config.projection_dim   )							,    requires_grad=_lowercase   )
								snake_case_   : Dict						=			nn.Parameter(torch.ones(3							,    config.projection_dim   )							,    requires_grad=_lowercase   )

								snake_case_   : Any						=			nn.Parameter(torch.ones(1_7   )							,    requires_grad=_lowercase   )
								snake_case_   : List[str]						=			nn.Parameter(torch.ones(3   )							,    requires_grad=_lowercase   )





		@torch.no_grad()
		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase   ) ->						Any:







								'''simple docstring'''

								snake_case_   : int						=			self.vision_model(_lowercase   )[1]  # pooled_output
								snake_case_   : str						=			self.visual_projection(_lowercase   )

								# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
								snake_case_   : Dict						=			cosine_distance(_lowercase							,    self.special_care_embeds   ).cpu().float().numpy()
								snake_case_   : List[str]						=			cosine_distance(_lowercase							,    self.concept_embeds   ).cpu().float().numpy()

								snake_case_   : Any						=			[]
								snake_case_   : Any						=			image_embeds.shape[0]
								for i in range(_lowercase   ):
														snake_case_   : List[Any]						=			{"""special_scores""": {}, """special_care""": [], """concept_scores""": {}, """bad_concepts""": []}

														# increase this value to create a stronger `nfsw` filter
														# at the cost of increasing the possibility of filtering benign images
														snake_case_   : int						=			0.0

														for concept_idx in range(len(special_cos_dist[0]   )   ):
																				snake_case_   : List[str]						=			special_cos_dist[i][concept_idx]
																				snake_case_   : Union[str, Any]						=			self.special_care_embeds_weights[concept_idx].item()
																				snake_case_   : Tuple						=			round(concept_cos - concept_threshold + adjustment							,    3   )
																				if result_img["special_scores"][concept_idx] > 0:
																										result_img["special_care"].append({concept_idx, result_img["""special_scores"""][concept_idx]}   )
																										snake_case_   : Dict						=			0.01

														for concept_idx in range(len(cos_dist[0]   )   ):
																				snake_case_   : int						=			cos_dist[i][concept_idx]
																				snake_case_   : List[Any]						=			self.concept_embeds_weights[concept_idx].item()
																				snake_case_   : List[str]						=			round(concept_cos - concept_threshold + adjustment							,    3   )
																				if result_img["concept_scores"][concept_idx] > 0:
																										result_img["bad_concepts"].append(_lowercase   )

														result.append(_lowercase   )

								snake_case_   : Union[str, Any]						=			[len(res["""bad_concepts"""]   ) > 0 for res in result]

								return images, has_nsfw_concepts





		@torch.no_grad()
		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase   ) ->						Tuple:







								'''simple docstring'''

								snake_case_   : Optional[Any]						=			self.vision_model(_lowercase   )[1]  # pooled_output
								snake_case_   : List[str]						=			self.visual_projection(_lowercase   )

								snake_case_   : str						=			cosine_distance(_lowercase							,    self.special_care_embeds   )
								snake_case_   : Optional[int]						=			cosine_distance(_lowercase							,    self.concept_embeds   )

								# increase this value to create a stronger `nsfw` filter
								# at the cost of increasing the possibility of filtering benign images
								snake_case_   : Tuple						=			0.0

								snake_case_   : List[Any]						=			special_cos_dist - self.special_care_embeds_weights + adjustment
								# special_scores = special_scores.round(decimals=3)
								snake_case_   : str						=			torch.any(special_scores > 0							,    dim=1   )
								snake_case_   : List[str]						=			special_care * 0.01
								snake_case_   : Optional[int]						=			special_adjustment.unsqueeze(1   ).expand(-1							,    cos_dist.shape[1]   )

								snake_case_   : Optional[Any]						=			(cos_dist - self.concept_embeds_weights) + special_adjustment
								# concept_scores = concept_scores.round(decimals=3)
								snake_case_   : str						=			torch.any(concept_scores > 0							,    dim=1   )

								return images, has_nsfw_concepts

