


"""simple docstring"""




from typing import Callable, List, Optional, Union

import PIL
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPSegForImageSegmentation,
    CLIPSegProcessor,
    CLIPTextModel,
    CLIPTokenizer,
)

from diffusers import DiffusionPipeline
from diffusers.configuration_utils import FrozenDict
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionInpaintPipeline
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import deprecate, is_accelerate_available, logging


lowercase_			     =  logging.get_logger(__name__)  # pylint: disable=invalid-name
class 					__lowerCAmelCase					(   __SCREAMING_SNAKE_CASE       ):


		'''simple docstring'''



		def __init__(  self							, _a							, _a							, _a							, _a							, _a							, _a							, _a							, _a							, _a							, ):
				super().__init__()

				if hasattr(scheduler.config							, '''steps_offset'''      ) and scheduler.config.steps_offset != 1:
						__a           =			(
						    f'''The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`'''
						    f''' should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure '''
						    '''to update the config accordingly as leaving `steps_offset` might led to incorrect results'''
						    ''' in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,'''
						    ''' it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`'''
						    ''' file'''
						)
						deprecate('''steps_offset!=1'''							, '''1.0.0'''							, _a							, standard_warn=_a      )
						__a           =			dict(scheduler.config      )
						__a           =			1
						__a           =			FrozenDict(_a      )

				if hasattr(scheduler.config							, '''skip_prk_steps'''      ) and scheduler.config.skip_prk_steps is False:
						__a           =			(
						    f'''The configuration file of this scheduler: {scheduler} has not set the configuration'''
						    ''' `skip_prk_steps`. `skip_prk_steps` should be set to True in the configuration file. Please make'''
						    ''' sure to update the config accordingly as not setting `skip_prk_steps` in the config might lead to'''
						    ''' incorrect results in future versions. If you have downloaded this checkpoint from the Hugging Face'''
						    ''' Hub, it would be very nice if you could open a Pull request for the'''
						    ''' `scheduler/scheduler_config.json` file'''
						)
						deprecate('''skip_prk_steps not set'''							, '''1.0.0'''							, _a							, standard_warn=_a      )
						__a           =			dict(scheduler.config      )
						__a           =			True
						__a           =			FrozenDict(_a      )

				if safety_checker is None:
						logger.warning(
						    f'''You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure'''
						    ''' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'''
						    ''' results in services or applications open to the public. Both the diffusers team and Hugging Face'''
						    ''' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'''
						    ''' it only for use-cases that involve analyzing network behavior or auditing its results. For more'''
						    ''' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .'''      )

				self.register_modules(
				    segmentation_model=_a							, segmentation_processor=_a							, vae=_a							, text_encoder=_a							, tokenizer=_a							, unet=_a							, scheduler=_a							, safety_checker=_a							, feature_extractor=_a							, )

		def 			__UpperCAmelCase       (  self							, _a = "auto"      ):
				if slice_size == "auto":
						# half the attention head size is usually a good trade-off between
						# speed and memory
						__a           =			self.unet.config.attention_head_dim // 2
				self.unet.set_attention_slice(_a      )

		def 			__UpperCAmelCase       (  self      ):
				self.enable_attention_slicing(_a      )

		def 			__UpperCAmelCase       (  self      ):
				if is_accelerate_available():
						from accelerate import cpu_offload
				else:
						raise ImportError('''Please install accelerate via `pip install accelerate`'''      )

				__a           =			torch.device('''cuda'''      )

				for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae, self.safety_checker]:
						if cpu_offloaded_model is not None:
								cpu_offload(_a							, _a      )

		@property
		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
		def 			__UpperCAmelCase       (  self      ):
				if self.device != torch.device('''meta'''      ) or not hasattr(self.unet							, '''_hf_hook'''      ):
						return self.device
				for module in self.unet.modules():
						if (
						    hasattr(_a							, '''_hf_hook'''      )
						    and hasattr(module._hf_hook							, '''execution_device'''      )
						    and module._hf_hook.execution_device is not None
						):
								return torch.device(module._hf_hook.execution_device      )
				return self.device





		@torch.no_grad()
		def __call__(  self							, _a							, _a							, _a							, _a = 512							, _a = 512							, _a = 50							, _a = 7.5							, _a = None							, _a = 1							, _a = 0.0							, _a = None							, _a = None							, _a = "pil"							, _a = True							, _a = None							, _a = 1							, **_a							, ):
				__a           =			self.segmentation_processor(
				    text=[text]							, images=[image]							, padding='''max_length'''							, return_tensors='''pt'''      ).to(self.device      )
				__a           =			self.segmentation_model(**_a      )
				__a           =			torch.sigmoid(outputs.logits      ).cpu().detach().unsqueeze(-1      ).numpy()
				__a           =			self.numpy_to_pil(_a      )[0].resize(image.size      )

				# Run inpainting pipeline with the generated mask
				__a           =			StableDiffusionInpaintPipeline(
				    vae=self.vae							, text_encoder=self.text_encoder							, tokenizer=self.tokenizer							, unet=self.unet							, scheduler=self.scheduler							, safety_checker=self.safety_checker							, feature_extractor=self.feature_extractor							, )
				return inpainting_pipeline(
				    prompt=_a							, image=_a							, mask_image=_a							, height=_a							, width=_a							, num_inference_steps=_a							, guidance_scale=_a							, negative_prompt=_a							, num_images_per_prompt=_a							, eta=_a							, generator=_a							, latents=_a							, output_type=_a							, return_dict=_a							, callback=_a							, callback_steps=_a							, )










"""simple docstring"""




from __future__ import annotations

import unittest

from transformers import EsmConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
				import numpy
				import tensorflow as tf

				from transformers.models.esm.modeling_tf_esm import (
				    TF_ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
				    TFEsmForMaskedLM,
				    TFEsmForSequenceClassification,
				    TFEsmForTokenClassification,
				    TFEsmModel,
				)
class 					__lowerCAmelCase					:


		'''simple docstring'''



		def __init__(  self							, _a							, ):
				__a           =			parent
				__a           =			13
				__a           =			7
				__a           =			True
				__a           =			True
				__a           =			True
				__a           =			99
				__a           =			32
				__a           =			2
				__a           =			4
				__a           =			37
				__a           =			'''gelu'''
				__a           =			0.1
				__a           =			0.1
				__a           =			512
				__a           =			16
				__a           =			2
				__a           =			0.02
				__a           =			3
				__a           =			4
				__a           =			None

		def 			__UpperCAmelCase       (  self      ):
				__a           =			ids_tensor([self.batch_size, self.seq_length]							, self.vocab_size      )

				__a           =			None
				if self.use_input_mask:
						__a           =			random_attention_mask([self.batch_size, self.seq_length]      )

				__a           =			None
				__a           =			None
				__a           =			None
				if self.use_labels:
						__a           =			ids_tensor([self.batch_size]							, self.type_sequence_label_size      )
						__a           =			ids_tensor([self.batch_size, self.seq_length]							, self.num_labels      )
						__a           =			ids_tensor([self.batch_size]							, self.num_choices      )

				__a           =			EsmConfig(
				    vocab_size=self.vocab_size							, hidden_size=self.hidden_size							, num_hidden_layers=self.num_hidden_layers							, pad_token_id=1							, num_attention_heads=self.num_attention_heads							, intermediate_size=self.intermediate_size							, hidden_act=self.hidden_act							, hidden_dropout_prob=self.hidden_dropout_prob							, attention_probs_dropout_prob=self.attention_probs_dropout_prob							, max_position_embeddings=self.max_position_embeddings							, type_vocab_size=self.type_vocab_size							, initializer_range=self.initializer_range							, )

				return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels

		def 			__UpperCAmelCase       (  self      ):
				(
				    (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     
				)           =			self.prepare_config_and_inputs()

				__a           =			True
				__a           =			floats_tensor([self.batch_size, self.seq_length, self.hidden_size]      )
				__a           =			ids_tensor([self.batch_size, self.seq_length]							, vocab_size=2      )

				return (
				    config,
				    input_ids,
				    input_mask,
				    sequence_labels,
				    token_labels,
				    choice_labels,
				    encoder_hidden_states,
				    encoder_attention_mask,
				)

		def 			__UpperCAmelCase       (  self							, _a							, _a							, _a							, _a							, _a							, _a      ):
				__a           =			TFEsmModel(config=_a      )
				__a           =			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
				__a           =			model(_a      )

				__a           =			[input_ids, input_mask]
				__a           =			model(_a      )

				__a           =			model(_a      )

				self.parent.assertEqual(result.last_hidden_state.shape							, (self.batch_size, self.seq_length, self.hidden_size)      )

		def 			__UpperCAmelCase       (  self							, _a							, _a							, _a							, _a							, _a							, _a							, _a							, _a							, ):
				__a           =			True

				__a           =			TFEsmModel(config=_a      )
				__a           =			{
				    '''input_ids''': input_ids,
				    '''attention_mask''': input_mask,
				    '''encoder_hidden_states''': encoder_hidden_states,
				    '''encoder_attention_mask''': encoder_attention_mask,
				}
				__a           =			model(_a      )

				__a           =			[input_ids, input_mask]
				__a           =			model(_a							, encoder_hidden_states=_a      )

				# Also check the case where encoder outputs are not passed
				__a           =			model(_a							, attention_mask=_a      )

				self.parent.assertEqual(result.last_hidden_state.shape							, (self.batch_size, self.seq_length, self.hidden_size)      )

		def 			__UpperCAmelCase       (  self							, _a							, _a							, _a							, _a							, _a							, _a      ):
				__a           =			TFEsmForMaskedLM(config=_a      )
				__a           =			model([input_ids, input_mask]      )
				self.parent.assertEqual(result.logits.shape							, (self.batch_size, self.seq_length, self.vocab_size)      )

		def 			__UpperCAmelCase       (  self							, _a							, _a							, _a							, _a							, _a							, _a      ):
				__a           =			self.num_labels
				__a           =			TFEsmForTokenClassification(config=_a      )
				__a           =			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
				__a           =			model(_a      )
				self.parent.assertEqual(result.logits.shape							, (self.batch_size, self.seq_length, self.num_labels)      )





		def 			__UpperCAmelCase       (  self      ):
				__a           =			self.prepare_config_and_inputs()
				(
				    (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     (
				    __a
				)      ,     
				)           =			config_and_inputs
				__a           =			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
				return config, inputs_dict







@require_tf
class 					__lowerCAmelCase					(   __SCREAMING_SNAKE_CASE     ,			__SCREAMING_SNAKE_CASE     ,			unittest.TestCase       ):


		'''simple docstring'''



		__UpperCAmelCase		:			int										=				(
		    (
		        TFEsmModel,
		        TFEsmForMaskedLM,
		        TFEsmForSequenceClassification,
		        TFEsmForTokenClassification,
		    )
		    if is_tf_available()
		    else ()
		)
		__UpperCAmelCase		:			Tuple										=				(
		    {
		        'feature-extraction': TFEsmModel,
		        'fill-mask': TFEsmForMaskedLM,
		        'text-classification': TFEsmForSequenceClassification,
		        'token-classification': TFEsmForTokenClassification,
		        'zero-shot': TFEsmForSequenceClassification,
		    }
		    if is_tf_available()
		    else {}
		)
		__UpperCAmelCase		:			Tuple										=				False
		__UpperCAmelCase		:			Union[str, Any]										=				False

		def 			__UpperCAmelCase       (  self      ):
				__a           =			TFEsmModelTester(self      )
				__a           =			ConfigTester(self							, config_class=_a							, hidden_size=37      )

		def 			__UpperCAmelCase       (  self      ):
				self.config_tester.run_common_tests()

		def 			__UpperCAmelCase       (  self      ):
				__a           =			self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_model(*_a      )

		def 			__UpperCAmelCase       (  self      ):
				__a           =			self.model_tester.prepare_config_and_inputs_for_decoder()
				self.model_tester.create_and_check_model_as_decoder(*_a      )

		def 			__UpperCAmelCase       (  self      ):
				__a           =			self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_for_masked_lm(*_a      )

		def 			__UpperCAmelCase       (  self      ):
				__a           =			self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_for_token_classification(*_a      )

		@slow
		def 			__UpperCAmelCase       (  self      ):
				for model_name in TF_ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
						__a           =			TFEsmModel.from_pretrained(_a      )
						self.assertIsNotNone(_a      )

		@unittest.skip('''Protein models do not support embedding resizing.'''      )
		def 			__UpperCAmelCase       (  self      ):
				pass

		@unittest.skip('''Protein models do not support embedding resizing.'''      )
		def 			__UpperCAmelCase       (  self      ):
				pass





		def 			__UpperCAmelCase       (  self      ):
				__a      ,     __a           =			self.model_tester.prepare_config_and_inputs_for_common()

				for model_class in self.all_model_classes:
						__a           =			model_class(_a      )
						assert isinstance(model.get_input_embeddings()							, tf.keras.layers.Layer      )
						if model_class is TFEsmForMaskedLM:
								# Output embedding test differs from the main test because they're a matrix, not a layer
								__a           =			model.get_bias()
								assert isinstance(_a							, _a      )
								for k, v in name.items():
										assert isinstance(_a							, tf.Variable      )
						else:
								__a           =			model.get_output_embeddings()
								assert x is None
								__a           =			model.get_bias()
								assert name is None







@require_tf
class 					__lowerCAmelCase					(   unittest.TestCase       ):


		'''simple docstring'''



		@slow
		def 			__UpperCAmelCase       (  self      ):
				__a           =			TFEsmForMaskedLM.from_pretrained('''facebook/esm2_t6_8M_UR50D'''      )

				__a           =			tf.constant([[0, 1, 2, 3, 4, 5]]      )
				__a           =			model(_a      )[0]
				__a           =			[1, 6, 33]
				self.assertEqual(list(output.numpy().shape      )							, _a      )
				# compare the actual values for a slice.
				__a           =			tf.constant(
				    [
				        [
				            [8.92_1518, -10.58_9814, -6.467_1307],
				            [-6.396_7156, -13.91_1377, -1.121_1915],
				            [-7.78_1247, -13.95_1557, -3.74_0592],
				        ]
				    ]      )
				self.assertTrue(numpy.allclose(output[:, :3, :3].numpy()							, expected_slice.numpy()							, atol=1E-2      )      )





		@slow
		def 			__UpperCAmelCase       (  self      ):
				__a           =			TFEsmModel.from_pretrained('''facebook/esm2_t6_8M_UR50D'''      )

				__a           =			tf.constant([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]]      )
				__a           =			model(_a      )[0]
				# compare the actual values for a slice.
				__a           =			tf.constant(
				    [
				        [
				            [0.1444_3092, 0.5412_5327, 0.324_7739],
				            [0.3034_0484, 0.0052_6676, 0.3107_7722],
				            [0.3227_8043, -0.2498_7096, 0.341_4628],
				        ]
				    ]      )
				self.assertTrue(numpy.allclose(output[:, :3, :3].numpy()							, expected_slice.numpy()							, atol=1E-4      )      )






